cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964139165
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -867,37 +870,47 @@ void runOnce() {
}
private void initializeAndRestorePhase() {
- // only try to initialize the assigned tasks
- // if the state is still in PARTITION_ASSIGNED after the poll call
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter
= partitions -> resetOffsets(partitions, null);
final State stateSnapshot = state;
- if (stateSnapshot == State.PARTITIONS_ASSIGNED
- || stateSnapshot == State.RUNNING &&
taskManager.needsInitializationOrRestoration()) {
+ if (stateUpdaterEnabled) {
+ checkStateUpdater();
+ } else {
+ // only try to initialize the assigned tasks
+ // if the state is still in PARTITION_ASSIGNED after the poll call
+ if (stateSnapshot == State.PARTITIONS_ASSIGNED
+ || stateSnapshot == State.RUNNING &&
taskManager.needsInitializationOrRestoration()) {
- log.debug("State is {}; initializing tasks if necessary",
stateSnapshot);
+ log.debug("State is {}; initializing tasks if necessary",
stateSnapshot);
- // transit to restore active is idempotent so we can call it
multiple times
- changelogReader.enforceRestoreActive();
+ if (taskManager.tryToCompleteRestoration(now, offsetResetter))
{
+ log.info("Restoration took {} ms for all tasks {}",
time.milliseconds() - lastPartitionAssignedMs,
+ taskManager.allTasks().keySet());
+ setState(State.RUNNING);
+ }
- if (taskManager.tryToCompleteRestoration(now, partitions ->
resetOffsets(partitions, null))) {
- changelogReader.transitToUpdateStandby();
- log.info("Restoration took {} ms for all tasks {}",
time.milliseconds() - lastPartitionAssignedMs,
- taskManager.allTasks().keySet());
- setState(State.RUNNING);
+ if (log.isDebugEnabled()) {
+ log.debug("Initialization call done. State is {}", state);
+ }
}
if (log.isDebugEnabled()) {
- log.debug("Initialization call done. State is {}", state);
+ log.debug("Idempotently invoking restoration logic in state
{}", state);
}
+ // we can always let changelog reader try restoring in order to
initialize the changelogs;
+ // if there's no active restoring or standby updating it would not
try to fetch any data
+ // After KAFKA-13873, we only restore the not paused tasks.
+ changelogReader.restore(taskManager.notPausedTasks());
+ log.debug("Idempotent restore call done. Thread state has not
changed.");
}
+ }
- if (log.isDebugEnabled()) {
- log.debug("Idempotently invoking restoration logic in state {}",
state);
+ private void checkStateUpdater() {
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter
= partitions -> resetOffsets(partitions, null);
+ final State stateSnapshot = state;
Review Comment:
I have to admit that I copied this from `initializeAndRestorePhase()` and
also there I did not 100% understand why it was done there since I also thought
that there is no concurrency. I then decided to be conservative and trust the
original author.
--
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
