wip: Auto extend job timeout

spring-boot-starter-camunda/src/main/java/io/camunda/zeebe/spring/client/configuration/ZeebeClientAllAutoConfiguration.java

@@ -53,8 +53,10 @@ public CommandExceptionHandlingStrategy commandExceptionHandlingStrategy(
   public JobWorkerManager jobWorkerManager(
       final CommandExceptionHandlingStrategy commandExceptionHandlingStrategy,
       final JsonMapper jsonMapper,
-      final MetricsRecorder metricsRecorder) {
-    return new JobWorkerManager(commandExceptionHandlingStrategy, jsonMapper, metricsRecorder);
+      final MetricsRecorder metricsRecorder,
+      final ZeebeClientExecutorService zeebeClientExecutorService) {
+    return new JobWorkerManager(
+        commandExceptionHandlingStrategy, jsonMapper, metricsRecorder, zeebeClientExecutorService);
   }
 
   @Bean

spring-boot-starter-camunda/src/main/java/io/camunda/zeebe/spring/client/properties/PropertyBasedZeebeWorkerValueCustomizer.java

@@ -41,6 +41,7 @@ public void customize(ZeebeWorkerValue zeebeWorker) {
     applyDefaultWorkerName(zeebeWorker);
     applyDefaultJobWorkerType(zeebeWorker);
     applyFetchVariables(zeebeWorker);
+    applyAutoExtendTimeout(zeebeWorker);
     applyOverrides(zeebeWorker);
   }
 
@@ -155,4 +156,15 @@ private void applyDefaultJobWorkerType(ZeebeWorkerValue zeebeWorker) {
       }
     }
   }
+
+  private void applyAutoExtendTimeout(ZeebeWorkerValue workerValue) {
+    if (!workerValue.isAutoExtendTimeout()) {
+      workerValue.setAutoExtendTimeout(
+          zeebeClientConfigurationProperties.getJob().isAutoExtendTimeout());
+    }
+    if (workerValue.getExtendTimeoutPeriod() == null) {
+      workerValue.setExtendTimeoutPeriod(
+          zeebeClientConfigurationProperties.getJob().getExtendTimeoutPeriod());
+    }
+  }
 }

spring-boot-starter-camunda/src/main/java/io/camunda/zeebe/spring/client/properties/ZeebeClientConfigurationProperties.java

