configure temporal workflow execution ttl (#9838)

Author: jrhizor

airbyte-workers/src/main/java/io/airbyte/workers/WorkerApp.java

@@ -323,6 +323,8 @@ public static void main(final String[] args) throws IOException, InterruptedExce
 
     final WorkflowServiceStubs temporalService = TemporalUtils.createTemporalService(temporalHost);
 
+    TemporalUtils.configureTemporalNamespace(temporalService);
+
     final Database configDatabase = new ConfigsDatabaseInstance(
         configs.getConfigDatabaseUser(),
         configs.getConfigDatabasePassword(),

airbyte-workers/src/main/java/io/airbyte/workers/temporal/TemporalUtils.java

@@ -10,9 +10,12 @@
 import io.airbyte.scheduler.models.JobRunConfig;
 import io.temporal.activity.Activity;
 import io.temporal.api.common.v1.WorkflowExecution;
+import io.temporal.api.namespace.v1.NamespaceConfig;
 import io.temporal.api.namespace.v1.NamespaceInfo;
+import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest;
 import io.temporal.api.workflowservice.v1.DescribeNamespaceResponse;
 import io.temporal.api.workflowservice.v1.ListNamespacesRequest;
+import io.temporal.api.workflowservice.v1.UpdateNamespaceRequest;
 import io.temporal.client.ActivityCompletionException;
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowOptions;
@@ -30,6 +33,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.time.DurationFormatUtils;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -56,6 +60,30 @@ public static WorkflowServiceStubs createTemporalService(final String temporalHo
 
   public static final String DEFAULT_NAMESPACE = "default";
 
+  private static final Duration WORKFLOW_EXECUTION_TTL = Duration.ofDays(7);
+  private static final String HUMAN_READABLE_WORKFLOW_EXECUTION_TTL =
+      DurationFormatUtils.formatDurationWords(WORKFLOW_EXECUTION_TTL.toMillis(), true, true);
+
+  public static void configureTemporalNamespace(WorkflowServiceStubs temporalService) {
+    final var client = temporalService.blockingStub();
+    final var describeNamespaceRequest = DescribeNamespaceRequest.newBuilder().setNamespace(DEFAULT_NAMESPACE).build();
+    final var currentRetentionGrpcDuration = client.describeNamespace(describeNamespaceRequest).getConfig().getWorkflowExecutionRetentionTtl();
+    final var currentRetention = Duration.ofSeconds(currentRetentionGrpcDuration.getSeconds());
+
+    if (currentRetention.equals(WORKFLOW_EXECUTION_TTL)) {
+      LOGGER.info("Workflow execution TTL already set for namespace " + DEFAULT_NAMESPACE + ". Remains unchanged as: "
+          + HUMAN_READABLE_WORKFLOW_EXECUTION_TTL);
+    } else {
+      final var newGrpcDuration = com.google.protobuf.Duration.newBuilder().setSeconds(WORKFLOW_EXECUTION_TTL.getSeconds()).build();
+      final var humanReadableCurrentRetention = DurationFormatUtils.formatDurationWords(currentRetention.toMillis(), true, true);
+      final var namespaceConfig = NamespaceConfig.newBuilder().setWorkflowExecutionRetentionTtl(newGrpcDuration).build();
+      final var updateNamespaceRequest = UpdateNamespaceRequest.newBuilder().setNamespace(DEFAULT_NAMESPACE).setConfig(namespaceConfig).build();
+      LOGGER.info("Workflow execution TTL differs for namespace " + DEFAULT_NAMESPACE + ". Changing from (" + humanReadableCurrentRetention + ") to ("
+          + HUMAN_READABLE_WORKFLOW_EXECUTION_TTL + "). ");
+      client.updateNamespace(updateNamespaceRequest);
+    }
+  }
+
   @FunctionalInterface
   public interface TemporalJobCreator<T extends Serializable> {
 

airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/shared/ActivityConfiguration.java

@@ -32,7 +32,7 @@ public class ActivityConfiguration {
       .setScheduleToStartTimeout(Duration.ofDays(MAX_SYNC_TIMEOUT_DAYS))
       .setCancellationType(ActivityCancellationType.WAIT_CANCELLATION_COMPLETED)
       .setRetryOptions(TemporalUtils.NO_RETRY)
-      .setHeartbeatTimeout(Duration.ofSeconds(30))
+      .setHeartbeatTimeout(TemporalUtils.HEARTBEAT_TIMEOUT)
       .build();
 
 }