Add retry to query retention time for destination topic (#863)

* Add retry to query retention time for destination topic

Since migrating to AdminClient to create destination topic it was observed that sometimes there is a
lag in topic metadata to be synced to all brokers. This may result in getting incorrect (null)
retention time. Adding a retry logic to account for delay in topic metadata sync.

* Adding some extra log messages

Author: Suraj Nagaraja Kasi

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProviderAdmin.java

@@ -39,6 +39,7 @@
 import com.linkedin.datastream.common.DatastreamRuntimeException;
 import com.linkedin.datastream.common.DatastreamSource;
 import com.linkedin.datastream.common.DatastreamUtils;
+import com.linkedin.datastream.common.PollUtils;
 import com.linkedin.datastream.common.VerifiableProperties;
 import com.linkedin.datastream.metrics.BrooklinMetricInfo;
 import com.linkedin.datastream.server.DatastreamTask;
@@ -71,6 +72,8 @@ public class KafkaTransportProviderAdmin implements TransportProviderAdmin {
   private static final int DEFAULT_NUMBER_PARTITIONS = 1;
   private static final String DEFAULT_MIN_INSYNC_REPLICAS_CONFIG_VALUE = "2";
   private static final String METADATA_KAFKA_BROKERS = DatastreamMetadataConstants.SYSTEM_DESTINATION_PREFIX + "KafkaBrokers";
+  private static final int GET_RETENTION_RETRY_PERIOD_MS = 2000;
+  private static final int GET_RETENTION_RETRY_TIMEOUT_MS = 10000;
 
   private final String _transportProviderMetricsNamesPrefix;
   private final int _numProducersPerConnector;
@@ -250,23 +253,26 @@ public Duration getRetention(Datastream datastream) {
     Validate.notNull(destination, "null destination URI");
     String topicName = KafkaTransportProviderUtils.getTopicName(destination);
 
-    try {
-      // TODO: In rare cases it may happen that even though topic has been created, its metadata hasn't synced
-      // to all the brokers yet and adminClient might query such a broker in which case topic retention will
-      // not be present in the topic config. To circumvent this we need to query only the controller node in
-      // Kafka cluster which will always return successful result, but that is more involved.
-      ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
-      Map<ConfigResource, Config> topicConfigMap = adminClient.describeConfigs(Collections.singletonList(topicResource)).all().get();
-      Config topicConfig = topicConfigMap.get(topicResource);
-      ConfigEntry entry = topicConfig.get(TOPIC_RETENTION_MS);
-      if (entry != null) {
-        return Duration.ofMillis(Long.parseLong(entry.value()));
+    // In rare cases it may happen that even though topic has been created, its metadata hasn't synced
+    // to all the brokers yet and adminClient might query such a broker in which case topic retention will
+    // not be present in the topic config. Therefore we retry for a few times before giving up.
+    return PollUtils.poll(() -> {
+      try {
+        ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
+        Map<ConfigResource, Config> topicConfigMap =
+            adminClient.describeConfigs(Collections.singletonList(topicResource)).all().get();
+        Config topicConfig = topicConfigMap.get(topicResource);
+        ConfigEntry entry = topicConfig.get(TOPIC_RETENTION_MS);
+        if (entry != null) {
+          LOG.info("Retention time for topic {} is {}", topicName, entry.value());
+          return Duration.ofMillis(Long.parseLong(entry.value()));
+        }
+      } catch (ExecutionException e) {
+        LOG.warn("Failed to retrieve config for topic {}.", topicName, e);
       }
-    } catch (InterruptedException | ExecutionException e) {
-      LOG.warn("Failed to retrieve config for topic {}.", topicName, e);
-    }
-
-    return null;
+      LOG.warn("Failed to retrieve retention time for topic {}", topicName);
+      return null;
+    }, Objects::nonNull, GET_RETENTION_RETRY_PERIOD_MS, GET_RETENTION_RETRY_TIMEOUT_MS).orElse(null);
   }
 
   /**

datastream-server-restli/src/test/java/com/linkedin/datastream/server/TestCoordinator.java

@@ -2334,13 +2334,6 @@ private TestSetup createTestCoordinator() throws Exception {
   }
 
   private void validateRetention(Datastream stream, DatastreamResources resource, Duration expectedRetention) {
-    // Adding a sleep to simulate time required for topic metadata to be synced across all brokers of destination
-    // kafka cluster.
-    try {
-      Thread.sleep(5000);
-    } catch (Exception e) {
-    }
-
     Datastream queryStream = resource.get(stream.getName());
     Assert.assertNotNull(queryStream.getDestination());
     StringMap metadata = queryStream.getMetadata();