Skip to content

Replace deprecated methods to support Kafka 4.0.0 #2254

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Mar 17, 2025
Merged

Replace deprecated methods to support Kafka 4.0.0 #2254

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
5715b0d
Replace deprecated methods
ShubhamRwt Feb 26, 2025
57cb00d
Added suggestions
ShubhamRwt Mar 4, 2025
2eb5a98
using invoke method
ShubhamRwt Mar 6, 2025
237d547
Added suggestions by Kyle
ShubhamRwt Mar 11, 2025
1dbff28
Fixing checkstyle issues
ShubhamRwt Mar 11, 2025
21ae74b
Fix failing tests
ShubhamRwt Mar 12, 2025
93bd379
Added suggestions by Kyle
ShubhamRwt Mar 13, 2025
db8af2b
Removed not required logging
ShubhamRwt Mar 14, 2025
835816c
Added suggestion by Kyle
ShubhamRwt Mar 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 52 additions & 7 deletions ...n/java/com/linkedin/kafka/cruisecontrol/metricsreporter/CruiseControlMetricsReporter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
Expand All @@ -42,6 +43,7 @@
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;
Expand Down Expand Up @@ -370,16 +372,22 @@ protected void maybeUpdateTopicConfig() {

protected void maybeIncreaseTopicPartitionCount() {
String cruiseControlMetricsTopic = _metricsTopic.name();

try {
// Retrieve topic partition count to check and update.
TopicDescription topicDescription =
_adminClient.describeTopics(Collections.singletonList(cruiseControlMetricsTopic)).values()
.get(cruiseControlMetricsTopic).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS);
Method topicDescriptionMethod = topicNameValuesMethod();

DescribeTopicsResult describeTopicsResult = _adminClient.describeTopics(Collections.singletonList(cruiseControlMetricsTopic));

Map<String, KafkaFuture<TopicDescription>> topicDescriptionMap = (Map<String, KafkaFuture<TopicDescription>>) topicDescriptionMethod
.invoke(describeTopicsResult);

TopicDescription topicDescription = topicDescriptionMap.get(cruiseControlMetricsTopic).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS);
Copy link
Contributor

@kyguy kyguy Mar 13, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could also abstract these the above lines into a getTopicDescription() method to simplify the code across here and the tests even further. If you want to go that route you could separate the code like this:

  /**
   * 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.
   *
   * @param _adminClient The Kafka {@link AdminClient} used to interact with the Kafka cluster.
   * @param cruiseControlMetricsTopic 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.
   */
  /* test */ static TopicDescription getTopicDescription(AdminClient _adminClient, String cruiseControlMetricsTopic) 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(cruiseControlMetricsTopic));
      Map<String, KafkaFuture<TopicDescription>> topicDescriptionMap = (Map<String, KafkaFuture<TopicDescription>>) topicDescriptionMethod
              .invoke(describeTopicsResult);

      return topicDescriptionMap.get(cruiseControlMetricsTopic).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS);
    }
    catch (InvocationTargetException | IllegalAccessException | ExecutionException | InterruptedException | TimeoutException | NoSuchMethodException e) {
      throw new KafkaTopicDescriptionException(String.format("Unable to retrieve config of Cruise Cruise Control metrics topic {}.", cruiseControlMetricsTopic), e);
    }
  }

Note that if you did this, it would probably be a good idea to create a custom exception class in the cruisecontrol.metricsreporter/exception folder so you don't have to add all the Exception classes to the method signatures where this method is called.

public class KafkaTopicDescriptionException extends Exception {
    public KafkaTopicDescriptionException(String message, Throwable cause) {
        super(message, cause);
    }
}

This way, you can just call getTopicDescription() whenever you need the topic description for a topic and not worry about calling and invoking the right method there as well, it would just be done in the method here in the CruiseControlMetricsReporter class.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was looking into applying this suggestion but there is one more thing -> The call made in CruiseControlMetricsReporter is

      return topicDescriptionMap.get(cruiseControlMetricsTopic).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS);

