Nikita-Shupletsov commented on PR #23093: URL: https://github.com/apache/kafka/pull/23093#issuecomment-5207951577
>future.get() blocks the StreamThread inside the rebalance callback with zero log output — the old code at least printed "The state updater wasn't able to remove task X in time." Suggest looping future.get(1, MINUTES) with a WARN per iteration and retrying forever: keeps the fix, restores observability. Same applies to shutdownStateUpdater(), where KafkaStreams.close(Duration) will now return false and leak the StreamThread instead of the previously-bounded 5-min- per-future. sounds good. will do > Integration test coverage regression. StateUpdaterFailureIntegrationTest was added by https://github.com/apache/kafka/commit/c48c50d3e806f2d0dedbff7570ed11687799dd41 (KAFKA-19831) specifically for "potential failures in Task#maybeCheckpoint". The PR deletes the flush() override that injected a ProcessorStateException during maybeCheckpointTasks and replaces it with a throw from onRestoreEnd, i.e. inside restoreTasks. Different handler — maybeCheckpointTasks catches StreamsException per task and continues; restoreTasks goes through handleStreamsException. The flush case should stay as a third Arguments rather than being replaced. flush is no longer called because of the changes from https://issues.apache.org/jira/browse/KAFKA-19712. the original test kept passing because the failure was not induced anymore. That's why I rewrote the test a bit, so we test the old case again faithfully. > StateUpdater.shutdown() semantics changed for restored tasks. failRemainingTasks() moves everything in restoredActiveTasks into the failed queue, and shouldShutdownStateUpdater now asserts that. Those tasks restored successfully; being "failed" means closeTaskDirty, discarding the checkpoint. It doesn't bite production today because shutdownStateUpdater() drains them via removeRestoredTask() before calling shutdown() — but it makes shutdown() unsafe to call on its own, which the unit test now enshrines as expected behavior. so yes. it's another edge case my claude found: right now we don't and can't reuse a state updater. it lives as long as the thread it belongs to lives. but there was a unit test that was testing that we can shutdown and restart the same state updater which never happens in the real life. I decided to tightened it a bit to adhere to how we actually use it, to make it less errorprone. If we ever decide to reuse state updater or change the relationship between stream threads and state updater, there will at least be exceptions showing what we need to change instead of being in this gray zone. -- 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]
