Refactor CCKafkaClientsIntegrationTestHarness and dependent classes.

Signed-off-by: Kyle Liberti <kliberti.us@gmail.com>

Author: kyguy

cruise-control-metrics-reporter/src/main/java/com/linkedin/kafka/cruisecontrol/metricsreporter/CruiseControlMetricsReporter.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.TimeoutException;
 import org.apache.kafka.clients.ClientUtils;
 import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.AlterConfigsResult;
@@ -375,7 +376,6 @@ protected void maybeIncreaseTopicPartitionCount() {
     String cruiseControlMetricsTopic = _metricsTopic.name();
 
     try {
-      // For compatibility with Kafka 4.0 and beyond we must use new API methods.
       TopicDescription topicDescription = getTopicDescription(_adminClient, cruiseControlMetricsTopic);
 
       if (topicDescription.partitions().size() < _metricsTopic.numPartitions()) {
@@ -514,68 +514,22 @@ private void setIfAbsent(Properties props, String key, String value) {
   }
 
   /**
-   * Attempts to retrieve the method for mapping topic names to futures from the {@link org.apache.kafka.clients.admin.DescribeTopicsResult} class.
-   * This method first tries to get the {@code topicNameValues()} method, which is available in Kafka 3.1.0 and later.
-   * If the method is not found, it falls back to trying to retrieve the {@code values()} method, which is available in Kafka 3.9.0 and earlier.
-   *
-   * If neither of these methods is found, a {@link RuntimeException} is thrown.
-   *
-   * <p>This method is useful for ensuring compatibility with both older and newer versions of Kafka clients.</p>
-   *
-   * @return the {@link Method} object representing the {@code topicNameValues()} or {@code values()} method.
-   * @throws RuntimeException if neither the {@code values()} nor {@code topicNameValues()} methods are found.
-   */
-  /* test */ static Method topicNameValuesMethod() {
-    //
-    Method topicDescriptionMethod = null;
-    try {
-      // First we try to get the topicNameValues() method
-      topicDescriptionMethod = DescribeTopicsResult.class.getMethod("topicNameValues");
-    } catch (NoSuchMethodException exception) {
-      LOG.info("Failed to get method topicNameValues() from DescribeTopicsResult class since we are probably on kafka 3.0.0 or older: ", exception);
-    }
-
-    if (topicDescriptionMethod == null) {
-      try {
-        // Second we try to get the values() method
-        topicDescriptionMethod = DescribeTopicsResult.class.getMethod("values");
-      } catch (NoSuchMethodException exception) {
-        LOG.info("Failed to get method values() from DescribeTopicsResult class: ", exception);
-      }
-    }
-
-    if (topicDescriptionMethod != null) {
-      return topicDescriptionMethod;
-    } else {
-      throw new RuntimeException("Unable to find both values() and topicNameValues() method in the DescribeTopicsResult class ");
-    }
-  }
-
-  /**
-   * Retrieves the {@link TopicDescription} for the specified Kafka topic, handling compatibility
-   * with Kafka versions 4.0 and above. This method uses reflection to invoke the appropriate method
-   * for retrieving topic description information, depending on the Kafka version.
+   * Retrieves the {@link TopicDescription} for the specified Kafka topic.
    *
    * @param adminClient The Kafka {@link AdminClient} used to interact with the Kafka cluster.
    * @param ccMetricsTopic The name of the Kafka topic for which the description is to be retrieved.
    *
    * @return The {@link TopicDescription} for the specified Kafka topic.
    *
-   * @throws KafkaTopicDescriptionException If an error occurs while retrieving the topic description,
-   *         or if the topic name retrieval method cannot be found or invoked properly. This includes
-   *         exceptions related to reflection (e.g., {@link NoSuchMethodException}), invocation issues,
-   *         execution exceptions, timeouts, and interruptions.
+   * @throws KafkaTopicDescriptionException If an error occurs while retrieving the topic description.
    */
-  /* test */ static TopicDescription getTopicDescription(AdminClient adminClient, String ccMetricsTopic) throws KafkaTopicDescriptionException {
+  /* test */ static TopicDescription getTopicDescription(Admin adminClient, String ccMetricsTopic) throws KafkaTopicDescriptionException {
     try {
-      // For compatibility with Kafka 4.0 and beyond we must use new API methods.
-      Method topicDescriptionMethod = topicNameValuesMethod();
-      DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(ccMetricsTopic));
-      Map<String, KafkaFuture<TopicDescription>> topicDescriptionMap = (Map<String, KafkaFuture<TopicDescription>>) topicDescriptionMethod
-              .invoke(describeTopicsResult);
-      return topicDescriptionMap.get(ccMetricsTopic).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS);
-    } catch (InvocationTargetException | IllegalAccessException | ExecutionException | InterruptedException | TimeoutException e) {
-      throw new KafkaTopicDescriptionException(String.format("Unable to retrieve config of Cruise Cruise Control metrics topic {}.",
+        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(ccMetricsTopic));
+        Map<String, KafkaFuture<TopicDescription>> topicDescriptionMap = describeTopicsResult.topicNameValues();
+        return topicDescriptionMap.get(ccMetricsTopic).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    } catch (ExecutionException | InterruptedException | TimeoutException e) {
+      throw new KafkaTopicDescriptionException(String.format("Unable to retrieve config of Cruise Cruise Control metrics topic %s.",
               ccMetricsTopic), e);
     }
   }

cruise-control-metrics-reporter/src/test/java/com/linkedin/kafka/cruisecontrol/metricsreporter/CruiseControlMetricsReporterAutoCreateTopicTest.java

@@ -5,10 +5,8 @@
 package com.linkedin.kafka.cruisecontrol.metricsreporter;
 
 import com.linkedin.kafka.cruisecontrol.metricsreporter.exception.KafkaTopicDescriptionException;
-import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.CCContainerizedKraftCluster;
 import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.CCKafkaClientsIntegrationTestHarness;
 import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.KafkaServerConfigs;
-import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.NewTopic;
@@ -17,7 +15,6 @@
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
-import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import java.util.Collections;
@@ -28,22 +25,14 @@
 import static org.junit.Assert.assertEquals;
 
 public class CruiseControlMetricsReporterAutoCreateTopicTest extends CCKafkaClientsIntegrationTestHarness {
-    protected static final String TOPIC = "CruiseControlMetricsReporterTest";
     protected static final String TEST_TOPIC = "TestTopic";
-    private CCContainerizedKraftCluster _cluster;
 
     /**
      * Setup the unit test.
      */
     @Before
     public void setUp() {
-        Properties adminClientProps = new Properties();
-        setSecurityConfigs(adminClientProps, "admin");
-
-        _cluster = new CCContainerizedKraftCluster(2, buildBrokerConfigs(), adminClientProps);
-        _cluster.start();
-        _bootstrapUrl = _cluster.getExternalBootstrapAddress();
-
+        super.setUp();
         // creating the "TestTopic" explicitly because the topic auto-creation is disabled on the broker
         Properties adminProps = new Properties();
         adminProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
@@ -76,28 +65,9 @@ public void setUp() {
         assertEquals(0, producerFailed.get());
     }
 
-    /**
-     * Tear down the unit test.
-     */
-    @After
-    public void tearDown() {
-        if (_cluster != null) {
-            _cluster.close();
-        }
-    }
-
     @Override
     public Properties overridingProps() {
-        Properties props = new Properties();
-        props.setProperty(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, CruiseControlMetricsReporter.class.getName());
-        props.setProperty(CruiseControlMetricsReporterConfig.config(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG),
-          "127.0.0.1:" + CCContainerizedKraftCluster.CONTAINER_INTERNAL_LISTENER_PORT);
-        props.put("listener.security.protocol.map", String.join(",",
-          CCContainerizedKraftCluster.CONTROLLER_LISTENER_NAME + ":PLAINTEXT",
-          CCContainerizedKraftCluster.INTERNAL_LISTENER_NAME + ":PLAINTEXT",
-          CCContainerizedKraftCluster.EXTERNAL_LISTENER_NAME + ":PLAINTEXT"));
-        props.setProperty(CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_REPORTER_INTERVAL_MS_CONFIG, "100");
-        props.setProperty(CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_CONFIG, TOPIC);
+        Properties props = super.overridingProps();
         // configure metrics topic auto-creation by the metrics reporter
         props.setProperty(CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_AUTO_CREATE_CONFIG, "true");
         props.setProperty(CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_AUTO_CREATE_TIMEOUT_MS_CONFIG, "5000");
@@ -106,8 +76,6 @@ public Properties overridingProps() {
         props.setProperty(CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_REPLICATION_FACTOR_CONFIG, "1");
         // disable topic auto-creation to leave the metrics reporter to create the metrics topic
         props.setProperty(KafkaServerConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG, "false");
-        props.setProperty(KafkaServerConfigs.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, "1");
-        props.setProperty(KafkaServerConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, "2");
         props.setProperty(KafkaServerConfigs.NUM_PARTITIONS_CONFIG, "2");
         return props;
     }

cruise-control-metrics-reporter/src/test/java/com/linkedin/kafka/cruisecontrol/metricsreporter/CruiseControlMetricsReporterTest.java

@@ -6,21 +6,18 @@
 
 import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.CruiseControlMetric;
 import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.MetricSerde;
-import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.CCContainerizedKraftCluster;
 import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.CCKafkaClientsIntegrationTestHarness;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Pattern;
-import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -33,42 +30,26 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.serialization.StringDeserializer;
-import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.testcontainers.kafka.KafkaContainer;
 
 import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter.DEFAULT_BOOTSTRAP_SERVERS_HOST;
 import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter.DEFAULT_BOOTSTRAP_SERVERS_PORT;
-import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_REPORTER_INTERVAL_MS_CONFIG;
 import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_AUTO_CREATE_CONFIG;
-import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_CONFIG;
 import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_NUM_PARTITIONS_CONFIG;
 import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_REPLICATION_FACTOR_CONFIG;
 import static com.linkedin.kafka.cruisecontrol.metricsreporter.metric.RawMetricType.*;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 public class CruiseControlMetricsReporterTest extends CCKafkaClientsIntegrationTestHarness {
-  private static final int NUM_OF_BROKERS = 2;
-  protected static final String TOPIC = "CruiseControlMetricsReporterTest";
-  protected static final String HOST = "127.0.0.1";
-  protected CCContainerizedKraftCluster _cluster;
-  protected List<Map<Object, Object>> _brokerConfigs;
-
   /**
    * Setup the unit test.
    */
   @Before
   public void setUp() {
-    Properties adminClientProps = new Properties();
-    setSecurityConfigs(adminClientProps, "admin");
-
-    _brokerConfigs = buildBrokerConfigs();
-    _cluster = new CCContainerizedKraftCluster(NUM_OF_BROKERS, _brokerConfigs, adminClientProps);
-    _cluster.start();
-    _bootstrapUrl = _cluster.getExternalBootstrapAddress();
-
+    super.setUp();
     Properties props = new Properties();
     props.setProperty(ProducerConfig.ACKS_CONFIG, "-1");
     AtomicInteger failed = new AtomicInteger(0);
@@ -87,34 +68,6 @@ public void onCompletion(RecordMetadata recordMetadata, Exception e) {
     assertEquals(0, failed.get());
   }
 
-  /**
-   * Tear down the unit test.
-   */
-  @After
-  public void tearDown() {
-    if (_cluster != null) {
-      _cluster.close();
-    }
-  }
-
-  @Override
-  public Properties overridingProps() {
-    Properties props = new Properties();
-    props.setProperty(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, CruiseControlMetricsReporter.class.getName());
-    props.setProperty(CruiseControlMetricsReporterConfig.config(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG),
-      HOST + ":" + CCContainerizedKraftCluster.CONTAINER_INTERNAL_LISTENER_PORT);
-    props.put("listener.security.protocol.map", String.join(",",
-      CCContainerizedKraftCluster.CONTROLLER_LISTENER_NAME + ":PLAINTEXT",
-      CCContainerizedKraftCluster.INTERNAL_LISTENER_NAME + ":PLAINTEXT",
-      CCContainerizedKraftCluster.EXTERNAL_LISTENER_NAME + ":PLAINTEXT"));
-    props.setProperty(CRUISE_CONTROL_METRICS_REPORTER_INTERVAL_MS_CONFIG, "100");
-    props.setProperty(CRUISE_CONTROL_METRICS_TOPIC_CONFIG, TOPIC);
-    props.setProperty("log.flush.interval.messages", "1");
-    props.setProperty("offsets.topic.replication.factor", "1");
-    props.setProperty("default.replication.factor", "2");
-    return props;
-  }
-
   @Test
   public void testReportingMetrics() {
     Properties props = new Properties();

cruise-control-metrics-reporter/src/test/java/com/linkedin/kafka/cruisecontrol/metricsreporter/utils/CCKafkaClientsIntegrationTestHarness.java

@@ -5,21 +5,65 @@
 package com.linkedin.kafka.cruisecontrol.metricsreporter.utils;
 
 import java.io.File;
+import java.util.List;
+import java.util.Map;
 import java.util.Properties;
+import com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter;
+import com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporterConfig;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.After;
+
+import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_REPORTER_INTERVAL_MS_CONFIG;
+import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporterConfig.CRUISE_CONTROL_METRICS_TOPIC_CONFIG;
 
 
 public abstract class CCKafkaClientsIntegrationTestHarness extends CCKafkaIntegrationTestHarness {
+  protected static final String TOPIC = "CruiseControlMetricsReporterTest";
+  protected CCContainerizedKraftCluster _cluster;
+  protected List<Map<Object, Object>> _brokerConfigs;
 
   @Override
   public void setUp() {
-    super.setUp();
+    Properties adminClientProps = new Properties();
+    setSecurityConfigs(adminClientProps, "admin");
+
+    _brokerConfigs = buildBrokerConfigs();
+    _cluster = new CCContainerizedKraftCluster(2, _brokerConfigs, adminClientProps);
+    _cluster.start();
+    _bootstrapUrl = _cluster.getExternalBootstrapAddress();
+  }
+
+  /**
+   * Tear down the unit test.
+   */
+  @After
+  public void tearDown() {
+    if (_cluster != null) {
+      _cluster.close();
+    }
+  }
+
+  @Override
+  public Properties overridingProps() {
+    Properties props = new Properties();
+    props.setProperty(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, CruiseControlMetricsReporter.class.getName());
+    props.setProperty(CruiseControlMetricsReporterConfig.config(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG),
+       "localhost:" + CCContainerizedKraftCluster.CONTAINER_INTERNAL_LISTENER_PORT);
+    props.put("listener.security.protocol.map", String.join(",",
+      CCContainerizedKraftCluster.CONTROLLER_LISTENER_NAME + ":PLAINTEXT",
+      CCContainerizedKraftCluster.INTERNAL_LISTENER_NAME + ":PLAINTEXT",
+      CCContainerizedKraftCluster.EXTERNAL_LISTENER_NAME + ":PLAINTEXT"));
+    props.setProperty(CRUISE_CONTROL_METRICS_REPORTER_INTERVAL_MS_CONFIG, "100");
+    props.setProperty(CRUISE_CONTROL_METRICS_TOPIC_CONFIG, TOPIC);
+    props.setProperty(KafkaServerConfigs.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, "1");
+    props.setProperty(KafkaServerConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, "2");
+    return props;
   }
 
   @javax.annotation.Nonnull