Kafka connector loop retry (#2792)

* fix(kafka): retry the main consumer loop with 30 seconds backoff

* reduce test wait duration

* lint

* switch to failsafe

* set number of retries to -1 (unlimited)

* use constant for retry number

* try fixing the integration test

* try fixing the integration test

* try fixing the integration test

* try fixing the integration test

* try fixing the integration test

* formatting

(cherry picked from commit 807071e)

connectors/kafka/pom.xml

@@ -62,10 +62,16 @@ except in compliance with the proprietary license.</license.inlineheader>
       <artifactId>commons-text</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>dev.failsafe</groupId>
+      <artifactId>failsafe</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>org.testcontainers</groupId>
       <artifactId>kafka</artifactId>
       <version>${version.testcontainers}</version>
+      <scope>test</scope>
     </dependency>
 
     <dependency>

connectors/kafka/src/main/java/io/camunda/connector/kafka/inbound/KafkaConnectorConsumer.java

@@ -20,6 +20,9 @@
 import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import com.fasterxml.jackson.module.scala.DefaultScalaModule$;
+import dev.failsafe.Failsafe;
+import dev.failsafe.RetryPolicy;
+import dev.failsafe.function.CheckedSupplier;
 import io.camunda.connector.api.inbound.Health;
 import io.camunda.connector.api.inbound.InboundConnectorContext;
 import java.time.Duration;
@@ -36,12 +39,26 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.OffsetOutOfRangeException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class KafkaConnectorConsumer {
   private static final Logger LOG = LoggerFactory.getLogger(KafkaConnectorConsumer.class);
+
+  private final InboundConnectorContext context;
+
+  private final ExecutorService executorService;
+
+  public CompletableFuture<?> future;
+
+  Consumer<Object, Object> consumer;
+
+  KafkaConnectorProperties elementProps;
+
+  private Health consumerStatus = Health.unknown();
+
+  private final RetryPolicy<Object> retryPolicy;
+
   public static ObjectMapper objectMapper =
       new ObjectMapper()
           .registerModule(new Jdk8Module())
@@ -52,24 +69,20 @@ public class KafkaConnectorConsumer {
           .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
           .enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES)
           .enable(JsonReadFeature.ALLOW_UNESCAPED_CONTROL_CHARS.mappedFeature());
-  private final InboundConnectorContext context;
-  private final ExecutorService executorService;
   private final Function<Properties, Consumer<Object, Object>> consumerCreatorFunction;
-  public CompletableFuture<?> future;
-  Consumer<Object, Object> consumer;
-  KafkaConnectorProperties elementProps;
   boolean shouldLoop = true;
-  private Health consumerStatus = Health.up();
   private ObjectReader avroObjectReader;
 
   public KafkaConnectorConsumer(
       final Function<Properties, Consumer<Object, Object>> consumerCreatorFunction,
       final InboundConnectorContext connectorContext,
-      final KafkaConnectorProperties elementProps) {
+      final KafkaConnectorProperties elementProps,
+      final RetryPolicy<Object> retryPolicy) {
     this.consumerCreatorFunction = consumerCreatorFunction;
     this.context = connectorContext;
     this.elementProps = elementProps;
     this.executorService = Executors.newSingleThreadExecutor();
+    this.retryPolicy = retryPolicy;
   }
 
   public void startConsumer() {
@@ -80,13 +93,28 @@ public void startConsumer() {
       AvroMapper avroMapper = new AvroMapper();
       avroObjectReader = avroMapper.reader(avroSchema);
     }
-    this.future =
-        CompletableFuture.runAsync(
-            () -> {
-              prepareConsumer();
-              consume();
-            },
-            this.executorService);
+
+    CheckedSupplier<Void> retryableFutureSupplier =
+        () -> {
+          try {
+            prepareConsumer();
+            consume();
+            return null;
+          } catch (Exception ex) {
+            LOG.error("Consumer loop failure, retry pending: {}", ex.getMessage());
+            throw ex;
+          }
+        };
+
+    future =
+        Failsafe.with(retryPolicy)
+            .with(executorService)
+            .getAsync(retryableFutureSupplier)
+            .exceptionally(
+                (e) -> {
+                  shouldLoop = false;
+                  return null;
+                });
   }
 
   private void prepareConsumer() {
@@ -111,9 +139,7 @@ public void consume() {
         reportUp();
       } catch (Exception ex) {
         reportDown(ex);
-        if (ex instanceof OffsetOutOfRangeException) {
-          throw ex;
-        }
+        throw ex;
       }
     }
     LOG.debug("Kafka inbound loop finished");

connectors/kafka/src/main/java/io/camunda/connector/kafka/inbound/KafkaExecutable.java

@@ -6,10 +6,12 @@
  */
 package io.camunda.connector.kafka.inbound;
 
+import dev.failsafe.RetryPolicy;
 import io.camunda.connector.api.annotation.InboundConnector;
 import io.camunda.connector.api.inbound.Health;
 import io.camunda.connector.api.inbound.InboundConnectorContext;
 import io.camunda.connector.api.inbound.InboundConnectorExecutable;
+import java.time.Duration;
 import java.util.Properties;
 import java.util.function.Function;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -23,13 +25,25 @@ public class KafkaExecutable implements InboundConnectorExecutable {
   private final Function<Properties, Consumer<Object, Object>> consumerCreatorFunction;
   public KafkaConnectorConsumer kafkaConnectorConsumer;
 
+  private final RetryPolicy<Object> retryPolicy;
+
   public KafkaExecutable(
-      final Function<Properties, Consumer<Object, Object>> consumerCreatorFunction) {
+      final Function<Properties, Consumer<Object, Object>> consumerCreatorFunction,
+      final RetryPolicy<Object> retryConfig) {
     this.consumerCreatorFunction = consumerCreatorFunction;
+    this.retryPolicy = retryConfig;
   }
 
+  private static final int INFINITE_RETRIES = -1;
+
   public KafkaExecutable() {
-    this(KafkaConsumer::new);
+    this(
+        KafkaConsumer::new,
+        RetryPolicy.builder()
+            .handle(Exception.class)
+            .withDelay(Duration.ofSeconds(30))
+            .withMaxAttempts(INFINITE_RETRIES)
+            .build());
   }
 
   @Override
@@ -38,7 +52,8 @@ public void activate(InboundConnectorContext connectorContext) {
       KafkaConnectorProperties elementProps =
           connectorContext.bindProperties(KafkaConnectorProperties.class);
       this.kafkaConnectorConsumer =
-          new KafkaConnectorConsumer(consumerCreatorFunction, connectorContext, elementProps);
+          new KafkaConnectorConsumer(
+              consumerCreatorFunction, connectorContext, elementProps, retryPolicy);
       this.kafkaConnectorConsumer.startConsumer();
     } catch (Exception ex) {
       connectorContext.reportHealth(Health.down(ex));

connectors/kafka/src/test/java/io/camunda/connector/kafka/inbound/KafkaExecutableTest.java

@@ -19,14 +19,17 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import dev.failsafe.RetryPolicy;
 import io.camunda.connector.kafka.outbound.model.KafkaTopic;
 import io.camunda.connector.test.inbound.InboundConnectorContextBuilder;
 import io.camunda.connector.test.inbound.InboundConnectorDefinitionBuilder;
 import io.camunda.connector.validation.impl.DefaultValidationProvider;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
@@ -49,7 +52,6 @@
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.ArgumentCaptor;
-import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 @ExtendWith(MockitoExtension.class)
@@ -59,8 +61,6 @@ public class KafkaExecutableTest {
   private InboundConnectorContextBuilder.TestInboundConnectorContext originalContext;
   private List<PartitionInfo> topicPartitions;
   private KafkaConnectorProperties kafkaConnectorProperties;
-  @Mock private KafkaConsumer<Object, Object> mockConsumer;
-  private String topic;
 
   private static Stream<Arguments> provideStringsForGetOffsets() {
     return Stream.of(
@@ -73,7 +73,15 @@ private static Stream<Arguments> provideStringsForGetOffsets() {
         Arguments.of(Arrays.asList(10L, 12L), Arrays.asList(10L, 12L)));
   }
 
+  private KafkaTopic kafkaTopic;
+  private KafkaConsumer<Object, Object> mockConsumer;
+
+  private String topic;
+
+  private static final int MAX_ATTEMPTS = 3;
+
   @BeforeEach
+  @SuppressWarnings("unchecked")
   public void setUp() {
     topic = "my-topic";
     topicPartitions =
@@ -97,6 +105,7 @@ public void setUp() {
             .validation(new DefaultValidationProvider())
             .build();
     originalContext = context;
+    mockConsumer = mock(KafkaConsumer.class);
   }
 
   @Test
@@ -148,6 +157,29 @@ void testActivateAndDeactivate() {
     assertFalse(kafkaExecutable.kafkaConnectorConsumer.shouldLoop);
   }
 
+  @Test
+  void testActivateAndDeactivate_consumerThrows() {
+    // Given
+    KafkaExecutable kafkaExecutable = getConsumerMock();
+    var groupMetadataMock = mock(ConsumerGroupMetadata.class);
+    when(groupMetadataMock.groupId()).thenReturn("groupId");
+    when(groupMetadataMock.groupInstanceId()).thenReturn(Optional.of("groupInstanceId"));
+    when(groupMetadataMock.generationId()).thenReturn(1);
+    when(mockConsumer.groupMetadata()).thenReturn(groupMetadataMock);
+
+    // When
+    when(mockConsumer.poll(any())).thenThrow(new RuntimeException("Test exception"));
+    kafkaExecutable.activate(context);
+    await()
+        .atMost(Duration.ofSeconds(5))
+        .pollInterval(Duration.ofMillis(500))
+        .untilAsserted(() -> assertFalse(kafkaExecutable.kafkaConnectorConsumer.shouldLoop));
+    kafkaExecutable.deactivate();
+
+    // Then
+    verify(mockConsumer, times(MAX_ATTEMPTS)).poll(any(Duration.class));
+  }
+
   @Test
   void testGetKafkaProperties() {
     // When
@@ -236,7 +268,13 @@ public void testConvertDoubleEscapedCharactersRecordToKafkaInboundMessage() {
   }
 
   public KafkaExecutable getConsumerMock() {
-    return new KafkaExecutable(properties -> mockConsumer);
+    return new KafkaExecutable(
+        properties -> mockConsumer,
+        RetryPolicy.builder()
+            .handle(Exception.class)
+            .withDelay(Duration.ofMillis(50))
+            .withMaxAttempts(MAX_ATTEMPTS)
+            .build());
   }
 
   @ParameterizedTest

parent/pom.xml

@@ -82,6 +82,7 @@ limitations under the License.</license.inlineheader>
     <version.jackson-datatype-jsr310>2.17.2</version.jackson-datatype-jsr310>
     <version.hibernate-validator>8.0.1.Final</version.hibernate-validator>
     <version.jsonassert>1.5.3</version.jsonassert>
+    <version.failsafe>3.3.2</version.failsafe>
 
     <version.spring-boot>3.3.0</version.spring-boot>
     <version.spring-cloud-gcp-starter-logging>5.3.0</version.spring-cloud-gcp-starter-logging>
@@ -452,6 +453,12 @@ limitations under the License.</license.inlineheader>
         <version>${version.bouncycastle}</version>
       </dependency>
 
+      <dependency>
+        <groupId>dev.failsafe</groupId>
+        <artifactId>failsafe</artifactId>
+        <version>${version.failsafe}</version>
+      </dependency>
+
       <!-- test dependencies -->
       <dependency>
         <groupId>org.junit.jupiter</groupId>