KAFKA-17830 Cover unit tests for TBRLMM init failure scenarios (#19076)

Add unit tests for TBRLMM when initializing clients.

Reviewers: Chia-Ping Tsai <[email protected]>

Author: FrankYang0529

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

@@ -555,6 +555,10 @@ boolean isInitialized() {
         return initialized.get();
     }
 
+    boolean isInitializationFailed() {
+        return initializationFailed;
+    }
+
     private void ensureInitializedAndNotClosed() {
         if (initializationFailed) {
             // If initialization is failed, shutdown the broker.

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

@@ -29,12 +29,14 @@
 import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
 import org.apache.kafka.server.log.remote.storage.RemoteResourceNotFoundException;
 import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
+import org.apache.kafka.test.TestUtils;
 
 import org.junit.jupiter.api.AfterEach;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -307,4 +309,16 @@ public void testRemoteLogSizeCalculationWithSegmentsHavingNonExistentEpochs() th
         assertEquals(0, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 9001));
     }
 
+    @ClusterTest
+    public void testInitializationFailure() throws IOException, InterruptedException {
+        try (TopicBasedRemoteLogMetadataManager rlmm = new TopicBasedRemoteLogMetadataManager()) {
+            // configure rlmm without bootstrap servers, so it will fail to initialize admin client.
+            Map<String, Object> configs = Map.of(
+                TopicBasedRemoteLogMetadataManagerConfig.LOG_DIR, TestUtils.tempDirectory("rlmm_segs_").getAbsolutePath(),
+                TopicBasedRemoteLogMetadataManagerConfig.BROKER_ID, 0
+            );
+            rlmm.configure(configs);
+            TestUtils.waitForCondition(rlmm::isInitializationFailed, "Initialization should fail");
+        }
+    }
 }