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

Code before:
        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) 
            throws ExecutionException, InterruptedException {
        try (Admin admin = clusterInstance.admin()) {
            String metadataTopic = TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_NAME;
            
            // Wait for the topic to be created
            // RemoteLogMetadataManagerTestUtils uses 3 partitions for the metadata topic
            clusterInstance.waitTopicCreation(metadataTopic, 3);
            
            // Verify the topic exists
            assertTrue(rlmm.doesTopicExist(admin, metadataTopic));
            
            // Describe the topic configs to verify min.insync.replicas
            ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, metadataTopic);
            DescribeConfigsResult describeResult = admin.describeConfigs(List.of(topicResource));
            Config config = describeResult.all().get().get(topicResource);
            
            assertTrue(config != null, "Topic config should not be null");
            ConfigEntry minIsrEntry = config.get(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG);
            assertTrue(minIsrEntry != null, "min.insync.replicas config should exist");
            assertEquals(String.valueOf(expectedMinIsr), minIsrEntry.value(), 
                "min.insync.replicas should be " + expectedMinIsr + " (" + valueDescription + ")");
        }
    }
}


Review comment (suggestion):
```suggestion
            clusterInstance.waitTopicCreation(metadataTopic, RemoteLogMetadataManagerTestUtils.METADATA_TOPIC_PARTITIONS_COUNT);
```

Code after:
    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()) {
            String metadataTopic = TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_NAME;
            
            // Wait for the topic to be created
            clusterInstance.waitTopicCreation(metadataTopic, RemoteLogMetadataManagerTestUtils.METADATA_TOPIC_PARTITIONS_COUNT);
            
            // Verify the topic exists
            assertTrue(rlmm.doesTopicExist(admin, metadataTopic));
            
            // Describe the topic configs to verify min.insync.replicas
            ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, metadataTopic);
            DescribeConfigsResult describeResult = admin.describeConfigs(List.of(topicResource));
            Config config = describeResult.all().get().get(topicResource);

            assertNotNull(config, "Topic config should not be null");
            ConfigEntry minIsrEntry = config.get(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG);
            assertNotNull(minIsrEntry, "min.insync.replicas config should exist");
            assertEquals(String.valueOf(expectedMinIsr), minIsrEntry.value(), 
                "min.insync.replicas should be " + expectedMinIsr + " (" + valueDescription + ")");
        }
    }
}