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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
