Combine unclean leader election enabled/disabled tests into single loop

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pthirun

internal/venice-common/src/test/java/com/linkedin/venice/pubsub/adapter/kafka/admin/ApacheKafkaAdminAdapterTest.java

@@ -4,6 +4,7 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -597,52 +598,32 @@ public void testGetAllTopicRetentionsThrowsException() throws ExecutionException
   }
 
   @Test
-  public void testSetTopicConfigWithUncleanLeaderElectionDisabled() {
-    PubSubTopicConfiguration topicConfiguration =
-        new PubSubTopicConfiguration(Optional.of(1111L), true, Optional.of(222), 333L, Optional.empty());
-    topicConfiguration.setUncleanLeaderElectionEnable(Optional.of(false));
-
-    AlterConfigsResult alterConfigsResultMock = mock(AlterConfigsResult.class);
-    KafkaFuture<Void> alterConfigsKafkaFutureMock = mock(KafkaFuture.class);
-
-    Map<ConfigResource, Config> resourceConfigMap = new HashMap<>();
-    when(internalKafkaAdminClientMock.alterConfigs(any())).thenAnswer(invocation -> {
-      resourceConfigMap.putAll(invocation.getArgument(0));
-      return alterConfigsResultMock;
-    });
-    when(alterConfigsResultMock.all()).thenReturn(alterConfigsKafkaFutureMock);
-
-    kafkaAdminAdapter.setTopicConfig(testPubSubTopic, topicConfiguration);
-
-    assertEquals(resourceConfigMap.size(), 1);
-    for (Map.Entry<ConfigResource, Config> entry: resourceConfigMap.entrySet()) {
-      Config config = entry.getValue();
-      assertEquals(config.get(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG).value(), "false");
-    }
-  }
-
-  @Test
-  public void testSetTopicConfigWithUncleanLeaderElectionEnabled() {
-    PubSubTopicConfiguration topicConfiguration =
-        new PubSubTopicConfiguration(Optional.of(1111L), true, Optional.of(222), 333L, Optional.empty());
-    topicConfiguration.setUncleanLeaderElectionEnable(Optional.of(true));
-
-    AlterConfigsResult alterConfigsResultMock = mock(AlterConfigsResult.class);
-    KafkaFuture<Void> alterConfigsKafkaFutureMock = mock(KafkaFuture.class);
-
-    Map<ConfigResource, Config> resourceConfigMap = new HashMap<>();
-    when(internalKafkaAdminClientMock.alterConfigs(any())).thenAnswer(invocation -> {
-      resourceConfigMap.putAll(invocation.getArgument(0));
-      return alterConfigsResultMock;
-    });
-    when(alterConfigsResultMock.all()).thenReturn(alterConfigsKafkaFutureMock);
-
-    kafkaAdminAdapter.setTopicConfig(testPubSubTopic, topicConfiguration);
-
-    assertEquals(resourceConfigMap.size(), 1);
-    for (Map.Entry<ConfigResource, Config> entry: resourceConfigMap.entrySet()) {
-      Config config = entry.getValue();
-      assertEquals(config.get(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG).value(), "true");
+  public void testSetTopicConfigWithUncleanLeaderElectionSet() {
+    for (boolean enabled: new boolean[] { false, true }) {
+      // Reset mock state between iterations
+      reset(internalKafkaAdminClientMock);
+
+      PubSubTopicConfiguration topicConfiguration =
+          new PubSubTopicConfiguration(Optional.of(1111L), true, Optional.of(222), 333L, Optional.empty());
+      topicConfiguration.setUncleanLeaderElectionEnable(Optional.of(enabled));
+
+      AlterConfigsResult alterConfigsResultMock = mock(AlterConfigsResult.class);
+      KafkaFuture<Void> alterConfigsKafkaFutureMock = mock(KafkaFuture.class);
+
+      Map<ConfigResource, Config> resourceConfigMap = new HashMap<>();
+      when(internalKafkaAdminClientMock.alterConfigs(any())).thenAnswer(invocation -> {
+        resourceConfigMap.putAll(invocation.getArgument(0));
+        return alterConfigsResultMock;
+      });
+      when(alterConfigsResultMock.all()).thenReturn(alterConfigsKafkaFutureMock);
+
+      kafkaAdminAdapter.setTopicConfig(testPubSubTopic, topicConfiguration);
+
+      assertEquals(resourceConfigMap.size(), 1);
+      for (Map.Entry<ConfigResource, Config> entry: resourceConfigMap.entrySet()) {
+        Config config = entry.getValue();
+        assertEquals(config.get(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG).value(), Boolean.toString(enabled));
+      }
     }
   }