Merge branch 'main' into upgrade-jetty

Author: JvD-Ericsson

.github/workflows/ci.yaml

@@ -62,7 +62,7 @@ jobs:
       matrix:
         java-ver: [17]
         java-dist: ['temurin']
-        hw_platform: ['s390x']
+        hw_platform: ['s390x', 'ppc64le']
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

build.gradle

@@ -189,7 +189,6 @@ project(':cruise-control-core') {
     implementation "org.apache.logging.log4j:log4j-slf4j-impl:2.17.2"
     implementation 'org.apache.commons:commons-math3:3.6.1'
     api "org.eclipse.jetty.ee10:jetty-ee10-servlet:${jettyVersion}"
-    implementation 'com.github.spotbugs:spotbugs-annotations:4.8.6'
 
     api "io.vertx:vertx-core:${vertxVersion}"
     api "io.vertx:vertx-web:${vertxVersion}"
@@ -304,7 +303,7 @@ project(':cruise-control') {
     implementation 'com.google.code.gson:gson:2.9.0'
     implementation "org.eclipse.jetty:jetty-server:${jettyVersion}"
     implementation 'io.dropwizard.metrics:metrics-jmx:4.2.9'
-    implementation 'com.nimbusds:nimbus-jose-jwt:9.45'
+    implementation 'com.nimbusds:nimbus-jose-jwt:10.0.2'
     implementation 'io.swagger.parser.v3:swagger-parser-v3:2.1.16'
     implementation 'io.github.classgraph:classgraph:4.8.141'
     implementation 'com.google.code.findbugs:jsr305:3.0.2'
@@ -340,6 +339,7 @@ project(':cruise-control') {
     testImplementation 'com.jayway.jsonpath:json-path:2.7.0'
     testImplementation 'org.powermock:powermock-module-junit4:2.0.9'
     testImplementation 'org.powermock:powermock-api-easymock:2.0.9'
+    testImplementation "org.testcontainers:kafka:$testcontainersVersion"
   }
 
   publishing {
@@ -484,7 +484,7 @@ project(':cruise-control-metrics-reporter') {
     testImplementation "org.apache.kafka:kafka-raft:$kafkaVersion"
     testImplementation "org.apache.kafka:kafka-storage:$kafkaVersion"
     testImplementation 'commons-io:commons-io:2.11.0'
-    testImplementation "org.testcontainers:kafka:1.21.3"
+    testImplementation "org.testcontainers:kafka:$testcontainersVersion"
     testOutput sourceSets.test.output
   }
 

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

@@ -4,7 +4,6 @@
 
 package com.linkedin.kafka.cruisecontrol.metricsreporter;
 
-import com.linkedin.kafka.cruisecontrol.metricsreporter.exception.KafkaTopicDescriptionException;
 import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.CruiseControlMetric;
 import com.linkedin.kafka.cruisecontrol.metricsreporter.metric.MetricSerde;
 import com.linkedin.kafka.cruisecontrol.metricsreporter.utils.CCContainerizedKraftCluster;
@@ -20,11 +19,8 @@
 import java.util.Set;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Predicate;
 import java.util.regex.Pattern;
 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.TopicDescription;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -36,7 +32,6 @@
 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.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.junit.After;
 import org.junit.Before;
@@ -45,7 +40,6 @@
 
 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.CruiseControlMetricsReporter.getTopicDescription;
 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;
@@ -206,41 +200,9 @@ public void testReportingMetrics() {
     assertEquals("Expected " + expectedMetricTypes + ", but saw " + metricTypes, expectedMetricTypes, metricTypes);
   }
 
-  private TopicDescription waitForTopicMetadata(Admin adminClient,
-                                                Duration timeout,
-                                                Predicate<TopicDescription> condition)
-    throws InterruptedException, TimeoutException {
-
-    long deadline = System.currentTimeMillis() + timeout.toMillis();
-
-    while (System.currentTimeMillis() < deadline) {
-      try {
-        TopicDescription topicDescription = getTopicDescription((AdminClient) adminClient, TOPIC);
-
-        if (condition.test(topicDescription)) {
-          return topicDescription;
-        }
-      } catch (KafkaTopicDescriptionException e) {
-        if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) {
-          throw new RuntimeException("Failed to describe topic: " + TOPIC, e);
-        }
-        // else ignore and retry
-      }
-
-      Thread.sleep(500);
-    }
-
-    throw new TimeoutException("Timeout waiting for topic metadata condition to be met: " + TOPIC);
-  }
-
   @Test
   public void testUpdatingMetricsTopicConfig() throws InterruptedException, TimeoutException {
-    Properties props = new Properties();
-    setSecurityConfigs(props, "admin");
-    props.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
-    Admin adminClient = Admin.create(props);
-
-    TopicDescription topicDescription = waitForTopicMetadata(adminClient, Duration.ofSeconds(30), td -> true);
+    TopicDescription topicDescription = _cluster.waitForTopicMetadata(TOPIC, Duration.ofSeconds(30), td -> true);
     assertEquals(1, topicDescription.partitions().size());
 
     KafkaContainer broker = _cluster.getBrokers().get(0);
@@ -261,7 +223,7 @@ public void testUpdatingMetricsTopicConfig() throws InterruptedException, Timeou
 
     // Wait for topic metadata configuration change to propagate
     int oldPartitionCount = topicDescription.partitions().size();
-    TopicDescription newTopicDescription = waitForTopicMetadata(adminClient, Duration.ofSeconds(30),
+    TopicDescription newTopicDescription = _cluster.waitForTopicMetadata(TOPIC, Duration.ofSeconds(30),
       td -> td.partitions().size() != oldPartitionCount);
 
     assertEquals(2, newTopicDescription.partitions().size());