vamossagar12 commented on code in PR #11433:
URL: https://github.com/apache/kafka/pull/11433#discussion_r1164262483
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -342,7 +342,18 @@ public void handleAssignment(final Map<TaskId,
Set<TopicPartition>> activeTasks,
maybeThrowTaskExceptions(taskCloseExceptions);
- createNewTasks(activeTasksToCreate, standbyTasksToCreate);
+ final Collection<Task> newActiveTasks =
createNewTasks(activeTasksToCreate, standbyTasksToCreate);
+ // If there are any transactions in flight and there are newly created
active tasks, commit the tasks
+ // to avoid potential long restoration times.
+ if (processingMode == EXACTLY_ONCE_V2 &&
threadProducer().transactionInFlight() && !newActiveTasks.isEmpty()) {
+ log.info("New active tasks were added and there is an inflight
transaction. Attempting to commit tasks.");
+ final int numCommitted =
commitTasksAndMaybeUpdateCommittableOffsets(newActiveTasks, new HashMap<>());
+ if (numCommitted == -1) {
Review Comment:
Yeah that test didn't trigger `onAssignment` and invoked
`handleRebalanceComplete` directly which explains the problem which means that
for that test to work, we need to retain the `rebalanceInProgress = false` in
`handleRebalanceComplete` as well. I think it should be ok to have this line in
both `handleAssignment` and `handleRebalanceComplete` because the latter
anyways signifies a rebalance completion.
But, there are a couple of other things which seem to break now (sorry :( )
because of this new condition. One of them is
`StreamThread#shouldOnlyCompleteShutdownAfterRebalanceNotInProgress` . This one
is interesting because it is invoking `handleAssignment` and then `start` and
`stop` on the StreamThread and asserting that while rebalanance is in progress,
active tasks are still returned. Now because, we are explicitly setting
`rebalanceInProgress` to false in `handleAssignment` itself, this test fails.
Since this is EOS-V1 related test, I moved the `rebalanceInProgress = false`
inside the EOSv2 check and this test passes.
But then, because of `rebalanceInProgress = false` within
`handleAssignment`,
`EosV2UpgradeIntegrationTest#shouldUpgradeFromEosAlphaToEosV2` fails when
clients are closed cleanly. This one passes the moment I remove the line
`rebalanceInProgress = false` from `handleAssignment`.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
