Increase Kafka topic polling time in test to 10 seconds (#18040) (#18042)

We've seen multiple failures where the 3 second timeout was breached.
This happens in the test setup of many different tests so multiple
individual test failures have been reported.

Also replace deprecated method in Kafka API and simplify a couple other
code paths.


(cherry picked from commit 54e02a7)

Signed-off-by: Andrew Ross <andrross@amazon.com>
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: opensearch-trigger-bot[bot]

plugins/ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaUtils.java

@@ -53,33 +53,29 @@ public static void createTopic(String topicName, int numOfPartitions, String boo
         }
 
         // validates topic is created
-        await().atMost(3, TimeUnit.SECONDS).until(() -> checkTopicExistence(topicName, bootstrapServers));
+        await().atMost(10, TimeUnit.SECONDS).until(() -> checkTopicExistence(topicName, bootstrapServers));
     }
 
     public static boolean checkTopicExistence(String topicName, String bootstrapServers) {
         return getAdminClient(bootstrapServers, (client -> {
-            Map<String, KafkaFuture<TopicDescription>> topics = client.describeTopics(List.of(topicName)).values();
+            Map<String, KafkaFuture<TopicDescription>> topics = client.describeTopics(List.of(topicName)).topicNameValues();
 
             try {
                 return topics.containsKey(topicName) && topics.get(topicName).get().name().equals(topicName);
-            } catch (InterruptedException e) {
-                LOGGER.error("error on checkTopicExistence", e);
-                return false;
-            } catch (ExecutionException e) {
+            } catch (InterruptedException | ExecutionException e) {
                 LOGGER.error("error on checkTopicExistence", e);
                 return false;
             }
         }));
     }
 
     private static <Rep> Rep getAdminClient(String bootstrapServer, Function<AdminClient, Rep> function) {
-        AdminClient adminClient = KafkaAdminClient.create(
-            Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer, AdminClientConfig.CLIENT_ID_CONFIG, "test")
-        );
-        try {
+        try (
+            AdminClient adminClient = KafkaAdminClient.create(
+                Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer, AdminClientConfig.CLIENT_ID_CONFIG, "test")
+            )
+        ) {
             return function.apply(adminClient);
-        } finally {
-            adminClient.close();
         }
     }
 }