Merge pull request #535 from conductor-oss/wf-cache-config

Added cache config for workflow

Author: v1r3n

common/src/main/java/com/netflix/conductor/common/metadata/workflow/CacheConfig.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2025 Conductor Authors.
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.netflix.conductor.common.metadata.workflow;
+
+import com.netflix.conductor.annotations.protogen.ProtoField;
+import com.netflix.conductor.annotations.protogen.ProtoMessage;
+
+@ProtoMessage
+public class CacheConfig {
+
+    @ProtoField(id = 1)
+    private String key;
+
+    @ProtoField(id = 2)
+    private int ttlInSecond;
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public int getTtlInSecond() {
+        return ttlInSecond;
+    }
+
+    public void setTtlInSecond(int ttlInSecond) {
+        this.ttlInSecond = ttlInSecond;
+    }
+}

common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowDef.java

@@ -109,6 +109,9 @@ public enum TimeoutPolicy {
     @ProtoField(id = 22)
     private Map<String, Object> metadata = new HashMap<>();
 
+    @ProtoField(id = 23)
+    private CacheConfig cacheConfig;
+
     public boolean isEnforceSchema() {
         return enforceSchema;
     }
@@ -377,6 +380,14 @@ public void setMetadata(Map<String, Object> metadata) {
         this.metadata = metadata;
     }
 
+    public CacheConfig getCacheConfig() {
+        return cacheConfig;
+    }
+
+    public void setCacheConfig(final CacheConfig cacheConfig) {
+        this.cacheConfig = cacheConfig;
+    }
+
     public boolean containsType(String taskType) {
         return collectTasks().stream().anyMatch(t -> t.getType().equals(taskType));
     }

common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowTask.java

@@ -38,32 +38,6 @@
 @ProtoMessage
 public class WorkflowTask {
 
-    @ProtoMessage
-    public static class CacheConfig {
-
-        @ProtoField(id = 1)
-        private String key;
-
-        @ProtoField(id = 2)
-        private int ttlInSecond;
-
-        public String getKey() {
-            return key;
-        }
-
-        public void setKey(String key) {
-            this.key = key;
-        }
-
-        public int getTtlInSecond() {
-            return ttlInSecond;
-        }
-
-        public void setTtlInSecond(int ttlInSecond) {
-            this.ttlInSecond = ttlInSecond;
-        }
-    }
-
     @ProtoField(id = 1)
     @NotEmpty(message = "WorkflowTask name cannot be empty or null")
     private String name;

grpc/src/main/java/com/netflix/conductor/grpc/AbstractProtoMapper.java

@@ -10,6 +10,7 @@
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
 import com.netflix.conductor.common.metadata.tasks.TaskExecLog;
 import com.netflix.conductor.common.metadata.tasks.TaskResult;
+import com.netflix.conductor.common.metadata.workflow.CacheConfig;
 import com.netflix.conductor.common.metadata.workflow.DynamicForkJoinTask;
 import com.netflix.conductor.common.metadata.workflow.DynamicForkJoinTaskList;
 import com.netflix.conductor.common.metadata.workflow.RateLimitConfig;
@@ -25,6 +26,7 @@
 import com.netflix.conductor.common.run.TaskSummary;
 import com.netflix.conductor.common.run.Workflow;
 import com.netflix.conductor.common.run.WorkflowSummary;
+import com.netflix.conductor.proto.CacheConfigPb;
 import com.netflix.conductor.proto.DynamicForkJoinTaskListPb;
 import com.netflix.conductor.proto.DynamicForkJoinTaskPb;
 import com.netflix.conductor.proto.EventExecutionPb;
@@ -61,6 +63,22 @@
 
 @Generated("com.netflix.conductor.annotationsprocessor.protogen")
 public abstract class AbstractProtoMapper {
+    public CacheConfigPb.CacheConfig toProto(CacheConfig from) {
+        CacheConfigPb.CacheConfig.Builder to = CacheConfigPb.CacheConfig.newBuilder();
+        if (from.getKey() != null) {
+            to.setKey( from.getKey() );
+        }
+        to.setTtlInSecond( from.getTtlInSecond() );
+        return to.build();
+    }
+
+    public CacheConfig fromProto(CacheConfigPb.CacheConfig from) {
+        CacheConfig to = new CacheConfig();
+        to.setKey( from.getKey() );
+        to.setTtlInSecond( from.getTtlInSecond() );
+        return to;
+    }
+
     public DynamicForkJoinTaskPb.DynamicForkJoinTask toProto(DynamicForkJoinTask from) {
         DynamicForkJoinTaskPb.DynamicForkJoinTask.Builder to = DynamicForkJoinTaskPb.DynamicForkJoinTask.newBuilder();
         if (from.getTaskName() != null) {
@@ -1342,6 +1360,9 @@ public WorkflowDefPb.WorkflowDef toProto(WorkflowDef from) {
         for (Map.Entry<String, Object> pair : from.getMetadata().entrySet()) {
             to.putMetadata( pair.getKey(), toProto( pair.getValue() ) );
         }
+        if (from.getCacheConfig() != null) {
+            to.setCacheConfig( toProto( from.getCacheConfig() ) );
+        }
         return to.build();
     }
 
@@ -1390,6 +1411,9 @@ public WorkflowDef fromProto(WorkflowDefPb.WorkflowDef from) {
             metadataMap.put( pair.getKey(), fromProto( pair.getValue() ) );
         }
         to.setMetadata(metadataMap);
+        if (from.hasCacheConfig()) {
+            to.setCacheConfig( fromProto( from.getCacheConfig() ) );
+        }
         return to;
     }
 
@@ -1651,22 +1675,6 @@ public WorkflowTask fromProto(WorkflowTaskPb.WorkflowTask from) {
         return to;
     }
 
-    public WorkflowTaskPb.WorkflowTask.CacheConfig toProto(WorkflowTask.CacheConfig from) {
-        WorkflowTaskPb.WorkflowTask.CacheConfig.Builder to = WorkflowTaskPb.WorkflowTask.CacheConfig.newBuilder();
-        if (from.getKey() != null) {
-            to.setKey( from.getKey() );
-        }
-        to.setTtlInSecond( from.getTtlInSecond() );
-        return to.build();
-    }
-
-    public WorkflowTask.CacheConfig fromProto(WorkflowTaskPb.WorkflowTask.CacheConfig from) {
-        WorkflowTask.CacheConfig to = new WorkflowTask.CacheConfig();
-        to.setKey( from.getKey() );
-        to.setTtlInSecond( from.getTtlInSecond() );
-        return to;
-    }
-
     public abstract WorkflowTaskPb.WorkflowTask.WorkflowTaskList toProto(List<WorkflowTask> in);
 
     public abstract List<WorkflowTask> fromProto(WorkflowTaskPb.WorkflowTask.WorkflowTaskList in);

grpc/src/main/proto/model/cacheconfig.proto

@@ -0,0 +1,12 @@
+syntax = "proto3";
+package conductor.proto;
+
+
+option java_package = "com.netflix.conductor.proto";
+option java_outer_classname = "CacheConfigPb";
+option go_package = "github.com/netflix/conductor/client/gogrpc/conductor/model";
+
+message CacheConfig {
+    string key = 1;
+    int32 ttl_in_second = 2;
+}

grpc/src/main/proto/model/workflowdef.proto

@@ -5,6 +5,7 @@ import "model/ratelimitconfig.proto";
 import "model/workflowtask.proto";
 import "google/protobuf/struct.proto";
 import "model/schemadef.proto";
+import "model/cacheconfig.proto";
 
 option java_package = "com.netflix.conductor.proto";
 option java_outer_classname = "WorkflowDefPb";
@@ -36,4 +37,5 @@ message WorkflowDef {
     SchemaDef output_schema = 20;
     bool enforce_schema = 21;
     map<string, google.protobuf.Value> metadata = 22;
+    CacheConfig cache_config = 23;
 }

grpc/src/main/proto/model/workflowtask.proto

@@ -4,16 +4,13 @@ package conductor.proto;
 import "model/taskdef.proto";
 import "model/subworkflowparams.proto";
 import "google/protobuf/struct.proto";
+import "model/cacheconfig.proto";
 
 option java_package = "com.netflix.conductor.proto";
 option java_outer_classname = "WorkflowTaskPb";
 option go_package = "github.com/netflix/conductor/client/gogrpc/conductor/model";
 
 message WorkflowTask {
-    message CacheConfig {
-        string key = 1;
-        int32 ttl_in_second = 2;
-    }
     message WorkflowTaskList {
         repeated WorkflowTask tasks = 1;
     }
@@ -46,6 +43,6 @@ message WorkflowTask {
     string evaluator_type = 27;
     string expression = 28;
     string join_status = 30;
-    WorkflowTask.CacheConfig cache_config = 31;
+    CacheConfig cache_config = 31;
     bool permissive = 32;
 }