Fix Snappy tests in native and refactor them (#1851)

quarkus-qe · Jun 25, 2024 · 6df539e · 6df539e
1 parent f8f619f
commit 6df539e
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 12 deletions.
diff --git a/...a-producer/src/test/java/io/quarkus/ts/messaging/kafka/producer/DevModeKafkaSnappyIT.java b/...a-producer/src/test/java/io/quarkus/ts/messaging/kafka/producer/DevModeKafkaSnappyIT.java
@@ -1,11 +1,13 @@
 package io.quarkus.ts.messaging.kafka.producer;
 
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import io.quarkus.test.bootstrap.RestService;
 import io.quarkus.test.scenarios.QuarkusScenario;
 import io.quarkus.test.services.DevModeQuarkusApplication;
 
+@Disabled("Needs refactoring to test Snappy in a DEV mode") // TODO: refactor to test snappy a DEV mode, not prod mode
 @QuarkusScenario
 public class DevModeKafkaSnappyIT extends SnappyCompressionIT {
 

diff --git a/...producer/src/test/java/io/quarkus/ts/messaging/kafka/producer/OpenShiftKafkaSnappyIT.java b/...producer/src/test/java/io/quarkus/ts/messaging/kafka/producer/OpenShiftKafkaSnappyIT.java
@@ -12,7 +12,6 @@
 import io.quarkus.test.bootstrap.Protocol;
 import io.quarkus.test.bootstrap.RestService;
 import io.quarkus.test.scenarios.OpenShiftScenario;
-import io.quarkus.test.scenarios.annotations.DisabledOnNative;
 import io.quarkus.test.services.KafkaContainer;
 import io.quarkus.test.services.QuarkusApplication;
 import io.quarkus.test.services.containers.model.KafkaVendor;
@@ -21,7 +20,6 @@
 import io.vertx.mutiny.core.buffer.Buffer;
 
 @OpenShiftScenario
-@DisabledOnNative(reason = "In native mode, Snappy is disabled by default as the use of Snappy requires embedding a native library and unpacking it when the application starts.")
 public class OpenShiftKafkaSnappyIT {
 
     private static final int TIMEOUT_SEC = 5;
@@ -31,7 +29,9 @@ public class OpenShiftKafkaSnappyIT {
     static final KafkaService kafka = new KafkaService().withProperty("auto.create.topics.enable", "false");
 
     @QuarkusApplication
-    static RestService app = new RestService().withProperty("kafka.bootstrap.servers", kafka::getBootstrapUrl);
+    static RestService app = new RestService()
+            .withProperty("quarkus.kafka.snappy.enabled", "true")
+            .withProperty("kafka.bootstrap.servers", kafka::getBootstrapUrl);
 
     @Test
     public void checkCompressCodecSnappy() throws IOException, InterruptedException {

diff --git a/...ka-producer/src/test/java/io/quarkus/ts/messaging/kafka/producer/SnappyCompressionIT.java b/...ka-producer/src/test/java/io/quarkus/ts/messaging/kafka/producer/SnappyCompressionIT.java
@@ -44,7 +44,9 @@ public class SnappyCompressionIT {
     static final KafkaService kafka = new KafkaService().withProperty("auto.create.topics.enable", "false");
 
     @QuarkusApplication
-    static RestService app = new RestService().withProperty("kafka.bootstrap.servers", kafka::getBootstrapUrl);
+    static RestService app = new RestService()
+            .withProperty("quarkus.kafka.snappy.enabled", "true")
+            .withProperty("kafka.bootstrap.servers", kafka::getBootstrapUrl);
 
     @Test
     public void checkCompressCodecSnappy() throws IOException, InterruptedException {
@@ -78,14 +80,15 @@ public void checkSnappyCompressionSize() throws IOException, InterruptedExceptio
     @Test
     public void checkIntegrityMessageAfterCompression() {
         String msg = "This is the message";
-        KafkaConsumer<Integer, String> consumer = createConsumer();
-        UniAssertSubscriber<Object> subscriber = makeHttpReqWithMessage("/messageEvent", msg)
-                .subscribe().withSubscriber(UniAssertSubscriber.create());
-        subscriber.awaitItem(Duration.ofSeconds(TIMEOUT_SEC)).getItem();
-
-        ConsumerRecord<Integer, String> records = consumer.poll(Duration.ofMillis(10000))
-                .iterator().next();
-        Assertions.assertEquals(records.value(), msg);
+        try (var consumer = createConsumer()) {
+            UniAssertSubscriber<Object> subscriber = makeHttpReqWithMessage("/messageEvent", msg)
+                    .subscribe().withSubscriber(UniAssertSubscriber.create());
+            subscriber.awaitItem(Duration.ofSeconds(TIMEOUT_SEC)).getItem();
+
+            ConsumerRecord<Integer, String> records = consumer.poll(Duration.ofMillis(10000))
+                    .iterator().next();
+            Assertions.assertEquals(records.value(), msg);
+        }
     }
 
     @Test