weiqingy commented on issue #939: URL: https://github.com/apache/flink-agents/issues/939#issuecomment-5150033381
Thanks for the detailed repro. Confirmed the mechanism on current `main`, not just 0.2.1 / Flink 1.20.3: - `actionTasksKState` and `pendingInputEventsKState` are keyed state, while `currentProcessingKeysOpState` is operator state. That is why the job still reports FINISHED: `endInput` waits on the operator state that survived, while the per-key task lists are already gone. - At Flink 2.3.0, which `main` builds against, `BatchExecutionKeyedStateBackend.setCurrentKey` still clears every registered state on any key change, so this is not specific to 1.20.x. - The operator crosses that boundary more than once per record: `processEvent` submits a mail, `processActionTaskForKey` starts with `setCurrentKey(key)`, and an unfinished task submits another mail. Agreed that `enabled=false` is an escape hatch, not a fix. The option defaults to `true` (`ExecutionOptions.USE_BATCH_STATE_BACKEND`), so a plain BATCH job hits this with no configuration at all, and that default path is the one that has to work. Two ways I can see to get there: 1. Finish the current key's whole action chain before returning from `processElement`, which is what the batch backend assumes callers do anyway. `waitInFlightEventsFinished` already does this at `endInput` by spinning on `mailboxExecutor.yield()`. 2. Keep the in-flight structures off keyed state in batch, since BATCH does not checkpoint and a plain per-key map would survive the key changes. I lean toward the first. The second does not stop at those two lists: `sequenceNumberKState`, `sensoryMemState` and `shortTermMemState` are keyed too, so preserving cross-key overlap means moving all per-key runtime state off the batch backend, which is close to reimplementing `enabled=false` inside the operator. And the overlap that draining gives up is not something `enabled=true` can support anyway. One open question either way: the behavior has to be conditional on the batch backend, and `getKeyedStateBackend() instanceof BatchExecutionKeyedStateBackend` is the signal I found. Is there a cleaner one? Batch also has no coverage today, every `set_runtime_mode` in the e2e suites passes `STREAMING`, so the regression test you described would be the first of its kind. Glad to take this on. WDYT? -- 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]
