Merge branch '7.9.x' into master by armittal-0077

Author: ConfluentSemaphore

ksqldb-engine/src/main/java/io/confluent/ksql/services/KafkaTopicClientImpl.java

@@ -205,14 +205,22 @@ public Set<String> listTopicNames() {
   }
 
   @Override
-  public Map<String, TopicDescription> describeTopics(final Collection<String> topicNames) {
+  public Map<String, TopicDescription> describeTopics(final Collection<String> topicNames,
+                                                      final Boolean skipRetriesOnFailure) {
     try {
-      return ExecutorUtil.executeWithRetries(
-          () -> adminClient.get().describeTopics(
-              topicNames,
-              new DescribeTopicsOptions().includeAuthorizedOperations(true)
-          ).allTopicNames().get(),
-          ExecutorUtil.RetryBehaviour.ON_RETRYABLE);
+      if (skipRetriesOnFailure) {
+        return adminClient.get().describeTopics(
+                topicNames,
+                new DescribeTopicsOptions().includeAuthorizedOperations(true)
+        ).allTopicNames().get();
+      } else {
+        return ExecutorUtil.executeWithRetries(
+                () -> adminClient.get().describeTopics(
+                        topicNames,
+                        new DescribeTopicsOptions().includeAuthorizedOperations(true)
+                ).allTopicNames().get(),
+                ExecutorUtil.RetryBehaviour.ON_RETRYABLE);
+      }
     } catch (final ExecutionException e) {
       throw new KafkaResponseGetFailedException(
           "Failed to Describe Kafka Topic(s): " + topicNames, e.getCause());

ksqldb-engine/src/main/java/io/confluent/ksql/services/SandboxedKafkaTopicClient.java

@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -63,7 +64,9 @@ static KafkaTopicClient createProxy(final KafkaTopicClient delegate,
         .forward("isTopicExists", methodParams(String.class), sandbox)
         .forward("describeTopic", methodParams(String.class), sandbox)
         .forward("getTopicConfig", methodParams(String.class), sandbox)
+        .forward("describeTopic", methodParams(String.class, Boolean.class), sandbox)
         .forward("describeTopics", methodParams(Collection.class), sandbox)
+        .forward("describeTopics", methodParams(Collection.class, Boolean.class), sandbox)
         .forward("deleteTopics", methodParams(Collection.class), sandbox)
         .forward("listTopicsStartOffsets", methodParams(Collection.class), sandbox)
         .forward("listTopicsEndOffsets", methodParams(Collection.class), sandbox)
@@ -159,21 +162,32 @@ public TopicDescription describeTopic(final String topicName) {
     return describeTopics(ImmutableList.of(topicName)).get(topicName);
   }
 
+  public TopicDescription describeTopic(final String topicName,
+                                        final Boolean skipRetriesOnFailure) {
+    return describeTopics(ImmutableList.of(topicName), skipRetriesOnFailure).get(topicName);
+  }
+
   private Map<String, TopicDescription> describeTopics(final Collection<String> topicNames) {
+    return describeTopics(topicNames, false);
+  }
+
+  private Map<String, TopicDescription> describeTopics(final Collection<String> topicNames,
+                                                       final Boolean skipRetriesOnFailure) {
     final Map<String, TopicDescription> descriptions = topicNames.stream()
         .map(createdTopics::get)
         .filter(Objects::nonNull)
         .collect(Collectors.toMap(TopicDescription::name, Function.identity()));
 
-    final HashSet<String> remaining = new HashSet<>(topicNames);
-    remaining.removeAll(descriptions.keySet());
-    if (remaining.isEmpty()) {
+    final Set<String> topicsToFetch = new HashSet<>(topicNames);
+    topicsToFetch.removeAll(descriptions.keySet());
+    if (topicsToFetch.isEmpty()) {
       return descriptions;
     }
 
-    final Map<String, TopicDescription> fromKafka = delegate.describeTopics(remaining);
+    final Map<String, TopicDescription> remainingTopicDescriptionMap =
+                delegate.describeTopics(topicsToFetch, skipRetriesOnFailure);
 
-    descriptions.putAll(fromKafka);
+    descriptions.putAll(remainingTopicDescriptionMap);
     return descriptions;
   }
 

ksqldb-engine/src/test/java/io/confluent/ksql/services/FakeKafkaTopicClient.java

@@ -167,7 +167,8 @@ public Set<String> listTopicNames() {
   }
 
   @Override
-  public Map<String, TopicDescription> describeTopics(final Collection<String> topicNames) {
+  public Map<String, TopicDescription> describeTopics(Collection<String> topicNames,
+                                                      Boolean skipRetriesOnFailure) {
     return topicNames.stream()
         .collect(Collectors.toMap(Function.identity(), this::describeTopic));
   }

ksqldb-engine/src/test/java/io/confluent/ksql/services/KafkaTopicClientImplTest.java

@@ -923,6 +923,23 @@ public void shouldNotRetryIsTopicExistsOnUnknownTopicException() {
     verify(adminClient, times(1)).describeTopics(anyCollection(), any());
   }
 
+  @Test
+  public void shouldNotRetryDescribeTopicsExistsAnyException() {
+    // When
+    when(adminClient.describeTopics(anyCollection(), any()))
+            .thenAnswer(describeTopicsResult(new UnknownTopicOrPartitionException("meh")));
+    final String topicName = "foobar";
+    final Exception e = assertThrows(
+            KafkaResponseGetFailedException.class,
+            () -> kafkaTopicClient.describeTopic(topicName, true)
+    );
+
+    // Then
+    verify(adminClient, times(1)).describeTopics(anyCollection(), any());
+    assertThat(e.getMessage(),
+            containsString("Failed to Describe Kafka Topic(s): [" + topicName + "]"));
+  }
+
   @Test
   public void shouldThrowIsTopicExistsOnAuthorizationException() {
     // Given

ksqldb-engine/src/test/java/io/confluent/ksql/services/SandboxedKafkaTopicClientTest.java

@@ -90,7 +90,9 @@ public static Collection<TestCase<KafkaTopicClient>> getMethodsToTest() {
           .ignore("isTopicExists", String.class)
           .ignore("describeTopic", String.class)
           .ignore("getTopicConfig", String.class)
+          .ignore("describeTopic", String.class, Boolean.class)
           .ignore("describeTopics", Collection.class)
+          .ignore("describeTopics", Collection.class, Boolean.class)
           .ignore("deleteTopics", Collection.class)
           .ignore("listTopicsStartOffsets", Collection.class)
           .ignore("listTopicsEndOffsets", Collection.class)
@@ -461,7 +463,7 @@ private void givenTopicExists(
         final int numReplicas
     ) {
       when(delegate.isTopicExists(topic)).thenReturn(true);
-      when(delegate.describeTopics(Collections.singleton(topic)))
+      when(delegate.describeTopics(Collections.singleton(topic), false))
           .thenReturn(Collections.singletonMap(
               topic,
               new TopicDescription(topic, false, topicPartitions(numPartitions, numReplicas))));

ksqldb-execution/src/main/java/io/confluent/ksql/services/KafkaTopicClient.java

@@ -154,7 +154,20 @@ boolean createTopic(
    * @return map of topic name to description.
    * @throws KafkaTopicExistsException if any of the topic do not exist.
    */
-  Map<String, TopicDescription> describeTopics(Collection<String> topicNames);
+  default Map<String, TopicDescription> describeTopics(Collection<String> topicNames) {
+    return describeTopics(topicNames, false);
+  }
+
+  /**
+   * Call to get one or more topic's description, considering if we want to skip retries on failure.
+   *
+   * @param topicNames topicNames to describe
+   * @param skipRetriesOnFailure  whether to skip retries on failure
+   * @return  map of topic name to description.
+   * @throws KafkaTopicExistsException if any of the topic do not exist.
+   */
+  Map<String, TopicDescription> describeTopics(Collection<String> topicNames,
+                                               Boolean skipRetriesOnFailure);
 
   /**
    * Call to get a one topic's description.
@@ -167,6 +180,18 @@ default TopicDescription describeTopic(final String topicName) {
     return describeTopics(ImmutableList.of(topicName)).get(topicName);
   }
 
+  /**
+   * Call to get one topic's description, considering if we want to skip retries on failure.
+   *
+   * @param topicName topicName to describe
+   * @param skipRetriesOnFailure whether to skip retries on failure
+   * @return  the description if the topic
+   * @throws KafkaTopicExistsException if the topic does not exist.
+   */
+  default TopicDescription describeTopic(final String topicName, Boolean skipRetriesOnFailure) {
+    return describeTopics(ImmutableList.of(topicName), skipRetriesOnFailure).get(topicName);
+  }
+
   /**
    * Call to get the config of a topic.
    *

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/ListSourceExecutor.java

@@ -295,7 +295,7 @@ private static SourceDescriptionWithWarnings describeSource(
     final List<KsqlWarning> warnings = new LinkedList<>();
     try {
       topicDescription = Optional.of(
-          serviceContext.getTopicClient().describeTopic(dataSource.getKafkaTopicName())
+          serviceContext.getTopicClient().describeTopic(dataSource.getKafkaTopicName(), true)
       );
       sourceConstraints = getSourceConstraints(name, ksqlExecutionContext.getMetaStore());
     } catch (final KafkaException | KafkaResponseGetFailedException e) {

ksqldb-testing-tool/src/main/java/io/confluent/ksql/tools/test/stubs/StubKafkaTopicClient.java

@@ -141,7 +141,8 @@ public Set<String> listTopicNames() {
   }
 
   @Override
-  public Map<String, TopicDescription> describeTopics(final Collection<String> topicNames) {
+  public Map<String, TopicDescription> describeTopics(final Collection<String> topicNames,
+                                                      final Boolean skipRetriesOnFailure) {
     return topicNames.stream()
         .collect(Collectors.toMap(Function.identity(), this::describeTopic));
   }