You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "aliehsaeedii (via GitHub)" <gi...@apache.org> on 2024/09/16 12:44:20 UTC
[PR] KAFKA-17109 [kafka]
aliehsaeedii opened a new pull request, #17209:
URL: https://github.com/apache/kafka/pull/17209
This PR aims at resolving the issue made by https://github.com/apache/kafka/pull/17116
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17209:
URL: https://github.com/apache/kafka/pull/17209#discussion_r1773147020
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1247,6 +1248,36 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ final TaskManager.BackoffRecord backoffRecord = mock(TaskManager.BackoffRecord.class);
Review Comment:
Advancing `time` alone wont help since the backoff record corresponding to task00 is not existing in the map.
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17209:
URL: https://github.com/apache/kafka/pull/17209#discussion_r1774661420
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
Review Comment:
We use 4 spaces and not 8 for indentation.
```suggestion
.withInputPartitions(taskId00Partitions)
.inState(State.RESTORING).build();
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
Review Comment:
The test is very good!
I just have some formatting comments.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
Review Comment:
Please fix the indentation also here.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ // initializeIfNeeded() has NOT been called this time
+ verify(task00, Mockito.times(1)).initializeIfNeeded();
+ verify(tasks, Mockito.times(2)).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00))
+ );
+ verify(stateUpdater, never()).add(task00);
+
+ time.sleep(10000);
+ // do not throw lock exception this time
+ doNothing().when(task00).initializeIfNeeded();
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00, Mockito.times(2)).initializeIfNeeded();
+ verify(tasks, Mockito.times(2)).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00))
Review Comment:
Please fix the indentation
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
Review Comment:
Please fix the indentation.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
Review Comment:
Could you add an inline comment here stating:
```java
// task00 should not be initialized due to LockException, task01 should be initialized
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
Review Comment:
Could you add a `time.sleep(5000)` before this call, please?
Please add a new line between `time.sleep()` and `checkStateUpdater()`.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ // initializeIfNeeded() has NOT been called this time
+ verify(task00, Mockito.times(1)).initializeIfNeeded();
+ verify(tasks, Mockito.times(2)).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00))
Review Comment:
Please fix the indentation
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ // initializeIfNeeded() has NOT been called this time
+ verify(task00, Mockito.times(1)).initializeIfNeeded();
+ verify(tasks, Mockito.times(2)).addPendingTasksToInit(
Review Comment:
In this test class we use `times()` without `Mockito.`. Even in this test you use `never()` without `Mockito.`. Please remove the `Mockito.` to be consistent.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ // initializeIfNeeded() has NOT been called this time
+ verify(task00, Mockito.times(1)).initializeIfNeeded();
+ verify(tasks, Mockito.times(2)).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00))
+ );
+ verify(stateUpdater, never()).add(task00);
+
+ time.sleep(10000);
Review Comment:
I would add a new line after this line to highlight that time passed between the two initialization attempts.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
Review Comment:
Could you please rename the test to `shouldRetryInitializationWithBackoffWhenInitializationFails`?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ // initializeIfNeeded() has NOT been called this time
Review Comment:
This inline comment is not really clear. Could you please change it to something like:
```java
// task00 should not be initialized since the backoff period has not passed.
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ // initializeIfNeeded() has NOT been called this time
+ verify(task00, Mockito.times(1)).initializeIfNeeded();
+ verify(tasks, Mockito.times(2)).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00))
+ );
+ verify(stateUpdater, never()).add(task00);
+
+ time.sleep(10000);
+ // do not throw lock exception this time
Review Comment:
Could you change this comment to something like:
```java
// task00 should call initialize since the backoff period has passed
```
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17209:
URL: https://github.com/apache/kafka/pull/17209#discussion_r1763114787
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -2116,4 +2132,37 @@ boolean needsInitializationOrRestoration() {
void addTask(final Task task) {
tasks.addTask(task);
}
+
+ private boolean canTryLock(final TaskId taskId, final long nowMs) {
Review Comment:
@cadonna makes sense!
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2353620544
> Thanks for the PR, @aliehsaeedii !
>
> Here my feedback.
>
> I am missing unit tests.
Thanks Utest is added + review is addressed
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2457486322
@cadonna @aliehsaeedii Should we not port back these fixes to 3.9, 3.8 ?
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2514109625
@aliehsaeedii since the cherry-pick to 3.9 has conflicts, could you please open a PR for the port to 3.8?
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2569135872
@aliehsaeedii Have you opened a PR to resolve the conflicts against 3.9?
Once the commit cherry-picked to 3.9, I suspect that cherry-picking from 3.9 to 3.8 will be easier.
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2515059693
> @aliehsaeedii since the cherry-pick to 3.9 has conflicts, could you please open a PR for the port to 3.8?
There is conflict in `StreamTaskTest` for `3.8` as well.
--
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] KAFKA-17109 [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2352971503
@aliehsaeedii please update the PR title to have a description of the patch. Thanks!
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2511653675
> Yeah, that is a good idea! As far as I understand, the change seems to be well tested, right? @aliehsaeedii
yes. Should we do that?
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17209:
URL: https://github.com/apache/kafka/pull/17209#discussion_r1761706584
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -2116,4 +2132,37 @@ boolean needsInitializationOrRestoration() {
void addTask(final Task task) {
tasks.addTask(task);
}
+
+ private boolean canTryLock(final TaskId taskId, final long nowMs) {
Review Comment:
Sorry, I forgot to add this comment before in my review. Could you please rename this method to `canTryInitializeTask()`? I think that makes more sense.
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2376116982
@aliehsaeedii the following test fails consistently with a NPE with this PR:
`StreamThreadTest.shouldRecordCommitLatency()`.
See https://github.com/apache/kafka/actions/runs/11030675352?pr=17209
Could you have a look?
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #17209:
URL: https://github.com/apache/kafka/pull/17209#discussion_r1774921517
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
Review Comment:
For the 2nd try `5000` does not work but anything less than `1000` is good!
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17209:
URL: https://github.com/apache/kafka/pull/17209#discussion_r1775033648
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1243,6 +1245,50 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ doThrow(new LockException("Lock Exception!")).when(task00).initializeIfNeeded();
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
+
+ verify(task00).initializeIfNeeded();
+ verify(task01).initializeIfNeeded();
+ verify(tasks).addPendingTasksToInit(
+ argThat(tasksToInit -> tasksToInit.contains(task00) && !tasksToInit.contains(task01))
+ );
+ verify(stateUpdater, never()).add(task00);
+ verify(stateUpdater).add(task01);
+
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
Review Comment:
Of course, you are right! My bad!
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17209:
URL: https://github.com/apache/kafka/pull/17209#discussion_r1765850540
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -150,6 +153,11 @@ void setMainConsumer(final Consumer<byte[], byte[]> mainConsumer) {
this.mainConsumer = mainConsumer;
}
+ /* For testing */
+ void setTaskIdToBackoffRecord(final Map<TaskId, BackoffRecord> taskIdToBackoffRecord) {
+ this.taskIdToBackoffRecord = taskIdToBackoffRecord;
+ }
Review Comment:
I do not think you need this method if you can control time as I describe on line 1017.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1247,6 +1248,36 @@ public void shouldRetryInitializationWhenLockExceptionInStateUpdater() {
verify(stateUpdater).add(task01);
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldRetryInitializationWhenCanNotInitializeTask() {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTasksToInit()).thenReturn(mkSet(task00, task01));
+ final TaskManager.BackoffRecord backoffRecord = mock(TaskManager.BackoffRecord.class);
Review Comment:
You do not need this mock. You can advance time with the `time` object.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1006,14 +1014,22 @@ private void addTasksToStateUpdater() {
}
private void addTaskToStateUpdater(final Task task) {
+ final long nowMs = System.currentTimeMillis();
Review Comment:
Here, you need to use
```suggestion
final long nowMs = time.milliseconds();
```
We inject the time object at creation, so that we can control time for example in tests.
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17209:
URL: https://github.com/apache/kafka/pull/17209#discussion_r1761355474
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -2116,4 +2126,42 @@ boolean needsInitializationOrRestoration() {
void addTask(final Task task) {
tasks.addTask(task);
}
+
+ private void handleUnsuccessfulLockAcquiring(final Task task, final long nowMs) {
+ updateOrCreateBackoffRecord(task.id(), nowMs);
+ tasks.addPendingTasksToInit(Collections.singleton(task));
+ }
+
+ private boolean canTryLock(final TaskId taskId, final long nowMs) {
+ return !taskIdToBackoffRecord.containsKey(taskId) || taskIdToBackoffRecord.get(taskId).canAttempt(nowMs);
+ }
+
+ private void updateOrCreateBackoffRecord(final TaskId taskId, final long nowMs) {
+ if (taskIdToBackoffRecord.containsKey(taskId)) {
+ taskIdToBackoffRecord.get(taskId).recordAttempt(nowMs);
+ } else {
+ taskIdToBackoffRecord.put(taskId, new BackoffRecord(nowMs));
+ }
+ }
+
+ public static class BackoffRecord {
+ private long attempts;
+ private long lastAttemptMs;
+ private static final ExponentialBackoff EXPONENTIAL_BACKOFF = new ExponentialBackoff(1, 2, 10000, 0.5);
Review Comment:
Should the exponential back-off be specified in terms of poll time? Something like
```
new ExponentialBackoff(pollTime, 2, 10000, 0.5);
```
If it is to much trouble getting that config into the task manager, just choose something larger than 1ms. 1 ms sounds really small. The sequence of the back-offs would be 1ms, 2ms, 4ms, 8ms, 16ms, 32ms, 64, 128. At the same time, with default configs, the task initialization is attempted every 100ms. So, it seems there will not be much improvement to the current situation because the first 7 poll iterations you attempt to initialize the task.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1006,14 +1009,21 @@ private void addTasksToStateUpdater() {
}
private void addTaskToStateUpdater(final Task task) {
+ final long nowMs = System.currentTimeMillis();
try {
- task.initializeIfNeeded();
- stateUpdater.add(task);
+ if (canTryLock(task.id(), nowMs)) {
+ task.initializeIfNeeded();
+ stateUpdater.add(task);
+ taskIdToBackoffRecord.remove(task.id());
Review Comment:
Minor:
I would swap those two lines. Once the task is initialized, the backoff can be removed.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1006,14 +1009,21 @@ private void addTasksToStateUpdater() {
}
private void addTaskToStateUpdater(final Task task) {
+ final long nowMs = System.currentTimeMillis();
try {
- task.initializeIfNeeded();
- stateUpdater.add(task);
+ if (canTryLock(task.id(), nowMs)) {
+ task.initializeIfNeeded();
+ stateUpdater.add(task);
+ taskIdToBackoffRecord.remove(task.id());
+ } else {
+ log.trace("Task {} is still not allowed to retry acquiring the state directory lock", task.id());
+ handleUnsuccessfulLockAcquiring(task, nowMs);
Review Comment:
Is this correct?
Every time initialization is attempted before the back-off, the time of the last attempt is updated to the current time. If we assume an attempt every poll interval and the poll interval is less than the back-off time, the task will never be initialized.
Assume the last unsuccessful attempt occurred at time 200 and now the current call to `canTryLock()` is 100ms later at time 300. Furthermore, assume the current back-off is 250. That is, `canTryLock()` should return `false` because 300 - 200 >= 250 is not `true`. The last attempt is updated to 300 and the backoff is exponentially updated with the increased number of attempt (let's say 500). If you try again in 100ms at 400 `canTryLock()` will again return `false`, because 400 - 300 >= 500 is still not true and it will also not be true next time. You should only update the back-off record if you actually have attempted to initialize the task and it was unsuccessful and not when you skipped the attempt due to the back-off.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -2116,4 +2126,42 @@ boolean needsInitializationOrRestoration() {
void addTask(final Task task) {
tasks.addTask(task);
}
+
+ private void handleUnsuccessfulLockAcquiring(final Task task, final long nowMs) {
+ updateOrCreateBackoffRecord(task.id(), nowMs);
+ tasks.addPendingTasksToInit(Collections.singleton(task));
+ }
+
+ private boolean canTryLock(final TaskId taskId, final long nowMs) {
+ return !taskIdToBackoffRecord.containsKey(taskId) || taskIdToBackoffRecord.get(taskId).canAttempt(nowMs);
+ }
+
+ private void updateOrCreateBackoffRecord(final TaskId taskId, final long nowMs) {
+ if (taskIdToBackoffRecord.containsKey(taskId)) {
+ taskIdToBackoffRecord.get(taskId).recordAttempt(nowMs);
+ } else {
+ taskIdToBackoffRecord.put(taskId, new BackoffRecord(nowMs));
+ }
+ }
+
+ public static class BackoffRecord {
+ private long attempts;
+ private long lastAttemptMs;
+ private static final ExponentialBackoff EXPONENTIAL_BACKOFF = new ExponentialBackoff(1, 2, 10000, 0.5);
+
+
+ public BackoffRecord(final long nowMs) {
+ this.attempts = 1;
+ this.lastAttemptMs = nowMs;
+ }
+
+ public void recordAttempt(final long nowMs) {
+ this.attempts++;
+ this.lastAttemptMs = nowMs;
+ }
+
+ public boolean canAttempt(final long nowMs) {
+ return nowMs - lastAttemptMs >= EXPONENTIAL_BACKOFF.backoff(attempts);
Review Comment:
nit:
```suggestion
return nowMs - lastAttemptMs >= EXPONENTIAL_BACKOFF.backoff(attempts);
```
--
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] KAFKA-17109: Move lock backoff retry to streams TaskManager [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #17209:
URL: https://github.com/apache/kafka/pull/17209#issuecomment-2459228622
Yeah, that is a good idea! As far as I understand, the change seems to be well tested, right? @aliehsaeedii
--
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