Implement direct submission of tasks to Kafka. #47
Conversation
| * | ||
| * @param task task to execute | ||
| */ | ||
| void submitRootTaskDirect(Task task); |
There was a problem hiding this comment.
submitSimpleTaskDirect looks more relevant instead of RootTask
| @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
|
|
||
| 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.
| 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.
| } | ||
| } | ||
| }); | ||
| 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.
| 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
left a comment
pns-nirmata
left a comment
There was a problem hiding this comment.
Looking good. Please fix as per other comments, and then, a few unit tests for direct submit.
2 participants
Implement direct submission of tasks to Kafka. UT is yet to be done.