Introduce bulkhead anf circuitbreaker config on workflow task level

conductor-oss · Feb 5, 2025 · f57c6d2 · f57c6d2
1 parent 94cd17a
commit f57c6d2
Show file tree

Hide file tree

Showing 2 changed files with 179 additions and 0 deletions.
diff --git a/common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowTask.java b/common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowTask.java
@@ -188,6 +188,97 @@ public void setTasks(List<WorkflowTask> tasks) {
     @ProtoField(id = 32)
     private boolean permissive;
 
+    @ProtoField(id = 33)
+    private CircuitBreakerConfig circuitBreakerConfig;
+
+    @ProtoField(id = 34)
+    private BulkheadConfig bulkheadConfig;
+
+    @ProtoMessage
+    public static class CircuitBreakerConfig {
+        @ProtoField(id = 1)
+        private int failureRateThreshold;
+
+        @ProtoField(id = 2)
+        private int slidingWindowSize;
+
+        @ProtoField(id = 3)
+        private int minimumNumberOfCalls;
+
+        @ProtoField(id = 4)
+        private int waitDurationInOpenState;
+
+        @ProtoField(id = 5)
+        private int permittedNumberOfCallsInHalfOpenState;
+
+        // Getters and setters
+        public int getFailureRateThreshold() {
+            return failureRateThreshold;
+        }
+
+        public void setFailureRateThreshold(int failureRateThreshold) {
+            this.failureRateThreshold = failureRateThreshold;
+        }
+
+        public int getSlidingWindowSize() {
+            return slidingWindowSize;
+        }
+
+        public void setSlidingWindowSize(int slidingWindowSize) {
+            this.slidingWindowSize = slidingWindowSize;
+        }
+
+        public int getMinimumNumberOfCalls() {
+            return minimumNumberOfCalls;
+        }
+
+        public void setMinimumNumberOfCalls(int minimumNumberOfCalls) {
+            this.minimumNumberOfCalls = minimumNumberOfCalls;
+        }
+
+        public int getWaitDurationInOpenState() {
+            return waitDurationInOpenState;
+        }
+
+        public void setWaitDurationInOpenState(int waitDurationInOpenState) {
+            this.waitDurationInOpenState = waitDurationInOpenState;
+        }
+
+        public int getPermittedNumberOfCallsInHalfOpenState() {
+            return permittedNumberOfCallsInHalfOpenState;
+        }
+
+        public void setPermittedNumberOfCallsInHalfOpenState(int permittedNumberOfCallsInHalfOpenState) {
+            this.permittedNumberOfCallsInHalfOpenState = permittedNumberOfCallsInHalfOpenState;
+        }
+    }
+
+    @ProtoMessage
+    public static class BulkheadConfig {
+        @ProtoField(id = 1)
+        private int maxConcurrentCalls;
+
+        @ProtoField(id = 2)
+        private int maxWaitDuration;
+
+        // Getters and setters
+        public int getMaxConcurrentCalls() {
+            return maxConcurrentCalls;
+        }
+
+        public void setMaxConcurrentCalls(int maxConcurrentCalls) {
+            this.maxConcurrentCalls = maxConcurrentCalls;
+        }
+
+        public int getMaxWaitDuration() {
+            return maxWaitDuration;
+        }
+
+        public void setMaxWaitDuration(int maxWaitDuration) {
+            this.maxWaitDuration = maxWaitDuration;
+        }
+    }
+
     /**
      * @return the name
      */
@@ -346,6 +437,22 @@ public void setDynamicTaskNameParam(String dynamicTaskNameParam) {
         this.dynamicTaskNameParam = dynamicTaskNameParam;
     }
 
+    public CircuitBreakerConfig getCircuitBreakerConfig() {
+        return circuitBreakerConfig;
+    }
+
+    public void setCircuitBreakerConfig(CircuitBreakerConfig circuitBreakerConfig) {
+        this.circuitBreakerConfig = circuitBreakerConfig;
+    }
+
+    public BulkheadConfig getBulkheadConfig() {
+        return bulkheadConfig;
+    }
+
+    public void setBulkheadConfig(BulkheadConfig bulkheadConfig) {
+        this.bulkheadConfig = bulkheadConfig;
+    }
+
     /**
      * @deprecated Use {@link WorkflowTask#getEvaluatorType()} and {@link
      *     WorkflowTask#getExpression()} combination.

diff --git a/common/src/test/java/com/netflix/conductor/common/workflow/WorkflowTaskTest.java b/common/src/test/java/com/netflix/conductor/common/workflow/WorkflowTaskTest.java
@@ -29,6 +29,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 public class WorkflowTaskTest {
@@ -76,4 +77,75 @@ public void testWorkflowTaskName() {
                 validationErrors.contains(
                         "WorkflowTask taskReferenceName name cannot be empty or null"));
     }
+
+    @Test
+    public void testCircuitBreakerConfig() {
+        WorkflowTask task = new WorkflowTask();
+        assertNull(task.getCircuitBreakerConfig());
+
+        WorkflowTask.CircuitBreakerConfig config = new WorkflowTask.CircuitBreakerConfig();
+        config.setFailureRateThreshold(50);
+        config.setSlidingWindowSize(100);
+        config.setMinimumNumberOfCalls(10);
+        config.setWaitDurationInOpenState(60);
+        config.setPermittedNumberOfCallsInHalfOpenState(5);
+
+        task.setCircuitBreakerConfig(config);
+
+        assertNotNull(task.getCircuitBreakerConfig());
+        assertEquals(50, task.getCircuitBreakerConfig().getFailureRateThreshold());
+        assertEquals(100, task.getCircuitBreakerConfig().getSlidingWindowSize());
+        assertEquals(10, task.getCircuitBreakerConfig().getMinimumNumberOfCalls());
+        assertEquals(60, task.getCircuitBreakerConfig().getWaitDurationInOpenState());
+        assertEquals(5, task.getCircuitBreakerConfig().getPermittedNumberOfCallsInHalfOpenState());
+    }
+
+    @Test
+    public void testBulkheadConfig() {
+        WorkflowTask task = new WorkflowTask();
+        assertNull(task.getBulkheadConfig());
+
+        WorkflowTask.BulkheadConfig config = new WorkflowTask.BulkheadConfig();
+        config.setMaxConcurrentCalls(20);
+        config.setMaxWaitDuration(1000);
+
+        task.setBulkheadConfig(config);
+
+        assertNotNull(task.getBulkheadConfig());
+        assertEquals(20, task.getBulkheadConfig().getMaxConcurrentCalls());
+        assertEquals(1000, task.getBulkheadConfig().getMaxWaitDuration());
+    }
+
+    @Test
+    public void testCircuitBreakerConfigBuilder() {
+        WorkflowTask.CircuitBreakerConfig config = new WorkflowTask.CircuitBreakerConfig();
+
+        // Test individual setters
+        config.setFailureRateThreshold(75);
+        assertEquals(75, config.getFailureRateThreshold());
+
+        config.setSlidingWindowSize(200);
+        assertEquals(200, config.getSlidingWindowSize());
+
+        config.setMinimumNumberOfCalls(20);
+        assertEquals(20, config.getMinimumNumberOfCalls());
+
+        config.setWaitDurationInOpenState(30);
+        assertEquals(30, config.getWaitDurationInOpenState());
+
+        config.setPermittedNumberOfCallsInHalfOpenState(3);
+        assertEquals(3, config.getPermittedNumberOfCallsInHalfOpenState());
+    }
+
+    @Test
+    public void testBulkheadConfigBuilder() {
+        WorkflowTask.BulkheadConfig config = new WorkflowTask.BulkheadConfig();
+
+        // Test individual setters
+        config.setMaxConcurrentCalls(50);
+        assertEquals(50, config.getMaxConcurrentCalls());
+
+        config.setMaxWaitDuration(2000);
+        assertEquals(2000, config.getMaxWaitDuration());
+    }
 }