@@ -550,13 +550,26 @@ public int hashCode() {
 
   public static class Job {
 
-    @Override
-    public String toString() {
-      return "Job{" + "timeout=" + timeout + ", pollInterval=" + pollInterval + '}';
-    }
-
     private Duration timeout = DEFAULT.getDefaultJobTimeout();
     private Duration pollInterval = DEFAULT.getDefaultJobPollInterval();
+    private boolean autoExtendTimeout;
+    private Duration extendTimeoutPeriod = Duration.ofSeconds(30);
+
+    public Duration getExtendTimeoutPeriod() {
+      return extendTimeoutPeriod;
+    }
+
+    public void setExtendTimeoutPeriod(Duration extendTimeoutPeriod) {
+      this.extendTimeoutPeriod = extendTimeoutPeriod;
+    }
+
+    public boolean isAutoExtendTimeout() {
+      return autoExtendTimeout;
+    }
+
+    public void setAutoExtendTimeout(boolean autoExtendTimeout) {
+      this.autoExtendTimeout = autoExtendTimeout;
+    }
 
     public Duration getTimeout() {
       return timeout;
@@ -574,17 +587,34 @@ public void setPollInterval(Duration pollInterval) {
       this.pollInterval = pollInterval;
     }
 
+    @Override
+    public String toString() {
+      return "Job{"
+          + "timeout="
+          + timeout
+          + ", pollInterval="
+          + pollInterval
+          + ", autoExtendTimeout="
+          + autoExtendTimeout
+          + ", extendTimeoutPeriod="
+          + extendTimeoutPeriod
+          + '}';
+    }
+
     @Override
     public boolean equals(Object o) {
       if (this == o) return true;
       if (o == null || getClass() != o.getClass()) return false;
       Job job = (Job) o;
-      return Objects.equals(timeout, job.timeout) && Objects.equals(pollInterval, job.pollInterval);
+      return autoExtendTimeout == job.autoExtendTimeout
+          && Objects.equals(timeout, job.timeout)
+          && Objects.equals(pollInterval, job.pollInterval)
+          && Objects.equals(extendTimeoutPeriod, job.extendTimeoutPeriod);
     }
 
     @Override
     public int hashCode() {
-      return Objects.hash(timeout, pollInterval);
+      return Objects.hash(timeout, pollInterval, autoExtendTimeout, extendTimeoutPeriod);
     }
   }
 

spring-boot-starter-camunda/src/test/java/io/camunda/zeebe/spring/client/properties/PropertyBasedZeebeWorkerValueCustomizerTest.java

@@ -9,6 +9,7 @@
 import io.camunda.zeebe.spring.client.annotation.value.ZeebeWorkerValue;
 import io.camunda.zeebe.spring.client.bean.ClassInfo;
 import io.camunda.zeebe.spring.client.bean.MethodInfo;
+import java.time.Duration;
 import java.util.Arrays;
 import org.junit.jupiter.api.Test;
 
@@ -175,6 +176,23 @@ void shouldApplyOverrides() {
     assertThat(zeebeWorkerValue.getEnabled()).isFalse();
   }
 
+  @Test
+  void shouldApplyAutoExtendTimeout() {
+    // given
+    ZeebeClientConfigurationProperties properties = properties();
+    properties.getJob().setAutoExtendTimeout(true);
+    properties.getJob().setExtendTimeoutPeriod(Duration.ofSeconds(25));
+    PropertyBasedZeebeWorkerValueCustomizer customizer =
+        new PropertyBasedZeebeWorkerValueCustomizer(properties);
+    ZeebeWorkerValue zeebeWorkerValue = new ZeebeWorkerValue();
+    zeebeWorkerValue.setMethodInfo(methodInfo(this, "testBean", "sampleWorker"));
+    // when
+    customizer.customize(zeebeWorkerValue);
+    // then
+    assertThat(zeebeWorkerValue.getExtendTimeoutPeriod()).isEqualTo(Duration.ofSeconds(25));
+    assertThat(zeebeWorkerValue.isAutoExtendTimeout()).isEqualTo(true);
+  }
+
   private static class ComplexProcessVariable {
     private String var3;
     private String var4;

spring-client-annotations/src/main/java/io/camunda/zeebe/spring/client/annotation/JobWorker.java

@@ -81,5 +81,9 @@ String name() default
 
   boolean enabled() default true;
 
+  boolean autoExtendTimeout() default false;
+
+  long extendTimeoutPeriodSeconds() default -1;
+
   String[] tenantIds() default {};
 }

spring-client-annotations/src/main/java/io/camunda/zeebe/spring/client/annotation/ZeebeWorker.java

@@ -55,4 +55,8 @@ String name() default
   boolean enabled() default true;
 
   String[] tenantIds() default {};
+
+  boolean autoExtendTimeout() default false;
+
+  long extendTimeoutPeriodSeconds() default -1;
 }

spring-client-zeebe/src/main/java/io/camunda/zeebe/spring/client/annotation/processor/ZeebeWorkerAnnotationProcessor.java

@@ -11,6 +11,7 @@
 import io.camunda.zeebe.spring.client.bean.MethodInfo;
 import io.camunda.zeebe.spring.client.jobhandling.JobWorkerManager;
 import java.lang.invoke.MethodHandles;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -85,7 +86,9 @@ public Optional<ZeebeWorkerValue> readJobWorkerAnnotationForMethod(final MethodI
               annotation.enabled(),
               methodInfo,
               Arrays.asList(annotation.tenantIds()),
-              annotation.fetchAllVariables()));
+              annotation.fetchAllVariables(),
+              annotation.autoExtendTimeout(),
+              Duration.ofSeconds(annotation.extendTimeoutPeriodSeconds())));
     } else {
       Optional<ZeebeWorker> legacyAnnotation = methodInfo.getAnnotation(ZeebeWorker.class);
       if (legacyAnnotation.isPresent()) {
@@ -103,7 +106,9 @@ public Optional<ZeebeWorkerValue> readJobWorkerAnnotationForMethod(final MethodI
                 annotation.enabled(),
                 methodInfo,
                 Arrays.asList(annotation.tenantIds()),
-                annotation.forceFetchAllVariables()));
+                annotation.forceFetchAllVariables(),
+                annotation.autoExtendTimeout(),
+                Duration.ofSeconds(annotation.extendTimeoutPeriodSeconds())));
       }
     }
     return Optional.empty();

spring-client-zeebe/src/main/java/io/camunda/zeebe/spring/client/annotation/value/ZeebeWorkerValue.java

@@ -1,6 +1,7 @@
 package io.camunda.zeebe.spring.client.annotation.value;
 
 import io.camunda.zeebe.spring.client.bean.MethodInfo;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
