Merge pull request #1493 from ozangunalp/kafka-clients-3.0.0

smallrye · Dec 1, 2021 · ed5081f · ed5081f
2 parents acd0c16 + 746ae3e
commit ed5081f
Show file tree

Hide file tree

Showing 9 changed files with 26 additions and 11 deletions.
diff --git a/.github/workflows/build-main-branches.yml b/.github/workflows/build-main-branches.yml
@@ -18,6 +18,11 @@ jobs:
                 java: [
                     {'version': '8'},
                     {'version': '11'},
+                    {
+                        'version': '11',
+                        'build_opts': '-Dkafka-container-version=latest-kafka-2.8.0 -pl smallrye-reactive-messaging-kafka --also-make',
+                        'name': 'Kafka tests against Kafka Broker 2.8.0'
+                    },
                     {
                         'version': '16',
                         'opts': '--illegal-access=permit' # required for kotlin

diff --git a/.github/workflows/build-pull.yml b/.github/workflows/build-pull.yml
@@ -16,6 +16,11 @@ jobs:
                 java: [
                     {'version': '8'},
                     {'version': '11'},
+                    {
+                        'version': '11',
+                        'build_opts': '-Dkafka-container-version=latest-kafka-2.8.0 -pl smallrye-reactive-messaging-kafka --also-make',
+                        'name': 'Kafka tests against Kafka Broker 2.8.0'
+                    },
                     {
                         'version': '16',
                         'opts': '--illegal-access=permit' # required for kotlin

diff --git a/pom.xml b/pom.xml
@@ -83,7 +83,7 @@
 
     <jboss-log-manager.version>2.1.18.Final</jboss-log-manager.version>
 
-    <kafka.version>2.8.0</kafka.version>
+    <kafka.version>3.0.0</kafka.version>
 
     <opentelemetry.version>1.9.1</opentelemetry.version>
     <opentelemetry-semver.version>1.9.1-alpha</opentelemetry-semver.version>

diff --git a/...e-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/KafkaSourceTest.java b/...e-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/KafkaSourceTest.java
@@ -219,7 +219,7 @@ public void testBroadcastWithPartitions() {
     @Tag(TestTags.SLOW)
     public void testRetry() {
         // This test need an individual Kafka container
-        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.KAFKA_VERSION);) {
+        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.getKafkaContainerVersion())) {
             kafka.start();
             await().until(kafka::isRunning);
             MapBasedConfig config = newCommonConfigForSource()

diff --git a/...src/test/java/io/smallrye/reactive/messaging/kafka/KafkaSourceWithLegacyMetadataTest.java b/...src/test/java/io/smallrye/reactive/messaging/kafka/KafkaSourceWithLegacyMetadataTest.java
@@ -222,7 +222,7 @@ public void testBroadcastWithPartitions() {
     @Tag(TestTags.SLOW)
     public void testRetry() {
         // This test need an individual Kafka container
-        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.KAFKA_VERSION)) {
+        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.getKafkaContainerVersion())) {
             kafka.start();
             await().until(kafka::isRunning);
             MapBasedConfig config = newCommonConfigForSource()

diff --git a/...g-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/base/KafkaBrokerExtension.java b/...g-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/base/KafkaBrokerExtension.java
@@ -23,7 +23,7 @@
 public class KafkaBrokerExtension implements BeforeAllCallback, ParameterResolver, CloseableResource {
     public static final Logger LOGGER = Logger.getLogger(KafkaBrokerExtension.class.getName());
 
-    public static final String KAFKA_VERSION = "latest-kafka-2.8.0";
+    public static final String KAFKA_VERSION = "latest-kafka-3.0.0";
 
     private static boolean started = false;
     static StrimziKafkaContainer kafka;
@@ -44,8 +44,13 @@ public void close() {
         stopKafkaBroker();
     }
 
+    public static String getKafkaContainerVersion() {
+        String kafkaContainerVersion = System.getProperty("kafka-container-version");
+        return kafkaContainerVersion != null ? kafkaContainerVersion : KAFKA_VERSION;
+    }
+
     public static void startKafkaBroker() {
-        kafka = new StrimziKafkaContainer(KAFKA_VERSION)
+        kafka = new StrimziKafkaContainer(getKafkaContainerVersion())
                 .withExposedPorts(9092);
         kafka.start();
         LOGGER.info("Kafka broker started: " + kafka.getBootstrapServers() + " (" + kafka.getMappedPort(9092) + ")");

diff --git a/...afka/src/test/java/io/smallrye/reactive/messaging/kafka/base/KafkaToxiproxyExtension.java b/...afka/src/test/java/io/smallrye/reactive/messaging/kafka/base/KafkaToxiproxyExtension.java
@@ -67,7 +67,7 @@ static int startKafkaProxy(Network network) {
     }
 
     static void startKafkaBroker(Network network, int proxyPort) {
-        kafka = new ProxiedStrimziKafkaContainer(KAFKA_VERSION, proxyPort)
+        kafka = new ProxiedStrimziKafkaContainer(KafkaBrokerExtension.getKafkaContainerVersion(), proxyPort)
                 .withNetwork(network)
                 .withNetworkAliases(KAFKA_NETWORK_ALIAS)
                 .withExposedPorts(KAFKA_PORT);

diff --git a/...e-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/base/KafkaUsage.java b/...e-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/base/KafkaUsage.java
@@ -116,7 +116,7 @@ private static void sleep(Duration duration) {
      */
     public static class FixedKafkaContainer extends StrimziKafkaContainer {
         public FixedKafkaContainer(int port) {
-            super(KafkaBrokerExtension.KAFKA_VERSION);
+            super(KafkaBrokerExtension.getKafkaContainerVersion());
             super.addFixedExposedPort(port, KAFKA_PORT);
         }
 

diff --git a/...ng-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/client/BrokerRestartTest.java b/...ng-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/client/BrokerRestartTest.java
@@ -46,7 +46,7 @@ public void tearDown() {
 
     @Test
     public void testAcknowledgementUsingThrottledStrategyEvenAfterBrokerRestart() throws Exception {
-        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.KAFKA_VERSION)) {
+        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.getKafkaContainerVersion())) {
             kafka.start();
             await().until(kafka::isRunning);
 
@@ -71,7 +71,7 @@ public void testAcknowledgementUsingThrottledStrategyEvenAfterBrokerRestart() th
 
     @Test
     public void testResumingPausingWhileBrokerIsDown() throws Exception {
-        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.KAFKA_VERSION)) {
+        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.getKafkaContainerVersion())) {
             kafka.start();
             await().until(kafka::isRunning);
             Integer port = kafka.getMappedPort(KAFKA_PORT);
@@ -105,7 +105,7 @@ public void testResumingPausingWhileBrokerIsDown() throws Exception {
 
     @Test
     public void testPausingWhileBrokerIsDown() throws Exception {
-        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.KAFKA_VERSION)) {
+        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.getKafkaContainerVersion())) {
             kafka.start();
             await().until(kafka::isRunning);
             Integer port = kafka.getMappedPort(KAFKA_PORT);
@@ -163,7 +163,7 @@ public void testPausingWhileBrokerIsDown() throws Exception {
     @Test
     public void testWithBrokerRestart() throws Exception {
         int sendBatchSize = 10;
-        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.KAFKA_VERSION)) {
+        try (StrimziKafkaContainer kafka = new StrimziKafkaContainer(KafkaBrokerExtension.getKafkaContainerVersion())) {
             kafka.start();
             String groupId = UUID.randomUUID().toString();
             MapBasedConfig config = createConsumerConfig(groupId)