feat(restart-inbound): give the possibility to the developer to plan a restart of the connector in case of failure

connector-runtime/connector-runtime-core/src/main/java/io/camunda/connector/runtime/core/error/RestartException.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH
+ * under one or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership. Camunda licenses this file to you under the Apache License,
+ * Version 2.0; you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.camunda.connector.runtime.core.error;
+
+import java.time.Duration;
+
+public class RestartException extends RuntimeException {
+
+  private final Duration backoffTime;
+
+  public RestartException(String message, Throwable cause, Duration backoffTime) {
+    super(message, cause);
+    this.backoffTime = backoffTime;
+  }
+
+  public Duration getBackoffTime() {
+    return backoffTime;
+  }
+}

connector-runtime/connector-runtime-core/src/main/java/io/camunda/connector/runtime/core/inbound/DefaultInboundConnectorContextFactory.java

@@ -18,6 +18,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.EvictingQueue;
+import io.camunda.connector.api.inbound.Activity;
 import io.camunda.connector.api.inbound.InboundConnectorContext;
 import io.camunda.connector.api.inbound.InboundConnectorExecutable;
 import io.camunda.connector.api.inbound.InboundIntermediateConnectorContext;
@@ -28,6 +29,7 @@
 import io.camunda.document.factory.DocumentFactory;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.time.Duration;
 import java.util.function.Consumer;
 
 public class DefaultInboundConnectorContextFactory implements InboundConnectorContextFactory {
@@ -57,8 +59,9 @@ public DefaultInboundConnectorContextFactory(
   public <T extends InboundConnectorExecutable<?>> InboundConnectorContext createContext(
       final ValidInboundConnectorDetails connectorDetails,
       final Consumer<Throwable> cancellationCallback,
+      final Consumer<Duration> reactivationCallback,
       final Class<T> executableClass,
-      final EvictingQueue queue) {
+      final EvictingQueue<Activity> queue) {
 
     InboundConnectorReportingContext inboundContext =
         new InboundConnectorContextImpl(
@@ -68,6 +71,7 @@ public <T extends InboundConnectorExecutable<?>> InboundConnectorContext createC
             connectorDetails,
             correlationHandler,
             cancellationCallback,
+            reactivationCallback,
             objectMapper,
             queue);
 

connector-runtime/connector-runtime-core/src/main/java/io/camunda/connector/runtime/core/inbound/InboundConnectorContextFactory.java

@@ -17,9 +17,11 @@
 package io.camunda.connector.runtime.core.inbound;
 
 import com.google.common.collect.EvictingQueue;
+import io.camunda.connector.api.inbound.Activity;
 import io.camunda.connector.api.inbound.InboundConnectorContext;
 import io.camunda.connector.api.inbound.InboundConnectorExecutable;
 import io.camunda.connector.runtime.core.inbound.details.InboundConnectorDetails.ValidInboundConnectorDetails;
+import java.time.Duration;
 import java.util.function.Consumer;
 
 /**
@@ -50,6 +52,7 @@ public interface InboundConnectorContextFactory {
   <T extends InboundConnectorExecutable<?>> InboundConnectorContext createContext(
       final ValidInboundConnectorDetails connectorDetails,
       final Consumer<Throwable> cancellationCallback,
+      final Consumer<Duration> reactivationCallback,
       final Class<T> executableClass,
-      final EvictingQueue queue);
+      final EvictingQueue<Activity> queue);
 }

connector-runtime/connector-runtime-core/src/main/java/io/camunda/connector/runtime/core/inbound/InboundConnectorContextImpl.java

@@ -28,6 +28,7 @@
 import io.camunda.connector.api.validation.ValidationProvider;
 import io.camunda.connector.feel.FeelEngineWrapperException;
 import io.camunda.connector.runtime.core.AbstractConnectorContext;
+import io.camunda.connector.runtime.core.error.RestartException;
 import io.camunda.connector.runtime.core.inbound.correlation.InboundCorrelationHandler;
 import io.camunda.connector.runtime.core.inbound.details.InboundConnectorDetails;
 import io.camunda.connector.runtime.core.inbound.details.InboundConnectorDetails.ValidInboundConnectorDetails;
@@ -36,6 +37,7 @@
 import io.camunda.document.factory.DocumentFactoryImpl;
 import io.camunda.document.store.DocumentCreationRequest;
 import io.camunda.document.store.InMemoryDocumentStore;
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -56,6 +58,7 @@ public class InboundConnectorContextImpl extends AbstractConnectorContext
   private final ObjectMapper objectMapper;
 
   private final Consumer<Throwable> cancellationCallback;
+  private final Consumer<Duration> reactivationCallback;
   private final EvictingQueue<Activity> logs;
   private final DocumentFactory documentFactory;
   private Health health = Health.unknown();
@@ -68,8 +71,9 @@ public InboundConnectorContextImpl(
       ValidInboundConnectorDetails connectorDetails,
       InboundCorrelationHandler correlationHandler,
       Consumer<Throwable> cancellationCallback,
+      Consumer<Duration> reactivationCallback,
       ObjectMapper objectMapper,
-      EvictingQueue logs) {
+      EvictingQueue<Activity> logs) {
     super(secretProvider, validationProvider);
     this.documentFactory = documentFactory;
     this.correlationHandler = correlationHandler;
@@ -79,6 +83,7 @@ public InboundConnectorContextImpl(
             connectorDetails.rawPropertiesWithoutKeywords());
     this.objectMapper = objectMapper;
     this.cancellationCallback = cancellationCallback;
+    this.reactivationCallback = reactivationCallback;
     this.logs = logs;
   }
 
@@ -88,6 +93,7 @@ public InboundConnectorContextImpl(
       ValidInboundConnectorDetails connectorDetails,
       InboundCorrelationHandler correlationHandler,
       Consumer<Throwable> cancellationCallback,
+      Consumer<Duration> reactivationCallback,
       ObjectMapper objectMapper,
       EvictingQueue logs) {
     this(
@@ -97,6 +103,7 @@ public InboundConnectorContextImpl(
         connectorDetails,
         correlationHandler,
         cancellationCallback,
+        reactivationCallback,
         objectMapper,
         logs);
   }
@@ -125,6 +132,9 @@ public void cancel(Throwable exception) {
     } catch (Throwable e) {
       LOG.error("Failed to deliver the cancellation signal to the runtime", e);
     }
+    if (Objects.requireNonNull(exception) instanceof RestartException restartException) {
+      reactivationCallback.accept(restartException.getBackoffTime());
+    }
   }
 
   @Override

connector-runtime/connector-runtime-core/src/test/java/io/camunda/connector/runtime/core/inbound/DefaultInboundConnectorContextFactoryTest.java

@@ -28,6 +28,7 @@
 import io.camunda.connector.runtime.core.inbound.details.InboundConnectorDetails.ValidInboundConnectorDetails;
 import io.camunda.connector.runtime.core.secret.SecretProviderAggregator;
 import io.camunda.document.factory.DocumentFactory;
+import java.time.Duration;
 import java.util.function.Consumer;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -44,6 +45,7 @@ class DefaultInboundConnectorContextFactoryTest {
   @Mock private ValidationProvider validationProvider;
   @Mock private OperateClientAdapter operateClientAdapter;
   @Mock private Consumer<Throwable> cancellationCallback;
+  @Mock private Consumer<Duration> reactivationCallback;
   @Mock private ValidInboundConnectorDetails newConnector;
   @Mock private DocumentFactory documentFactory;
   private DefaultInboundConnectorContextFactory factory;
@@ -66,6 +68,7 @@ void shouldCreateInboundConnectorContext() {
         factory.createContext(
             newConnector,
             cancellationCallback,
+            reactivationCallback,
             ExecutableWithInboundContext.class,
             EvictingQueue.create(10));
 
@@ -78,6 +81,7 @@ void shouldCreateInboundConnectorContextWhenParameterizedTypeIsEmpty() {
         factory.createContext(
             newConnector,
             cancellationCallback,
+            reactivationCallback,
             ExecutableWithEmptyParameterizedType.class,
             EvictingQueue.create(10));
 
@@ -91,6 +95,7 @@ void shouldCreateInboundIntermediateConnectorContext() {
         factory.createContext(
             newConnector,
             cancellationCallback,
+            reactivationCallback,
             ExecutableWithIntermediate.class,
             EvictingQueue.create(10));
 

connector-runtime/connector-runtime-core/src/test/java/io/camunda/connector/runtime/core/inbound/InboundConnectorContextImplTest.java

@@ -41,6 +41,21 @@ class InboundConnectorContextImplTest {
   private final SecretProvider secretProvider = new FooBarSecretProvider();
   private final ObjectMapper mapper = ConnectorsObjectMapperSupplier.DEFAULT_MAPPER;
 
+  @NotNull
+  private static ValidInboundConnectorDetails getInboundConnectorDefinition(
+      Map<String, String> properties) {
+    properties = new HashMap<>(properties);
+    properties.put("inbound.type", "io.camunda:connector:1");
+    InboundConnectorElement element =
+        new InboundConnectorElement(
+            properties,
+            new StandaloneMessageCorrelationPoint("", "", null, null),
+            new ProcessElement("bool", 0, 0, "id", "<default>"));
+    var details = InboundConnectorDetails.of(element.deduplicationId(List.of()), List.of(element));
+    assertThat(details).isInstanceOf(ValidInboundConnectorDetails.class);
+    return (ValidInboundConnectorDetails) details;
+  }
+
   @Test
   void bindProperties_shouldThrowExceptionWhenWrongFormat() {
     // given
@@ -52,6 +67,7 @@ void bindProperties_shouldThrowExceptionWhenWrongFormat() {
             definition,
             null,
             (e) -> {},
+            (d) -> {},
             mapper,
             EvictingQueue.create(10));
     // when and then
@@ -73,6 +89,7 @@ void bindProperties_shouldParseNullValue() {
             definition,
             null,
             (e) -> {},
+            (d) -> {},
             mapper,
             EvictingQueue.create(10));
     // when
@@ -98,6 +115,7 @@ void bindProperties_shouldParseStringAsString() {
             definition,
             null,
             (e) -> {},
+            (d) -> {},
             mapper,
             EvictingQueue.create(10));
     // when
@@ -108,21 +126,6 @@ void bindProperties_shouldParseStringAsString() {
         .isInstanceOf(String.class);
   }
 
-  @NotNull
-  private static ValidInboundConnectorDetails getInboundConnectorDefinition(
-      Map<String, String> properties) {
-    properties = new HashMap<>(properties);
-    properties.put("inbound.type", "io.camunda:connector:1");
-    InboundConnectorElement element =
-        new InboundConnectorElement(
-            properties,
-            new StandaloneMessageCorrelationPoint("", "", null, null),
-            new ProcessElement("bool", 0, 0, "id", "<default>"));
-    var details = InboundConnectorDetails.of(element.deduplicationId(List.of()), List.of(element));
-    assertThat(details).isInstanceOf(ValidInboundConnectorDetails.class);
-    return (ValidInboundConnectorDetails) details;
-  }
-
   @Test
   void bindProperties_shouldParseAllObject() {
     // Given
@@ -156,6 +159,7 @@ void bindProperties_shouldParseAllObject() {
             definition,
             null,
             (e) -> {},
+            (d) -> {},
             mapper,
             EvictingQueue.create(10));
     // when
@@ -177,6 +181,7 @@ void getProperties_shouldNotParseFeel() {
             definition,
             null,
             (e) -> {},
+            (d) -> {},
             mapper,
             EvictingQueue.create(10));
 
@@ -265,8 +270,6 @@ public void setBool(final boolean bool) {
       this.bool = bool;
     }
 
-    public record InnerObject(List<String> stringList, boolean bool) {}
-
     @Override
     public boolean equals(final Object o) {
       if (this == o) return true;
@@ -325,5 +328,7 @@ public String toString() {
           + bool
           + "}";
     }
+
+    public record InnerObject(List<String> stringList, boolean bool) {}
   }
 }

connector-runtime/connector-runtime-spring/src/main/java/io/camunda/connector/runtime/inbound/executable/BatchExecutableProcessor.java

@@ -18,6 +18,7 @@
 
 import com.google.common.collect.EvictingQueue;
 import io.camunda.connector.api.inbound.Health;
+import io.camunda.connector.api.inbound.InboundConnectorContext;
 import io.camunda.connector.api.inbound.InboundConnectorExecutable;
 import io.camunda.connector.api.inbound.webhook.WebhookConnectorExecutable;
 import io.camunda.connector.runtime.core.inbound.InboundConnectorContextFactory;
@@ -33,14 +34,15 @@
 import io.camunda.connector.runtime.inbound.webhook.WebhookConnectorRegistry;
 import io.camunda.connector.runtime.metrics.ConnectorMetrics.Inbound;
 import io.camunda.zeebe.spring.client.metrics.MetricsRecorder;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.UUID;
-import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -49,15 +51,14 @@
 public class BatchExecutableProcessor {
 
   private static final Logger LOG = LoggerFactory.getLogger(BatchExecutableProcessor.class);
-
-  @Value("${camunda.connector.inbound.log.size:10}")
-  private int inboundLogsSize;
-
   private final InboundConnectorFactory connectorFactory;
   private final InboundConnectorContextFactory connectorContextFactory;
   private final MetricsRecorder metricsRecorder;
   private final WebhookConnectorRegistry webhookConnectorRegistry;
 
+  @Value("${camunda.connector.inbound.log.size:10}")
+  private int inboundLogsSize;
+
   public BatchExecutableProcessor(
       InboundConnectorFactory connectorFactory,
       InboundConnectorContextFactory connectorContextFactory,
@@ -76,7 +77,8 @@ public BatchExecutableProcessor(
    */
   public Map<UUID, RegisteredExecutable> activateBatch(
       Map<UUID, InboundConnectorDetails> request,
-      BiConsumer<Throwable, UUID> cancellationCallback) {
+      Function<UUID, Consumer<Throwable>> cancellationCallbackMaker,
+      Function<UUID, Consumer<Duration>> reactivationCallbackMaker) {
 
     final Map<UUID, RegisteredExecutable> alreadyActivated = new HashMap<>();
 
@@ -94,7 +96,8 @@ public Map<UUID, RegisteredExecutable> activateBatch(
       }
 
       final RegisteredExecutable result =
-          activateSingle(data, e -> cancellationCallback.accept(e, id));
+          activateSingle(
+              data, cancellationCallbackMaker.apply(id), reactivationCallbackMaker.apply(id));
 
       switch (result) {
         case Activated activated -> alreadyActivated.put(id, activated);
@@ -138,14 +141,16 @@ public Map<UUID, RegisteredExecutable> activateBatch(
   }
 
   private RegisteredExecutable activateSingle(
-      InboundConnectorDetails data, Consumer<Throwable> cancellationCallback) {
+      InboundConnectorDetails data,
+      Consumer<Throwable> cancellationCallback,
+      Consumer<Duration> reactivationCallback) {
 
     if (data instanceof InvalidInboundConnectorDetails invalid) {
       return new InvalidDefinition(invalid, invalid.error().getMessage());
     }
     var validData = (ValidInboundConnectorDetails) data;
 
-    final InboundConnectorExecutable executable;
+    final InboundConnectorExecutable<InboundConnectorContext> executable;
     final InboundConnectorReportingContext context;
 
     try {
@@ -155,6 +160,7 @@ private RegisteredExecutable activateSingle(
               connectorContextFactory.createContext(
                   validData,
                   cancellationCallback,
+                  reactivationCallback,
                   executable.getClass(),
                   EvictingQueue.create(inboundLogsSize));
     } catch (NoSuchElementException e) {