KAFKA-10199: Add remove operation with future to state updater #15852
Conversation
Adds a remove operation to the state updater that returns a future instead of adding the removed tasks to an output queue. Code that uses the state updater can then wait on the future.
There was a problem hiding this comment.
I did not care to keep the unit tests for the remove without future since that method will be removed in a future PR.
| break; | ||
| case REMOVE: | ||
| removeTask(taskAndAction.getTaskId()); | ||
| if (taskAndAction.futureForRemove() == null) { |
There was a problem hiding this comment.
Do we need the variant without the future?
There was a problem hiding this comment.
No, we do not but for the sake of breaking down my gigantic PR, I need to temporary keep it until one of my future PRs makes it obsolete. It will be simply called remove() after that.
| private void removeTask(final TaskId taskId, final CompletableFuture<RemovedTaskResult> future) { | ||
| try { | ||
| if (updatingTasks.containsKey(taskId)) { | ||
| removeUpdatingTask(taskId, future); |
There was a problem hiding this comment.
Could all four remove...Task methods just return a boolean if successful?
There was a problem hiding this comment.
Done! Code looks better now. Thanks a lot!
| * | ||
| * @param taskId ID of the task to remove | ||
| */ | ||
| CompletableFuture<RemovedTaskResult> removeWithFuture(final TaskId taskId); |
There was a problem hiding this comment.
What about remove()? 🙂
See my comment above.
| future.complete(null); | ||
| log.warn("Task " + taskId + " could not be removed from the state updater because " | ||
| + "the state updater does not own this task."); | ||
| if (!removeUpdatingTask(taskId, future)) { |
There was a problem hiding this comment.
nit: Used short-circuiting to avoid the nesting
if (!removeUpdatingTask(taskId, future)
&& !removePausedTask(taskId, future)
&& ..
There was a problem hiding this comment.
Oh, thanks! I am going to change that in one of my future PRs.
…e#15852) Adds a remove operation to the state updater that returns a future instead of adding the removed tasks to an output queue. Code that uses the state updater can then wait on the future. Reviewer: Lucas Brutschy <lbrutschy@confluent.io>
Adds a remove operation to the state updater that returns a future instead of adding the removed tasks to an output queue. Code that uses the state updater can then wait on the future.
Committer Checklist (excluded from commit message)