@@ -28,6 +29,8 @@ public class ZeebeWorkerValue implements ZeebeAnnotationValue<MethodInfo> {
   private MethodInfo methodInfo;
   private List<String> tenantIds;
   private boolean forceFetchAllVariables;
+  private boolean autoExtendTimeout;
+  private Duration extendTimeoutPeriod;
 
   public ZeebeWorkerValue() {}
 
@@ -43,7 +46,9 @@ public ZeebeWorkerValue(
       Boolean enabled,
       MethodInfo methodInfo,
       List<String> tenantIds,
-      boolean forceFetchAllVariables) {
+      boolean forceFetchAllVariables,
+      boolean autoExtendTimeout,
+      Duration extendTimeoutPeriod) {
     this.type = type;
     this.name = name;
     this.timeout = timeout;
@@ -56,6 +61,24 @@ public ZeebeWorkerValue(
     this.methodInfo = methodInfo;
     this.tenantIds = tenantIds;
     this.forceFetchAllVariables = forceFetchAllVariables;
+    this.autoExtendTimeout = autoExtendTimeout;
+    this.extendTimeoutPeriod = extendTimeoutPeriod;
+  }
+
+  public Duration getExtendTimeoutPeriod() {
+    return extendTimeoutPeriod;
+  }
+
+  public void setExtendTimeoutPeriod(Duration extendTimeoutPeriod) {
+    this.extendTimeoutPeriod = extendTimeoutPeriod;
+  }
+
+  public boolean isAutoExtendTimeout() {
+    return autoExtendTimeout;
+  }
+
+  public void setAutoExtendTimeout(boolean autoExtendTimeout) {
+    this.autoExtendTimeout = autoExtendTimeout;
   }
 
   public String getType() {
@@ -188,6 +211,10 @@ public String toString() {
         + tenantIds
         + ", forceFetchAllVariables="
         + forceFetchAllVariables
+        + ", autoExtendTimeout="
+        + autoExtendTimeout
+        + ", extendTimeoutPeriod="
+        + extendTimeoutPeriod
         + '}';
   }
 
@@ -197,6 +224,7 @@ public boolean equals(Object o) {
     if (o == null || getClass() != o.getClass()) return false;
     ZeebeWorkerValue that = (ZeebeWorkerValue) o;
     return forceFetchAllVariables == that.forceFetchAllVariables
+        && autoExtendTimeout == that.autoExtendTimeout
         && Objects.equals(type, that.type)
         && Objects.equals(name, that.name)
         && Objects.equals(timeout, that.timeout)
@@ -207,7 +235,8 @@ public boolean equals(Object o) {
         && Arrays.equals(fetchVariables, that.fetchVariables)
         && Objects.equals(enabled, that.enabled)
         && Objects.equals(methodInfo, that.methodInfo)
-        && Objects.equals(tenantIds, that.tenantIds);
+        && Objects.equals(tenantIds, that.tenantIds)
+        && Objects.equals(extendTimeoutPeriod, that.extendTimeoutPeriod);
   }
 
   @Override
@@ -224,7 +253,9 @@ public int hashCode() {
             enabled,
             methodInfo,
             tenantIds,
-            forceFetchAllVariables);
+            forceFetchAllVariables,
+            autoExtendTimeout,
+            extendTimeoutPeriod);
     result = 31 * result + Arrays.hashCode(fetchVariables);
     return result;
   }

spring-client-zeebe/src/main/java/io/camunda/zeebe/spring/client/jobhandling/AutoExtendTimeoutManager.java

@@ -0,0 +1,35 @@
+package io.camunda.zeebe.spring.client.jobhandling;
+
+import io.camunda.zeebe.client.ZeebeClient;
+import java.time.Duration;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AutoExtendTimeoutManager {
+  private static final Logger LOG = LoggerFactory.getLogger(AutoExtendTimeoutManager.class);
+  private final ZeebeClientExecutorService zeebeClientExecutorService;
+  private final ZeebeClient zeebeClient;
+
+  public AutoExtendTimeoutManager(
+      ZeebeClientExecutorService zeebeClientExecutorService, ZeebeClient zeebeClient) {
+    this.zeebeClientExecutorService = zeebeClientExecutorService;
+    this.zeebeClient = zeebeClient;
+  }
+
+  public ScheduledFuture<?> startAutoExtendTimeout(long deadline, long jobKey, Duration period) {
+    return zeebeClientExecutorService
+        .get()
+        .scheduleAtFixedRate(
+            () -> extendTimeout(jobKey, period),
+            Integer.max(0, (int) (deadline - System.currentTimeMillis() - 5000)),
+            period.minusSeconds(5).getSeconds(),
+            TimeUnit.SECONDS);
+  }
+
+  private void extendTimeout(long jobKey, Duration timeout) {
+    LOG.trace("Updating job timeout of job {} by {}", jobKey, timeout.toString());
+    zeebeClient.newUpdateTimeoutCommand(jobKey).timeout(timeout).send().join();
+  }
+}