You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2023/10/16 12:46:11 UTC
[PR] FIX: allow removing a suspended task from task registry. [kafka]
lucasbru opened a new pull request, #14555:
URL: https://github.com/apache/kafka/pull/14555
When we get a suspended task re-assigned in the eager rebalance protocol, we have to add the task back to the state updater so that it has a chance to catch up with its change log.
This was prevented by a check in `Tasks`, which disallows removing SUSPENDED tasks from the task registry. I couldn't find a reason why this must be an invariant of the task registry, so this weakens the check.
The error happens in the integration between TaskRegistry and TaskManager. However, this change anyway adds unit tests to more closely specify the intended behavior of the two modules.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: allow removing a suspended task from task registry. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14555:
URL: https://github.com/apache/kafka/pull/14555#discussion_r1361825789
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TasksTest.java:
##########
@@ -52,6 +55,20 @@ public class TasksTest {
private final Tasks tasks = new Tasks(new LogContext());
+ @Test
+ public void shouldCheckStateWhenClosedTasks() {
Review Comment:
Done
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: allow removing a suspended task from task registry. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #14555:
URL: https://github.com/apache/kafka/pull/14555
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: allow removing a suspended task from task registry. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14555:
URL: https://github.com/apache/kafka/pull/14555#discussion_r1360674675
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java:
##########
@@ -240,8 +241,8 @@ public synchronized void addTask(final Task task) {
public synchronized void removeTask(final Task taskToRemove) {
final TaskId taskId = taskToRemove.id();
- if (taskToRemove.state() != Task.State.CLOSED) {
- throw new IllegalStateException("Attempted to remove a task that is not closed: " + taskId);
+ if (taskToRemove.state() != Task.State.CLOSED && taskToRemove.state() != State.SUSPENDED) {
Review Comment:
:( I actually wanted to move the pause after the removal but forgot it. Good catch!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: allow removing a suspended task from task registry. [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14555:
URL: https://github.com/apache/kafka/pull/14555#discussion_r1360631466
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java:
##########
@@ -240,8 +241,8 @@ public synchronized void addTask(final Task task) {
public synchronized void removeTask(final Task taskToRemove) {
final TaskId taskId = taskToRemove.id();
- if (taskToRemove.state() != Task.State.CLOSED) {
- throw new IllegalStateException("Attempted to remove a task that is not closed: " + taskId);
+ if (taskToRemove.state() != Task.State.CLOSED && taskToRemove.state() != State.SUSPENDED) {
Review Comment:
I think that needs to be `RESTORING` instead of `SUSPENDED`. I assume you came to this change by looking at this code:
https://github.com/apache/kafka/blob/c7f730d9d9a95ce43b19a8af06e9f5be5e25609b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java#L511-L512
Here the task gets resumed (i.e., `SUSPENDED` -> `RESTORING`) before it is removed from the task registry.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: allow removing a suspended task from task registry. [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14555:
URL: https://github.com/apache/kafka/pull/14555#discussion_r1361732874
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TasksTest.java:
##########
@@ -52,6 +55,20 @@ public class TasksTest {
private final Tasks tasks = new Tasks(new LogContext());
+ @Test
+ public void shouldCheckStateWhenClosedTasks() {
Review Comment:
nit: the name of this unit test is a bit misleading. I would have expected something like `shouldCheckStateWhenRemoveTask()`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org