Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java (Java)

Code before:
        processRatioSensor.record((double) totalProcessLatency / runOnceLatency, now);
        punctuateRatioSensor.record((double) totalPunctuateLatency / runOnceLatency, now);
        pollRatioSensor.record((double) pollLatency / runOnceLatency, now);
        commitRatioSensor.record((double) totalCommitLatency / runOnceLatency, now);

        final boolean logProcessingSummary = now - lastLogSummaryMs > LOG_SUMMARY_INTERVAL_MS;
        if (logProcessingSummary) {
            log.info("Processed {} total records, ran {} punctuators, and committed {} total tasks since the last update",
                 totalRecordsProcessedSinceLastSummary, totalPunctuatorsSinceLastSummary, totalCommittedSinceLastSummary);

            totalRecordsProcessedSinceLastSummary = 0L;
            totalPunctuatorsSinceLastSummary = 0L;
            totalCommittedSinceLastSummary = 0L;
            lastLogSummaryMs = now;
        }
    }

    private void initializeAndRestorePhase() {
        final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
        final State stateSnapshot = state;
        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);

                if (taskManager.tryToCompleteRestoration(now, offsetResetter)) {
                    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("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.");
        }
    }

Review comment (refactor):
Shall we move this `if` to where `initializeAndRestorePhase()` is called? Then we would have both checks about whether the state updater is enabled in the same method not far from each other. 

Code after:
            // multiple iterations with reasonably large max.num.records and hence is less vulnerable to outliers
            taskManager.recordTaskProcessRatio(totalProcessLatency, now);
        }

        now = time.milliseconds();
        final long runOnceLatency = now - startMs;
        processRecordsSensor.record(totalProcessed, now);
        processRatioSensor.record((double) totalProcessLatency / runOnceLatency, now);
        punctuateRatioSensor.record((double) totalPunctuateLatency / runOnceLatency, now);
        pollRatioSensor.record((double) pollLatency / runOnceLatency, now);
        commitRatioSensor.record((double) totalCommitLatency / runOnceLatency, now);

        final boolean logProcessingSummary = now - lastLogSummaryMs > LOG_SUMMARY_INTERVAL_MS;
        if (logProcessingSummary) {
            log.info("Processed {} total records, ran {} punctuators, and committed {} total tasks since the last update",
                 totalRecordsProcessedSinceLastSummary, totalPunctuatorsSinceLastSummary, totalCommittedSinceLastSummary);

            totalRecordsProcessedSinceLastSummary = 0L;
            totalPunctuatorsSinceLastSummary = 0L;
            totalCommittedSinceLastSummary = 0L;
            lastLogSummaryMs = now;
        }
    }

    private void initializeAndRestorePhase() {
        final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
        final State stateSnapshot = state;
        // 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);

            if (taskManager.tryToCompleteRestoration(now, offsetResetter)) {
                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("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.