build: fix kafka connector integration test (#2898)

camunda · Jul 23, 2024 · 305a07c · 305a07c
1 parent 591c67c
commit 305a07c
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 8 deletions.
diff --git a/...ectors/kafka/src/main/java/io/camunda/connector/kafka/inbound/KafkaConnectorConsumer.java b/...ectors/kafka/src/main/java/io/camunda/connector/kafka/inbound/KafkaConnectorConsumer.java
@@ -126,7 +126,8 @@ public void startConsumer() {
             .exceptionally(
                 (e) -> {
                   shouldLoop = false;
-                  return null;
+                  throw new RuntimeException(
+                      "Consumer loop failed, retries exhausted: " + e.getMessage(), e);
                 });
   }
 

diff --git a/...tors/kafka/src/test/java/io/camunda/connector/kafka/integration/KafkaIntegrationTest.java b/...tors/kafka/src/test/java/io/camunda/connector/kafka/integration/KafkaIntegrationTest.java
@@ -14,6 +14,7 @@
 import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import dev.failsafe.RetryPolicy;
 import io.camunda.connector.api.outbound.OutboundConnectorContext;
 import io.camunda.connector.api.outbound.OutboundConnectorFunction;
 import io.camunda.connector.kafka.inbound.KafkaConnectorConsumer;
@@ -43,6 +44,7 @@
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.NewPartitions;
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetOutOfRangeException;
 import org.junit.ClassRule;
 import org.junit.jupiter.api.AfterAll;
@@ -97,7 +99,8 @@ private static void createTopics(String... topics) {
             .collect(Collectors.toList());
     try (var admin = AdminClient.create(Map.of(BOOTSTRAP_SERVERS_CONFIG, getKafkaBrokers()))) {
       admin.createTopics(newTopics);
-      admin.createPartitions(Map.of(TOPIC, NewPartitions.increaseTo(2)));
+      Arrays.stream(topics)
+          .forEach(topic -> admin.createPartitions(Map.of(topic, NewPartitions.increaseTo(2))));
     }
   }
 
@@ -197,12 +200,13 @@ void setInvalidOffsetForInboundConnectorWhenAutoOffsetResetIsNone() {
             .properties(kafkaConnectorProperties)
             .definition(InboundConnectorDefinitionBuilder.create().build())
             .build();
-    KafkaExecutable executable = new KafkaExecutable();
+    KafkaExecutable executable =
+        new KafkaExecutable(KafkaConsumer::new, RetryPolicy.builder().withMaxAttempts(1).build());
 
     // When
-    OffsetOutOfRangeException thrown =
+    Exception thrown =
         assertThrows(
-            OffsetOutOfRangeException.class,
+            Exception.class,
             () -> {
               try {
                 executable.activate(context);
@@ -213,6 +217,8 @@ void setInvalidOffsetForInboundConnectorWhenAutoOffsetResetIsNone() {
             },
             "OffsetOutOfRangeException was expected");
 
+    assertThat(thrown.getCause()).isInstanceOf(OffsetOutOfRangeException.class);
+
     // Then we except exception with message
     assertThat(thrown.getMessage()).contains("Fetch position FetchPosition");
     assertThat(thrown.getMessage()).contains("is out of range for partition " + TOPIC + "-");
@@ -448,7 +454,7 @@ void consumeMessageWithHeaders() {
             kafkaTopic,
             null,
             null,
-            List.of(0L, 0L),
+            null,
             KafkaConnectorProperties.AutoOffsetReset.EARLIEST,
             null);
 
@@ -469,8 +475,8 @@ void consumeMessageWithHeaders() {
         context.getCorrelations().stream()
             .filter(
                 m -> {
-                  var messge = (KafkaInboundMessage) m;
-                  return !(messge.getValue() instanceof String);
+                  var msg = (KafkaInboundMessage) m;
+                  return !(msg.getValue() instanceof String);
                 })
             .findFirst()
             .orElse(null);