This repository was archived by the owner on Dec 13, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 2.3k
Add spring support for annotated workers #3492
Merged
+535
−46
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
50 changes: 50 additions & 0 deletions
50
...g/src/main/java/com/netflix/conductor/client/spring/ConductorWorkerAutoConfiguration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| /* | ||
| * Copyright 2023 Netflix, Inc. | ||
| * <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.client.spring; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.context.ApplicationContext; | ||
| import org.springframework.context.ApplicationListener; | ||
| import org.springframework.context.event.ContextRefreshedEvent; | ||
| import org.springframework.core.env.Environment; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| import com.netflix.conductor.client.http.TaskClient; | ||
| import com.netflix.conductor.sdk.workflow.executor.task.AnnotatedWorkerExecutor; | ||
| import com.netflix.conductor.sdk.workflow.executor.task.WorkerConfiguration; | ||
|
|
||
| @Component | ||
| public class ConductorWorkerAutoConfiguration | ||
| implements ApplicationListener<ContextRefreshedEvent> { | ||
|
|
||
| @Autowired private TaskClient taskClient; | ||
|
|
||
| @Override | ||
| public void onApplicationEvent(ContextRefreshedEvent refreshedEvent) { | ||
| ApplicationContext applicationContext = refreshedEvent.getApplicationContext(); | ||
| Environment environment = applicationContext.getEnvironment(); | ||
| WorkerConfiguration configuration = new SpringWorkerConfiguration(environment); | ||
| AnnotatedWorkerExecutor annotatedWorkerExecutor = | ||
| new AnnotatedWorkerExecutor(taskClient, configuration); | ||
|
|
||
| Map<String, Object> beans = applicationContext.getBeansWithAnnotation(Component.class); | ||
| beans.values() | ||
| .forEach( | ||
| bean -> { | ||
| annotatedWorkerExecutor.addBean(bean); | ||
| }); | ||
| annotatedWorkerExecutor.startPolling(); | ||
| } | ||
| } | ||
38 changes: 38 additions & 0 deletions
38
...t-spring/src/main/java/com/netflix/conductor/client/spring/SpringWorkerConfiguration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| /* | ||
| * Copyright 2023 Netflix, Inc. | ||
| * <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.client.spring; | ||
|
|
||
| import org.springframework.core.env.Environment; | ||
|
|
||
| import com.netflix.conductor.sdk.workflow.executor.task.WorkerConfiguration; | ||
|
|
||
| public class SpringWorkerConfiguration extends WorkerConfiguration { | ||
|
|
||
| private final Environment environment; | ||
|
|
||
| public SpringWorkerConfiguration(Environment environment) { | ||
| this.environment = environment; | ||
| } | ||
|
|
||
| @Override | ||
| public int getPollingInterval(String taskName) { | ||
| String key = "conductor.worker." + taskName + ".pollingInterval"; | ||
| return environment.getProperty(key, Integer.class, 0); | ||
| } | ||
|
|
||
| @Override | ||
| public int getThreadCount(String taskName) { | ||
| String key = "conductor.worker." + taskName + ".threadCount"; | ||
| return environment.getProperty(key, Integer.class, 0); | ||
| } | ||
| } |
49 changes: 49 additions & 0 deletions
49
client-spring/src/test/java/com/netflix/conductor/client/spring/Workers.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| /* | ||
| * Copyright 2023 Netflix, Inc. | ||
| * <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.client.spring; | ||
|
|
||
| import java.util.Date; | ||
|
|
||
| import org.springframework.stereotype.Component; | ||
|
|
||
| import com.netflix.conductor.sdk.workflow.executor.task.TaskContext; | ||
| import com.netflix.conductor.sdk.workflow.task.InputParam; | ||
| import com.netflix.conductor.sdk.workflow.task.WorkerTask; | ||
|
|
||
| @Component | ||
| public class Workers { | ||
|
|
||
| @WorkerTask(value = "hello", threadCount = 3) | ||
| public String helloWorld(@InputParam("name") String name) { | ||
| TaskContext context = TaskContext.get(); | ||
| System.out.println(new Date() + ":: Poll count: " + context.getPollCount()); | ||
| if (context.getPollCount() < 5) { | ||
| context.addLog("Not ready yet, poll count is only " + context.getPollCount()); | ||
| context.setCallbackAfter(1); | ||
| } | ||
|
|
||
| return "Hello, " + name; | ||
| } | ||
|
|
||
| @WorkerTask(value = "hello_again", pollingInterval = 333) | ||
| public String helloAgain(@InputParam("name") String name) { | ||
| TaskContext context = TaskContext.get(); | ||
| System.out.println(new Date() + ":: Poll count: " + context.getPollCount()); | ||
| if (context.getPollCount() < 5) { | ||
| context.addLog("Not ready yet, poll count is only " + context.getPollCount()); | ||
| context.setCallbackAfter(1); | ||
| } | ||
|
|
||
| return "Hello (again), " + name; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,2 @@ | ||
| conductor.client.rootUri=http://localhost:8080/api/ | ||
| conductor.worker.hello.threadCount=100 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
32 changes: 32 additions & 0 deletions
32
...src/main/java/com/netflix/conductor/sdk/workflow/executor/task/NonRetryableException.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| * Copyright 2023 Netflix, Inc. | ||
| * <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.sdk.workflow.executor.task; | ||
|
|
||
| /** | ||
| * Runtime exception indicating the non-retriable error with the task execution. If thrown, the task | ||
| * will fail with FAILED_WITH_TERMINAL_ERROR and will not kick off retries. | ||
| */ | ||
| public class NonRetryableException extends RuntimeException { | ||
|
|
||
| public NonRetryableException(String message) { | ||
| super(message); | ||
| } | ||
|
|
||
| public NonRetryableException(String message, Throwable cause) { | ||
| super(message, cause); | ||
| } | ||
|
|
||
| public NonRetryableException(Throwable cause) { | ||
| super(cause); | ||
| } | ||
| } |
79 changes: 79 additions & 0 deletions
79
java-sdk/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/TaskContext.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| /* | ||
| * Copyright 2023 Netflix, Inc. | ||
| * <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.sdk.workflow.executor.task; | ||
|
|
||
| import com.netflix.conductor.common.metadata.tasks.Task; | ||
| import com.netflix.conductor.common.metadata.tasks.TaskResult; | ||
|
|
||
| /** Context for the task */ | ||
| public class TaskContext { | ||
|
|
||
| public static final ThreadLocal<TaskContext> TASK_CONTEXT_INHERITABLE_THREAD_LOCAL = | ||
| InheritableThreadLocal.withInitial(() -> null); | ||
|
|
||
| public TaskContext(Task task, TaskResult taskResult) { | ||
| this.task = task; | ||
| this.taskResult = taskResult; | ||
| } | ||
|
|
||
| public static TaskContext get() { | ||
| return TASK_CONTEXT_INHERITABLE_THREAD_LOCAL.get(); | ||
| } | ||
|
|
||
| public static TaskContext set(Task task) { | ||
| TaskResult result = new TaskResult(task); | ||
| TaskContext context = new TaskContext(task, result); | ||
| TASK_CONTEXT_INHERITABLE_THREAD_LOCAL.set(context); | ||
| return context; | ||
| } | ||
|
|
||
| private final Task task; | ||
|
|
||
| private final TaskResult taskResult; | ||
|
|
||
| public String getWorkflowInstanceId() { | ||
| return task.getWorkflowInstanceId(); | ||
| } | ||
|
|
||
| public String getTaskId() { | ||
| return task.getTaskId(); | ||
| } | ||
|
|
||
| public int getRetryCount() { | ||
| return task.getRetryCount(); | ||
| } | ||
|
|
||
| public int getPollCount() { | ||
| return task.getPollCount(); | ||
| } | ||
|
|
||
| public long getCallbackAfterSeconds() { | ||
| return task.getCallbackAfterSeconds(); | ||
| } | ||
|
|
||
| public void addLog(String log) { | ||
| this.taskResult.log(log); | ||
| } | ||
|
|
||
| public Task getTask() { | ||
| return task; | ||
| } | ||
|
|
||
| public TaskResult getTaskResult() { | ||
| return taskResult; | ||
| } | ||
|
|
||
| public void setCallbackAfter(int seconds) { | ||
| this.taskResult.setCallbackAfterSeconds(seconds); | ||
| } | ||
| } |
32 changes: 32 additions & 0 deletions
32
...k/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/WorkerConfiguration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| * Copyright 2023 Netflix, Inc. | ||
| * <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.sdk.workflow.executor.task; | ||
|
|
||
| public class WorkerConfiguration { | ||
|
|
||
| private int defaultPollingInterval = 0; | ||
|
|
||
| public WorkerConfiguration(int defaultPollingInterval) { | ||
| this.defaultPollingInterval = defaultPollingInterval; | ||
| } | ||
|
|
||
| public WorkerConfiguration() {} | ||
|
|
||
| public int getPollingInterval(String taskName) { | ||
| return defaultPollingInterval; | ||
| } | ||
|
|
||
| public int getThreadCount(String taskName) { | ||
| return 0; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
41 changes: 41 additions & 0 deletions
41
...-sdk/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/TestWorkerConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| /* | ||
| * Copyright 2023 Netflix, Inc. | ||
| * <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.sdk.workflow.executor.task; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| public class TestWorkerConfig extends WorkerConfiguration { | ||
|
|
||
| private Map<String, Integer> pollingIntervals = new HashMap<>(); | ||
|
|
||
| private Map<String, Integer> threadCounts = new HashMap<>(); | ||
|
|
||
| @Override | ||
| public int getPollingInterval(String taskName) { | ||
| return pollingIntervals.getOrDefault(taskName, 0); | ||
| } | ||
|
|
||
| public void setPollingInterval(String taskName, int interval) { | ||
| pollingIntervals.put(taskName, interval); | ||
| } | ||
|
|
||
| public void setThreadCount(String taskName, int threadCount) { | ||
| threadCounts.put(taskName, threadCount); | ||
| } | ||
|
|
||
| @Override | ||
| public int getThreadCount(String taskName) { | ||
| return threadCounts.getOrDefault(taskName, 0); | ||
| } | ||
| } |
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we need to make sure that this bean is created later than other beans to avoid omissions?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Addressed this by doing the initialization using context initialization event.