kfaraz commented on code in PR #13328:
URL: https://github.com/apache/druid/pull/13328#discussion_r1016866569
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -2276,17 +2280,33 @@ public Void apply(@Nullable Boolean result)
);
}
+ /**
+ * Determines whether a given task was created by the current version of the
supervisor
+ * One can choose to pass a map of activeTasks. If the taskId doesn't
correspond to any task in it, fallback to the db
+ * @param taskGroupId task group id
+ * @param taskId task id
+ * @param activeTaskMap Set of active tasks that were pre-fetched
+ * @return true if the task was created by the current supervisor
+ */
@VisibleForTesting
- public boolean isTaskCurrent(int taskGroupId, String taskId)
+ public boolean isTaskCurrent(int taskGroupId, String taskId, Map<String,
Task> activeTaskMap)
{
- Optional<Task> taskOptional = taskStorage.getTask(taskId);
- if (!taskOptional.isPresent() ||
!doesTaskTypeMatchSupervisor(taskOptional.get())) {
- return false;
+ @SuppressWarnings("unchecked")
+ SeekableStreamIndexTask<PartitionIdType, SequenceOffsetType, RecordType>
task;
+
+ if (activeTaskMap == null || !activeTaskMap.containsKey(taskId)) {
Review Comment:
The map is never updated in this method. We could just pass the `Task`
object itself to this method by calling `isTaskCurrent(taskGroupId,
activeTaskMap.get(taskId))` instead of `isTaskCurrent(groupId, taskId,
activeTaskMap)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
