Kafka container version is overridable from system property

smallrye · Nov 30, 2021 · 00f9ff3 · 00f9ff3
1 parent 5d95ccd
commit 00f9ff3
Show file tree

Hide file tree

Showing 6 changed files with 14 additions and 9 deletions.
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
@@ -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)