From 88b8071cf4cdd75da6c4450d1421f1211fee4cbe Mon Sep 17 00:00:00 2001
From: DanMiller192 <55767381+danmiller192@users.noreply.github.com>
Date: Tue, 12 Nov 2024 11:10:13 +0000
Subject: [PATCH] Added maximum workflow sweep postpone config

---
 .../conductor/core/config/ConductorProperties.java   | 12 ++++++++++++
 .../core/reconciliation/WorkflowSweeper.java         | 11 ++++++++++-
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/core/src/main/java/com/netflix/conductor/core/config/ConductorProperties.java b/core/src/main/java/com/netflix/conductor/core/config/ConductorProperties.java
index f07f1485f..31e8f6ff6 100644
--- a/core/src/main/java/com/netflix/conductor/core/config/ConductorProperties.java
+++ b/core/src/main/java/com/netflix/conductor/core/config/ConductorProperties.java
@@ -42,6 +42,10 @@ public class ConductorProperties {
     @DurationUnit(ChronoUnit.SECONDS)
     private Duration workflowOffsetTimeout = Duration.ofSeconds(30);
 
+    /** The maximum timeout duration to set when a workflow is pushed to the decider queue. */
+    @DurationUnit(ChronoUnit.SECONDS)
+    private Duration maxPostponeDurationSeconds = Duration.ofSeconds(120);
+    
     /** The number of threads to use to do background sweep on active workflows. */
     private int sweeperThreadCount = Runtime.getRuntime().availableProcessors() * 2;
 
@@ -251,6 +255,14 @@ public void setWorkflowOffsetTimeout(Duration workflowOffsetTimeout) {
         this.workflowOffsetTimeout = workflowOffsetTimeout;
     }
 
+    public Duration getMaxPostponeDurationSeconds() {
+        return maxPostponeDurationSeconds;
+    }
+
+    public void setMaxPostponeDurationSeconds(Duration maxPostponeDurationSeconds) {
+        this.maxPostponeDurationSeconds = maxPostponeDurationSeconds;
+    }
+
     public int getSweeperThreadCount() {
         return sweeperThreadCount;
     }
diff --git a/core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowSweeper.java b/core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowSweeper.java
index b39d122f0..51112029f 100644
--- a/core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowSweeper.java
+++ b/core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowSweeper.java
@@ -139,8 +139,17 @@ void unack(WorkflowModel workflowModel, long workflowOffsetTimeout) {
                 } else if (taskModel.getTaskType().equals(TaskType.TASK_TYPE_HUMAN)) {
                     postponeDurationSeconds = workflowOffsetTimeout;
                 } else {
-                    postponeDurationSeconds = workflowOffsetTimeout;
+                    postponeDurationSeconds =
+                            (taskModel.getResponseTimeoutSeconds() != 0)
+                                    ? taskModel.getResponseTimeoutSeconds() + 1
+                                    : workflowOffsetTimeout;
+                }
+
+                if(postponeDurationSeconds > properties.getMaxPostponeDurationSeconds().getSeconds())
+                {
+                    postponeDurationSeconds = properties.getMaxPostponeDurationSeconds().getSeconds();
                 }
+                
                 break;
             } else if (taskModel.getStatus() == Status.SCHEDULED) {
                 Optional<TaskDef> taskDefinition = taskModel.getTaskDefinition();