Repository: apache/kafka
File: storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java (Java)

Code before:
        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);
            rlmm.onBrokerReady();
            
            // Wait for initialization failure and exit procedure to be called
            TestUtils.waitForCondition(() -> exitCalled.get(), 
                "Exit procedure should be called due to initialization failure");
            
            // Verify exit code
            assertEquals(1, exitCode.get(), "Exit code should be 1");
        } finally {
            // Restore default exit procedure
            Exit.resetExitProcedure();
        }
    }

    @ClusterTest
    public void testRemoteLogMetadataTopicMinIsr() throws ExecutionException, InterruptedException {
        // Initialize the manager which will create the __remote_log_metadata topic
        TopicBasedRemoteLogMetadataManager rlmm = topicBasedRlmm();
        verifyRemoteLogMetadataTopicMinIsr(rlmm, (short) 2, "default value");
    }

    @ClusterTest
    public void testRemoteLogMetadataTopicMinIsrWithCustomValue() throws ExecutionException, InterruptedException, IOException {
        // Create a manager with custom min.isr value
        short customMinIsr = 3;
        Map<String, Object> overrideProps = Map.of(
            TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_MIN_ISR_PROP, customMinIsr
        );
        
        TopicBasedRemoteLogMetadataManager customRlmm = RemoteLogMetadataManagerTestUtils.builder()
            .bootstrapServers(clusterInstance.bootstrapServers())
            .overrideRemoteLogMetadataManagerProps(overrideProps)
            .build();
        
        try {
            verifyRemoteLogMetadataTopicMinIsr(customRlmm, customMinIsr, "custom value");
        } finally {
            customRlmm.close();
        }
    }

    private void verifyRemoteLogMetadataTopicMinIsr(TopicBasedRemoteLogMetadataManager rlmm, 
                                                     short expectedMinIsr, 
                                                     String valueDescription) 

Review comment (suggestion):
```suggestion
        verifyRemoteLogMetadataTopicMinIsr(remoteLogMetadataManager, (short) 2, "default value");
```

Code after:

        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);
            rlmm.onBrokerReady();
            
            // Wait for initialization failure and exit procedure to be called
            TestUtils.waitForCondition(() -> exitCalled.get(), 
                "Exit procedure should be called due to initialization failure");
            
            // Verify exit code
            assertEquals(1, exitCode.get(), "Exit code should be 1");
        } finally {
            // Restore default exit procedure
            Exit.resetExitProcedure();
        }
    }

    @ClusterTest
    public void testRemoteLogMetadataTopicWithDefaultMinIsr() throws ExecutionException, InterruptedException {
        // Initialize the manager which will create the __remote_log_metadata topic
        TopicBasedRemoteLogMetadataManager topicBasedRemoteLogMetadataManager = topicBasedRlmm();
        verifyRemoteLogMetadataTopicWithMinIsr(topicBasedRemoteLogMetadataManager,
                                               TopicBasedRemoteLogMetadataManagerConfig.DEFAULT_REMOTE_LOG_METADATA_TOPIC_MIN_ISR,
                                               "default value");
    }

    @ClusterTest
    public void testRemoteLogMetadataTopicWithCustomMinIsr() throws ExecutionException, InterruptedException, IOException {
        // Create a manager with custom min.isr value
        short customMinIsr = 3;
        Map<String, Object> overrideProps = Map.of(
            TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_MIN_ISR_PROP, customMinIsr
        );
        try (TopicBasedRemoteLogMetadataManager customRlmm = RemoteLogMetadataManagerTestUtils.builder()
                .bootstrapServers(clusterInstance.bootstrapServers())
                .overrideRemoteLogMetadataManagerProps(overrideProps)
                .build()) {
            verifyRemoteLogMetadataTopicWithMinIsr(customRlmm, customMinIsr, "custom value");
        }
    }

    private void verifyRemoteLogMetadataTopicWithMinIsr(TopicBasedRemoteLogMetadataManager rlmm,
                                                        short expectedMinIsr,
                                                        String valueDescription)
                                                        throws ExecutionException, InterruptedException {
        try (Admin admin = clusterInstance.admin()) {