while in the test we don't use the get method with CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS parameters

 `topicDescriptionMap.get(TOPIC).get()`

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does the inclusion of .get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS) interfere with the tests? From what I understand, it still returns the TopicDescription object we need.

Even if the inclusion of .get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS) did cause any problems in the tests, couldn't we remove it from the getTopicDescription() method itself and append it to the end of the places where we need it?


if (topicDescription.partitions().size() < _metricsTopic.numPartitions()) {
_adminClient.createPartitions(Collections.singletonMap(cruiseControlMetricsTopic,
NewPartitions.increaseTo(_metricsTopic.numPartitions())));
_adminClient.createPartitions(Collections.singletonMap(cruiseControlMetricsTopic, NewPartitions.increaseTo(_metricsTopic.numPartitions())));
}
} catch (InterruptedException | ExecutionException | TimeoutException e) {

} catch (InterruptedException | ExecutionException | TimeoutException | InvocationTargetException | IllegalAccessException e) {
LOG.warn("Partition count increase to {} for topic {} failed{}.", _metricsTopic.numPartitions(), cruiseControlMetricsTopic,
(e.getCause() instanceof ReassignmentInProgressException) ? " due to ongoing reassignment" : "", e);
}
Expand Down Expand Up @@ -511,4 +519,41 @@ 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 4.0.0 or later.
* If the method is not found, it falls back to trying to retrieve the {@code values()} method, which is available in Kafka 3.0.0 or 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 since we are probably on kafka 3.0.0 or older: ", exception);
}
}

