Replace deprecated methods to support Kafka 4.0.0

[ShubhamRwt]

Summary

1. Why: Cruise Control metrics reporter package seems to be using some deprecated method for some time now which are completely removed in Kafka 4.0.0. These deprecated method makes CC incompatible with Kafka 4.0 and beyond.

2025-02-22T22:27:38,802 ERROR Uncaught exception in thread 'CruiseControlMetricsReporterRunner': (org.apache.kafka.common.utils.KafkaThread) [CruiseControlMetricsReporterRunner]
java.lang.NoSuchMethodError: 'java.util.Map org.apache.kafka.clients.admin.DescribeTopicsResult.values()'
        at com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter.maybeIncreaseTopicPartitionCount(CruiseControlMetricsReporter.java:376) ~[cruise-control-metrics-reporter-2.5.141.jar:?]
        at com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter.maybeUpdateCruiseControlMetricsTopic(CruiseControlMetricsReporter.java:348) ~[cruise-control-metrics-reporter-2.5.141.jar:?]
        at com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter.run(CruiseControlMetricsReporter.java:395) ~[cruise-control-metrics-reporter-2.5.141.jar:?]
        at java.base/java.lang.Thread.run(Thread.java:840) [?:?] 

2. What: This PR replaces the deprecated method with the ones that should be used in place of them .

Categorization

• documentation
• bugfix
• new feature
• refactor
• security/CVE
• other

This PR resolves # if any.

[kyguy]

Hey @CCisGG could you take a look at this one? It is a small change, it future-proofs the Cruise Control metric reporter so it can be run on brokers using Kafka 4.0.

[CCisGG]

Looks good. Thanks! One minor question: will people use older kafka version be bothered from this change?

[ShubhamRwt]

The replaced method are deprecated since Kafka 3.1.0 and the new methods were also introduced in Kafka 3.1.0. Sof if brokers are using Kafka older then 3.1.0, only then it can be a problem

[kyguy]

The replaced method are deprecated since Kafka 3.1.0 and the new methods were also introduced in Kafka 3.1.0. Sof if brokers are using Kafka older then 3.1.0, only then it can be a problem

+1

Here is a reference to the deprecated methods in the Kafka 3.1 API docs [1]

[1] https://kafka.apache.org/31/javadoc/org/apache/kafka/clients/admin/DescribeTopicsResult.html#values()

[CCisGG]

Oops. We are internally still using kafka 3.0 so this will break our internal build. Can we choose the API based on the kafka version?

[CCisGG]

We are internally still using kafka 3.0 so this will break our internal build. Can we choose the API based on the kafka version?

[ShubhamRwt]

Thanks for the info @CCisGG, I will add the corresponding changes to the PR

[CCisGG]

Hi @ShubhamRwt thanks for the change!

However I think this won't work, because the compiler still see both methods at compile time and one of the method doesn't exist. I think it will cause build errors.

I think you need to use invoke to invoke the method calls, instead of directly calling these methods. Both the old method and new method might need to called by "invoke".

[kyguy]

Nice work @ShubhamRwt, just took a quick pass, was thinking that it might be easier/neater if we organize the DescribeTopicsResult method detection logic into its own method in the CruiseControlMetricReporter class, we could call it something like "topicNameValuesMethod()".

I left more specific details in the code comments, let me know what you think!

[kyguy]

I would do the same for the tests, we can separate the DescribeTopicsResult method detection logic into its own method, topicNameValuesMethod() in the CruiseControlMetricRepoter class, then call topicNameValuesMethod() in our tests to save us from the extra lines.

Suggested change

	Method topicDescriptionMethod = null;
	Method topicDescriptionMethod = topicNameValuesMethod();

[kyguy]

It may be worth separating out the DescribeTopicsResult method detection into a separate method. This would keep the focus of the maybeIncreaseTopicPartitionCount() method on the original logic.

We could separate the DescribeTopicsResult method detection logic out like this:

/**
   * 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.x 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.x 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 = Class.forName("org.apache.kafka.clients.admin.DescribeTopicsResult").getMethod("topicNameValues");
    } catch (ClassNotFoundException | 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 = Class.forName("org.apache.kafka.clients.admin.DescribeTopicsResult").getMethod("values");
      } catch (ClassNotFoundException | 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 ");
    }
  }

Then call it like this:

Suggested change

	Method topicDescriptionMethod = null;
	Method topicDescriptionMethod = topicNameValuesMethod();

[kyguy]

I would do the same for the tests, we can separate the DescribeTopicsResult method detection logic into its own method, topicNameValuesMethod() in the CruiseControlMetricRepoter class, then call topicNameValuesMethod() in our tests to save us from the extra lines.

Suggested change

	Method topicDescriptionMethod = null;
	Method topicDescriptionMethod = topicNameValuesMethod();

[ShubhamRwt]

@CCisGG @kyguy is it good to merge now?

[kyguy]

Thanks for the updates @ShubhamRwt, looking better, I have left a little more feedback if you are interested, let me know what you think!

[kyguy]

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.

[ShubhamRwt]

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()`

[kyguy]

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?

[kyguy]

Looking even better @ShubhamRwt, I left one last comment about creating a getTopicDescription() method to further reduce lines, let me know what you think!

[kyguy]

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?

[shaneikennedy]

Sorry for the drive by comment here but does this get us compatibility with 3.9 as well? I imagine there will be lots of teams upgrading older versions of kafka to 3.9 before making the switch to 4.0, so 3.9 compatibility feels quite important

[ShubhamRwt]

@shaneikennedy This PR is mostly to make sure that the cruise control metrics reporter can run on broker using 4.0 since it has multiple deprecated methods which are removed in Kafka 4.0.0(The metrics reporter works correctly with 3.9.0 since the deprecated method are not deleted in 3.9.0).

[shaneikennedy]

@shaneikennedy This PR is mostly to make sure that the cruise control metrics reporter can run on broker using 4.0 since it has multiple deprecated methods which are removed in Kafka 4.0.0(The metrics reporter works correctly with 3.9.0 since the deprecated method are not deleted in 3.9.0).

Gotcha, thanks for the info 🙌

[kyguy]

Looks great @ShubhamRwt!

@CCisGG Could you take another pass when you get a chance?

[CCisGG]

LGTM. Thanks @kyguy

[ShubhamRwt]

Hi @CCisGG, can you help me understand the release cadence of Cruise Control? Are we going to have a release coming out in the next few days?

We were planning to ship the new release of Cruise Control with Strimzi 0.46.0 release which is going to use Kafka 4.0.0 since it will contain the fix[1] for issues we are currently having with Kafka 4.0.0.

[1] #2254

[CCisGG]

@ShubhamRwt it's usually on demand. I'll make a new release soon.

[ShubhamRwt]

Thanks a lot @CCisGG