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/09/25 08:11:03 UTC
[GitHub] [kafka] lucasbru opened a new pull request, #14437: KAFKA-10199: Fix restoration behavior for paused tasks
lucasbru opened a new pull request, #14437:
URL: https://github.com/apache/kafka/pull/14437
State updater can get into a busy loop when all tasks are paused, because `changelogReader` will never return that all changelogs have been read completely. Fix this, by awaiting if `updatingTasks` is empty.
Related and included: if we are restoring and all tasks are paused, we should return immediately from `StoreChangelogReader`.
### 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
[GitHub] [kafka] lucasbru commented on a diff in pull request #14437: KAFKA-10199: Fix restoration behavior for paused tasks
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14437:
URL: https://github.com/apache/kafka/pull/14437#discussion_r1335540581
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -1043,6 +1046,28 @@ public void shouldResumeActiveStatefulTask() throws Exception {
verify(changelogReader, times(2)).enforceRestoreActive();
}
+ @Test
+ public void shouldAwaitWhenAllTasksPaused() throws Exception {
Review Comment:
This tests the absence of "restore" calls when all tasks are paused. This requires "smelly" constructs like `Mockito.reset` and `Thread.sleep`. We could alternatively not test this - the absence of something in a certain timeframe is not very amenable to unit testing.
--
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
[GitHub] [kafka] cadonna merged pull request #14437: KAFKA-10199: Fix restoration behavior for paused tasks
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna merged PR #14437:
URL: https://github.com/apache/kafka/pull/14437
--
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
[GitHub] [kafka] cadonna commented on pull request #14437: KAFKA-10199: Fix restoration behavior for paused tasks
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #14437:
URL: https://github.com/apache/kafka/pull/14437#issuecomment-1735413305
Build failures are unrelated.
--
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
[GitHub] [kafka] lucasbru commented on a diff in pull request #14437: KAFKA-10199: Fix restoration behavior for paused tasks
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14437:
URL: https://github.com/apache/kafka/pull/14437#discussion_r1336770650
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -1043,6 +1046,28 @@ public void shouldResumeActiveStatefulTask() throws Exception {
verify(changelogReader, times(2)).enforceRestoreActive();
}
+ @Test
+ public void shouldAwaitWhenAllTasksPaused() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ stateUpdater.start();
+ stateUpdater.add(task);
+
+ when(topologyMetadata.isPaused(null)).thenReturn(true);
+
+ verifyPausedTasks(task);
+
+ reset(changelogReader);
Review Comment:
I don't think this is quite true. Paused tasks will not be added to `updatingTasks`, but we may still go around the loop calling `restore` on a potentially empty set. This is testing that after a while, we do not go around the loop anymore. But it's actually not a good test, for the reason described above.
--
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
[GitHub] [kafka] cadonna commented on a diff in pull request #14437: KAFKA-10199: Fix restoration behavior for paused tasks
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14437:
URL: https://github.com/apache/kafka/pull/14437#discussion_r1335624082
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -1043,6 +1046,28 @@ public void shouldResumeActiveStatefulTask() throws Exception {
verify(changelogReader, times(2)).enforceRestoreActive();
}
+ @Test
+ public void shouldAwaitWhenAllTasksPaused() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ stateUpdater.start();
+ stateUpdater.add(task);
+
+ when(topologyMetadata.isPaused(null)).thenReturn(true);
+
+ verifyPausedTasks(task);
+
+ reset(changelogReader);
Review Comment:
I think, you do not need to reset the mock since we guarantee that paused tasks that are added to the state updater are not restored. We miss a test for that case, though.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -1043,6 +1046,28 @@ public void shouldResumeActiveStatefulTask() throws Exception {
verify(changelogReader, times(2)).enforceRestoreActive();
}
+ @Test
+ public void shouldAwaitWhenAllTasksPaused() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ stateUpdater.start();
+ stateUpdater.add(task);
+
+ when(topologyMetadata.isPaused(null)).thenReturn(true);
+
+ verifyPausedTasks(task);
+
+ reset(changelogReader);
+ Thread.sleep(100);
+ verify(changelogReader, never()).restore(any());
Review Comment:
Instead of this, we could expose the thread state on the implementation of the state updater, but not on the interface. With that and a wait until verification, we could verify that the thread changes to the waiting state. We similar with `getActiveTasks()` or `getPausedTasks()`. That does also smell a bit, but it avoids the sleep and maybe smells a little bit less? 🙂
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -307,7 +307,7 @@ private void waitIfAllChangelogsCompletelyRead() {
tasksAndActionsLock.lock();
try {
while (isRunning.get() &&
- changelogReader.allChangelogsCompleted() &&
+ (changelogReader.allChangelogsCompleted() || updatingTasks.isEmpty()) &&
Review Comment:
Nice catch!
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -1043,6 +1046,28 @@ public void shouldResumeActiveStatefulTask() throws Exception {
verify(changelogReader, times(2)).enforceRestoreActive();
}
+ @Test
+ public void shouldAwaitWhenAllTasksPaused() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ stateUpdater.start();
+ stateUpdater.add(task);
+
+ when(topologyMetadata.isPaused(null)).thenReturn(true);
+
+ verifyPausedTasks(task);
+
+ reset(changelogReader);
+ Thread.sleep(100);
+ verify(changelogReader, never()).restore(any());
Review Comment:
Instead of this, we could expose the thread state on the implementation of the state updater, but not on the interface. With that and a wait until verification, we could verify that the thread changes to the waiting state. We similar with `getActiveTasks()` or `getPausedTasks()`. That does also smell a bit, but it avoids the sleep and maybe smells a little bit less? 🙂
--
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
[GitHub] [kafka] lucasbru commented on a diff in pull request #14437: KAFKA-10199: Fix restoration behavior for paused tasks
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14437:
URL: https://github.com/apache/kafka/pull/14437#discussion_r1336772774
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -1043,6 +1046,28 @@ public void shouldResumeActiveStatefulTask() throws Exception {
verify(changelogReader, times(2)).enforceRestoreActive();
}
+ @Test
+ public void shouldAwaitWhenAllTasksPaused() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ stateUpdater.start();
+ stateUpdater.add(task);
+
+ when(topologyMetadata.isPaused(null)).thenReturn(true);
+
+ verifyPausedTasks(task);
+
+ reset(changelogReader);
+ Thread.sleep(100);
+ verify(changelogReader, never()).restore(any());
Review Comment:
I like the idea!
--
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