|
5 | 5 | package com.linkedin.kafka.cruisecontrol.metricsreporter; |
6 | 6 |
|
7 | 7 | import com.linkedin.kafka.cruisecontrol.metricsreporter.exception.CruiseControlMetricsReporterException; |
| 8 | +import com.linkedin.kafka.cruisecontrol.metricsreporter.exception.KafkaTopicDescriptionException; |
8 | 9 | import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.CruiseControlMetric; |
9 | 10 | import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.MetricsUtils; |
10 | 11 | import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.MetricSerde; |
|
33 | 34 | import org.apache.kafka.clients.admin.Config; |
34 | 35 | import org.apache.kafka.clients.admin.CreateTopicsResult; |
35 | 36 | import org.apache.kafka.clients.admin.DescribeConfigsResult; |
| 37 | +import org.apache.kafka.clients.admin.DescribeTopicsResult; |
36 | 38 | import org.apache.kafka.clients.admin.NewPartitions; |
37 | 39 | import org.apache.kafka.clients.admin.NewTopic; |
38 | 40 | import org.apache.kafka.clients.admin.TopicDescription; |
|
42 | 44 | import org.apache.kafka.clients.producer.ProducerRecord; |
43 | 45 | import org.apache.kafka.clients.producer.RecordMetadata; |
44 | 46 | import org.apache.kafka.common.KafkaException; |
| 47 | +import org.apache.kafka.common.KafkaFuture; |
45 | 48 | import org.apache.kafka.common.config.ConfigException; |
46 | 49 | import org.apache.kafka.common.config.ConfigResource; |
47 | 50 | import org.apache.kafka.common.config.TopicConfig; |
@@ -370,16 +373,15 @@ protected void maybeUpdateTopicConfig() { |
370 | 373 |
|
371 | 374 | protected void maybeIncreaseTopicPartitionCount() { |
372 | 375 | String cruiseControlMetricsTopic = _metricsTopic.name(); |
| 376 | + |
373 | 377 | try { |
374 | | - // Retrieve topic partition count to check and update. |
375 | | - TopicDescription topicDescription = |
376 | | - _adminClient.describeTopics(Collections.singletonList(cruiseControlMetricsTopic)).values() |
377 | | - .get(cruiseControlMetricsTopic).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS); |
| 378 | + // For compatibility with Kafka 4.0 and beyond we must use new API methods. |
| 379 | + TopicDescription topicDescription = getTopicDescription(_adminClient, cruiseControlMetricsTopic); |
| 380 | + |
378 | 381 | if (topicDescription.partitions().size() < _metricsTopic.numPartitions()) { |
379 | | - _adminClient.createPartitions(Collections.singletonMap(cruiseControlMetricsTopic, |
380 | | - NewPartitions.increaseTo(_metricsTopic.numPartitions()))); |
| 382 | + _adminClient.createPartitions(Collections.singletonMap(cruiseControlMetricsTopic, NewPartitions.increaseTo(_metricsTopic.numPartitions()))); |
381 | 383 | } |
382 | | - } catch (InterruptedException | ExecutionException | TimeoutException e) { |
| 384 | + } catch (KafkaTopicDescriptionException e) { |
383 | 385 | LOG.warn("Partition count increase to {} for topic {} failed{}.", _metricsTopic.numPartitions(), cruiseControlMetricsTopic, |
384 | 386 | (e.getCause() instanceof ReassignmentInProgressException) ? " due to ongoing reassignment" : "", e); |
385 | 387 | } |
@@ -511,4 +513,70 @@ private void setIfAbsent(Properties props, String key, String value) { |
511 | 513 | } |
512 | 514 | } |
513 | 515 |
|
| 516 | + /** |
| 517 | + * Attempts to retrieve the method for mapping topic names to futures from the {@link org.apache.kafka.clients.admin.DescribeTopicsResult} class. |
| 518 | + * This method first tries to get the {@code topicNameValues()} method, which is available in Kafka 3.1.0 and later. |
| 519 | + * 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. |
| 520 | + * |
| 521 | + * If neither of these methods is found, a {@link RuntimeException} is thrown. |
| 522 | + * |
| 523 | + * <p>This method is useful for ensuring compatibility with both older and newer versions of Kafka clients.</p> |
| 524 | + * |
| 525 | + * @return the {@link Method} object representing the {@code topicNameValues()} or {@code values()} method. |
| 526 | + * @throws RuntimeException if neither the {@code values()} nor {@code topicNameValues()} methods are found. |
| 527 | + */ |
| 528 | + /* test */ static Method topicNameValuesMethod() { |
| 529 | + // |
| 530 | + Method topicDescriptionMethod = null; |
| 531 | + try { |
| 532 | + // First we try to get the topicNameValues() method |
| 533 | + topicDescriptionMethod = DescribeTopicsResult.class.getMethod("topicNameValues"); |
| 534 | + } catch (NoSuchMethodException exception) { |
| 535 | + LOG.info("Failed to get method topicNameValues() from DescribeTopicsResult class since we are probably on kafka 3.0.0 or older: ", exception); |
| 536 | + } |
| 537 | + |
| 538 | + if (topicDescriptionMethod == null) { |
| 539 | + try { |
| 540 | + // Second we try to get the values() method |
| 541 | + topicDescriptionMethod = DescribeTopicsResult.class.getMethod("values"); |
| 542 | + } catch (NoSuchMethodException exception) { |
| 543 | + LOG.info("Failed to get method values() from DescribeTopicsResult class: ", exception); |
| 544 | + } |
| 545 | + } |
| 546 | + |
| 547 | + if (topicDescriptionMethod != null) { |
| 548 | + return topicDescriptionMethod; |
| 549 | + } else { |
| 550 | + throw new RuntimeException("Unable to find both values() and topicNameValues() method in the DescribeTopicsResult class "); |
| 551 | + } |
| 552 | + } |
| 553 | + |
| 554 | + /** |
| 555 | + * Retrieves the {@link TopicDescription} for the specified Kafka topic, handling compatibility |
| 556 | + * with Kafka versions 4.0 and above. This method uses reflection to invoke the appropriate method |
| 557 | + * for retrieving topic description information, depending on the Kafka version. |
| 558 | + * |
| 559 | + * @param adminClient The Kafka {@link AdminClient} used to interact with the Kafka cluster. |
| 560 | + * @param ccMetricsTopic The name of the Kafka topic for which the description is to be retrieved. |
| 561 | + * |
| 562 | + * @return The {@link TopicDescription} for the specified Kafka topic. |
| 563 | + * |
| 564 | + * @throws KafkaTopicDescriptionException If an error occurs while retrieving the topic description, |
| 565 | + * or if the topic name retrieval method cannot be found or invoked properly. This includes |
| 566 | + * exceptions related to reflection (e.g., {@link NoSuchMethodException}), invocation issues, |
| 567 | + * execution exceptions, timeouts, and interruptions. |
| 568 | + */ |
| 569 | + /* test */ static TopicDescription getTopicDescription(AdminClient adminClient, String ccMetricsTopic) throws KafkaTopicDescriptionException { |
| 570 | + try { |
| 571 | + // For compatibility with Kafka 4.0 and beyond we must use new API methods. |
| 572 | + Method topicDescriptionMethod = topicNameValuesMethod(); |
| 573 | + DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(ccMetricsTopic)); |
| 574 | + Map<String, KafkaFuture<TopicDescription>> topicDescriptionMap = (Map<String, KafkaFuture<TopicDescription>>) topicDescriptionMethod |
| 575 | + .invoke(describeTopicsResult); |
| 576 | + return topicDescriptionMap.get(ccMetricsTopic).get(CLIENT_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS); |
| 577 | + } catch (InvocationTargetException | IllegalAccessException | ExecutionException | InterruptedException | TimeoutException e) { |
| 578 | + throw new KafkaTopicDescriptionException(String.format("Unable to retrieve config of Cruise Cruise Control metrics topic {}.", |
| 579 | + ccMetricsTopic), e); |
| 580 | + } |
| 581 | + } |
514 | 582 | } |
0 commit comments