Implement direct submission of tasks to Kafka. #47
+76
−11
Conversation
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
pns-nirmata
reviewed
May 18, 2022
| * | ||
| * @param task task to execute | ||
| */ | ||
| void submitRootTaskDirect(Task task); |
There was a problem hiding this comment.
submitSimpleTaskDirect looks more relevant instead of RootTask
pns-nirmata
reviewed
May 18, 2022
| @Override | ||
| public void submitRootTaskDirect(Task task) | ||
| { | ||
| throw(new UnsupportedOperationException("Direct submission to Kafka is unsupported in Zookeeper based workflow")); |
There was a problem hiding this comment.
Direct submission is ....
"To Kafka" is not needed
pns-nirmata
reviewed
May 18, 2022
|
|
||
| StartedTask startedTask = new StartedTask(this.getInstanceName(), | ||
| LocalDateTime.now(Clock.systemUTC()), 0); | ||
| storageMgr.setStartedTask(runId, task.getTaskId(), this.getSerializer().serialize(startedTask)); |
There was a problem hiding this comment.
No storage manager here. There could be numerous directly submitted tasks with no relation between them.
pns-nirmata
reviewed
May 18, 2022
| @@ -85,7 +86,11 @@ public class WorkflowManagerKafkaImpl implements WorkflowManager, WorkflowAdmin | |||
| private final Serializer serializer; | |||
| private final Executor taskRunnerService; | |||
|
|
|||
| private Map<TaskType, Producer<String, byte[]>> taskQueues = new HashMap<TaskType, Producer<String, byte[]>>(); | |||
| private Map<String, Set<String>> startedTasksCache = new HashMap<String, Set<String>>(); | |||
There was a problem hiding this comment.
This startedTasksCache has no use.
pns-nirmata
reviewed
May 18, 2022
| } | ||
| } | ||
| }); | ||
| startedTasksCache.get(runId.getId()).add(task.getTaskId().getId()); |
There was a problem hiding this comment.
Cache of no use and will keep hogging memory, ultimately stopping the program.
pns-nirmata
reviewed
May 18, 2022
| } catch (Exception e) { | ||
| log.error("Could not set completed data for executable task: {}", executableTask, e); | ||
| throw e; | ||
| } | ||
| } |
There was a problem hiding this comment.
Maybe, a DEBUG message in else, to say, executed the direct-submit task.
pns-nirmata
suggested changes
May 18, 2022
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
Implement direct submission of tasks to Kafka. UT is yet to be done.