Merge pull request #3553 from v1r3n/support_domains

Support task domains in the worker spring configuration

Author: v1r3n

client-spring/src/main/java/com/netflix/conductor/client/spring/SpringWorkerConfiguration.java

@@ -35,4 +35,10 @@ public int getThreadCount(String taskName) {
         String key = "conductor.worker." + taskName + ".threadCount";
         return environment.getProperty(key, Integer.class, 0);
     }
+
+    @Override
+    public String getDomain(String taskName) {
+        String key = "conductor.worker." + taskName + ".domain";
+        return environment.getProperty(key, String.class, null);
+    }
 }

client-spring/src/test/resources/application.properties

@@ -1,2 +1,3 @@
 conductor.client.rootUri=http://localhost:8080/api/
-conductor.worker.hello.threadCount=100
+conductor.worker.hello.threadCount=100
+conductor.worker.hello_again.domain=test

java-sdk/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java

@@ -24,6 +24,7 @@
 import com.netflix.conductor.sdk.workflow.task.WorkerTask;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
 import com.google.common.reflect.ClassPath;
 
 public class AnnotatedWorkerExecutor {
@@ -42,6 +43,8 @@ public class AnnotatedWorkerExecutor {
 
     private Map<String, Integer> workerToPollingInterval = new HashMap<>();
 
+    private Map<String, String> workerDomains = new HashMap<>();
+
     private Map<String, Object> workerClassObjs = new HashMap<>();
 
     private static Set<String> scannedPackages = new HashSet<>();
@@ -163,6 +166,14 @@ private void addMethod(WorkerTask annotation, Method method, Object bean) {
         }
         workerToPollingInterval.put(name, pollingInterval);
 
+        String domain = workerConfiguration.getDomain(name);
+        if (Strings.isNullOrEmpty(domain)) {
+            domain = annotation.domain();
+        }
+        if (!Strings.isNullOrEmpty(domain)) {
+            workerDomains.put(name, domain);
+        }
+
         workerClassObjs.put(name, bean);
         workerExecutors.put(name, method);
         LOGGER.info(
@@ -187,10 +198,12 @@ public void startPolling() {
         }
 
         LOGGER.info("Starting workers with threadCount {}", workerToThreadCount);
+        LOGGER.info("Worker domains {}", workerDomains);
 
         taskRunner =
                 new TaskRunnerConfigurer.Builder(taskClient, executors)
                         .withTaskThreadCount(workerToThreadCount)
+                        .withTaskToDomain(workerDomains)
                         .build();
 
         taskRunner.init();

java-sdk/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/WorkerConfiguration.java

@@ -29,4 +29,8 @@ public int getPollingInterval(String taskName) {
     public int getThreadCount(String taskName) {
         return 0;
     }
+
+    public String getDomain(String taskName) {
+        return null;
+    }
 }

java-sdk/src/main/java/com/netflix/conductor/sdk/workflow/task/WorkerTask.java

@@ -27,4 +27,6 @@
     int threadCount() default 1;
 
     int pollingInterval() default 100;
+
+    String domain() default "";
 }