if (topicDescriptionMethod != null) {
return topicDescriptionMethod;
} else {
throw new RuntimeException("Unable to find both values() and topicNameValues() method in the DescribeTopicsResult class ");
}
}
}
61 changes: 60 additions & 1 deletion .../kafka/cruisecontrol/metricsreporter/CruiseControlMetricsReporterAutoCreateTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,20 +9,27 @@
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.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.network.SocketServerConfigs;
import org.apache.kafka.server.config.ReplicationConfigs;
import org.apache.kafka.server.config.ServerLogConfigs;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
Expand All @@ -32,6 +39,7 @@
public class CruiseControlMetricsReporterAutoCreateTopicTest extends CCKafkaClientsIntegrationTestHarness {
protected static final String TOPIC = "CruiseControlMetricsReporterTest";
protected static final String TEST_TOPIC = "TestTopic";
private static final Logger LOG = LoggerFactory.getLogger(CruiseControlMetricsReporterAutoCreateTopicTest.class);

/**
* Setup the unit test.
Expand Down Expand Up @@ -107,9 +115,60 @@ public void testAutoCreateMetricsTopic() throws ExecutionException, InterruptedE
Properties props = new Properties();
props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
AdminClient adminClient = AdminClient.create(props);
TopicDescription topicDescription = adminClient.describeTopics(Collections.singleton(TOPIC)).values().get(TOPIC).get();

Method topicDescriptionMethod = topicNameValuesMethod();

Map<String, KafkaFuture<TopicDescription>> topicDescriptionMap;

try {
topicDescriptionMap =
(Map<String, KafkaFuture<TopicDescription>>) topicDescriptionMethod
.invoke(adminClient.describeTopics(Collections.singleton(TOPIC)));
} catch (IllegalAccessException | InvocationTargetException e) {
throw new RuntimeException(e);
}

TopicDescription topicDescription = topicDescriptionMap.get(TOPIC).get();
// assert that the metrics topic was created with partitions and replicas as configured for the metrics report auto-creation
assertEquals(1, topicDescription.partitions().size());
assertEquals(1, topicDescription.partitions().get(0).replicas().size());
}

/**
* 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 4.0.0 or later.
* If the method is not found, it falls back to trying to retrieve the {@code values()} method, which is available in Kafka 3.0.0 or 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 e) {
LOG.info("Failed to get method topicNameValues() from DescribeTopicsResult class since we are probably on kafka 3.0.0 or older: ", e);
}

if (topicDescriptionMethod == null) {
try {
// Second we try to get the values() method
topicDescriptionMethod = DescribeTopicsResult.class.getMethod("values");
} catch (NoSuchMethodException e) {
LOG.info("Failed to get method values() from DescribeTopicsResult class since we are probably on kafka 3.0.0 or older: ", e);
}
}

if (topicDescriptionMethod != null) {
return topicDescriptionMethod;
} else {
throw new RuntimeException("Unable to find both values() and topicNameValues() method in the DescribeTopicsResult class ");
}
}
}
69 changes: 67 additions & 2 deletions ...va/com/linkedin/kafka/cruisecontrol/metricsreporter/CruiseControlMetricsReporterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@
import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.MetricSerde;
import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.CCEmbeddedBroker;
import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.CCKafkaClientsIntegrationTestHarness;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
Expand All @@ -22,6 +24,7 @@
import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.CCKafkaTestUtils;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
Expand All @@ -33,6 +36,7 @@
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.network.SocketServerConfigs;
Expand All @@ -41,6 +45,8 @@
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter.DEFAULT_BOOTSTRAP_SERVERS_HOST;
import static com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter.DEFAULT_BOOTSTRAP_SERVERS_PORT;
Expand All @@ -54,6 +60,7 @@
public class CruiseControlMetricsReporterTest extends CCKafkaClientsIntegrationTestHarness {
protected static final String TOPIC = "CruiseControlMetricsReporterTest";
private static final String HOST = "127.0.0.1";
private static final Logger LOG = LoggerFactory.getLogger(CruiseControlMetricsReporterAutoCreateTopicTest.class);

/**
* Setup the unit test.
Expand Down Expand Up @@ -190,7 +197,19 @@ public void testUpdatingMetricsTopicConfig() throws ExecutionException, Interrup
setSecurityConfigs(props, "admin");
props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
AdminClient adminClient = AdminClient.create(props);
TopicDescription topicDescription = adminClient.describeTopics(Collections.singleton(TOPIC)).values().get(TOPIC).get();

Method topicDescriptionMethod = topicNameValuesMethod();

Map<String, KafkaFuture<TopicDescription>> topicDescriptionMap;

try {
topicDescriptionMap = (Map<String, KafkaFuture<TopicDescription>>) topicDescriptionMethod
.invoke(adminClient.describeTopics(Collections.singleton(TOPIC)));
} catch (IllegalAccessException | InvocationTargetException e) {
throw new RuntimeException(e);
}

TopicDescription topicDescription = topicDescriptionMap.get(TOPIC).get();
assertEquals(1, topicDescription.partitions().size());
// Shutdown broker
_brokers.get(0).shutdown();
Expand All @@ -204,8 +223,16 @@ public void testUpdatingMetricsTopicConfig() throws ExecutionException, Interrup
broker.startup();
// Wait for broker to boot up
Thread.sleep(5000);

try {
topicDescriptionMap = (Map<String, KafkaFuture<TopicDescription>>) topicDescriptionMethod
.invoke(adminClient.describeTopics(Collections.singleton(TOPIC)));
} catch (IllegalAccessException | InvocationTargetException e) {
throw new RuntimeException(e);
}

// Check whether the topic config is updated
topicDescription = adminClient.describeTopics(Collections.singleton(TOPIC)).values().get(TOPIC).get();
topicDescription = topicDescriptionMap.get(TOPIC).get();
assertEquals(2, topicDescription.partitions().size());
}

Expand Down Expand Up @@ -244,4 +271,42 @@ public void testGetKafkaBootstrapServersConfigure() {
assertEquals(DEFAULT_BOOTSTRAP_SERVERS_HOST, bootstrapServers.split(":")[0]);
assertEquals(DEFAULT_BOOTSTRAP_SERVERS_PORT, bootstrapServers.split(":")[1]);
}

/**
* 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 4.0.0 or later.
* If the method is not found, it falls back to trying to retrieve the {@code values()} method, which is available in Kafka 3.0.0 or 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 since we are probably on kafka 3.0.0 or older: ", exception);
}
}

if (topicDescriptionMethod != null) {
return topicDescriptionMethod;
} else {
throw new RuntimeException("Unable to find both values() and topicNameValues() method in the DescribeTopicsResult class ");
}
}
}