KAFKA-19371: Don't create the __remote_log_metadata topic when it already exists during broker restarts (#19899)

* The CREATE_TOPIC request gets issued only when it is clear that the
topic does not exist in the cluster.
* When the request to describe the topic gets timed-out or any exception
thrown other than UnknownTopicOrPartitionException, then the same gets
re-thrown and the describe/create topic request gets retried in the next
iteration until the initializationRetryMaxTimeoutMs gets breached.

Fixes: https://issues.apache.org/jira/browse/KAFKA-19371

Reviewers: Luke Chen <[email protected]>, Kamal Chandraprakash
<[email protected]>

---------

Co-authored-by: stroller.fu <[email protected]>

Author: jiafu1115

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java

@@ -26,6 +26,7 @@
 import org.apache.kafka.common.config.TopicConfig;
 import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.internals.FatalExitError;
 import org.apache.kafka.common.utils.KafkaThread;
 import org.apache.kafka.common.utils.Time;
@@ -468,7 +469,7 @@ private void initializeResources() {
         }
     }
 
-    boolean doesTopicExist(Admin adminClient, String topic) {
+    boolean doesTopicExist(Admin adminClient, String topic) throws ExecutionException, InterruptedException {
         try {
             TopicDescription description = adminClient.describeTopics(Set.of(topic))
                     .topicNameValues()
@@ -477,13 +478,14 @@ boolean doesTopicExist(Admin adminClient, String topic) {
             if (description != null) {
                 log.info("Topic {} exists. TopicId: {}, numPartitions: {}, ", topic,
                         description.topicId(), description.partitions().size());
-            } else {
-                log.info("Topic {} does not exist.", topic);
             }
-            return description != null;
+            return true;
         } catch (ExecutionException | InterruptedException ex) {
-            log.info("Topic {} does not exist. Error: {}", topic, ex.getCause().getMessage());
-            return false;
+            if (ex.getCause() instanceof UnknownTopicOrPartitionException) {
+                log.info("Topic {} does not exist", topic);
+                return false;
+            }
+            throw ex;
         }
     }
 
@@ -544,7 +546,7 @@ private boolean createTopic(Admin adminClient, NewTopic newTopic) {
                 log.info("Topic [{}] already exists", topic);
                 doesTopicExist = true;
             } else {
-                log.error("Encountered error while creating {} topic.", topic, e);
+                log.error("Encountered error while querying or creating {} topic.", topic, e);
             }
         }
         return doesTopicExist;

storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java

@@ -17,7 +17,10 @@
 package org.apache.kafka.server.log.remote.metadata.storage;
 
 import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.DescribeTopicsResult;
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
@@ -46,9 +49,12 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anySet;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 @ClusterTestDefaults(brokers = 3)
 public class TopicBasedRemoteLogMetadataManagerTest {
@@ -89,14 +95,33 @@ public void testDoesTopicExist() throws ExecutionException, InterruptedException
     }
 
     @ClusterTest
-    public void testTopicDoesNotExist() {
+    public void testTopicDoesNotExist() throws ExecutionException, InterruptedException {
         try (Admin admin = clusterInstance.admin()) {
             String topic = "dummy-test-topic";
             boolean doesTopicExist = topicBasedRlmm().doesTopicExist(admin, topic);
             assertFalse(doesTopicExist);
         }
     }
 
+    @ClusterTest
+    public void testDoesTopicExistWithAdminClientExecutionError() throws ExecutionException, InterruptedException {
+        // Create a mock Admin client that throws an ExecutionException (not UnknownTopicOrPartitionException)
+        Admin mockAdmin = mock(Admin.class);
+        DescribeTopicsResult mockDescribeTopicsResult = mock(DescribeTopicsResult.class);
+        KafkaFuture<TopicDescription> mockFuture = mock(KafkaFuture.class);
+        
+        String topic = "test-topic";
+        
+        // Set up the mock to throw a RuntimeException wrapped in ExecutionException
+        when(mockAdmin.describeTopics(anySet())).thenReturn(mockDescribeTopicsResult);
+        when(mockDescribeTopicsResult.topicNameValues()).thenReturn(Map.of(topic, mockFuture));
+        when(mockFuture.get()).thenThrow(new ExecutionException("Admin client connection error", new RuntimeException("Connection failed")));
+        
+        // The method should re-throw the ExecutionException since it's not an UnknownTopicOrPartitionException
+        TopicBasedRemoteLogMetadataManager rlmm = topicBasedRlmm();
+        assertThrows(ExecutionException.class, () -> rlmm.doesTopicExist(mockAdmin, topic));
+    }
+
     @ClusterTest
     public void testWithNoAssignedPartitions() {
         // This test checks simple lifecycle of TopicBasedRemoteLogMetadataManager with out assigning any leader/follower partitions.