SEPURI-SAI-KRISHNA commented on PR #29037:
URL: https://github.com/apache/flink/pull/29037#issuecomment-5455504781
Not a reviewer on this, just read it closely because it touches a path I've
been in. The change looks right to me, I traced the drain path through the
valve as it stands on this base (so including #29024): `ACTIVE` reactivates the
subpartition, realigning it immediately if its stored watermark is still at or
above `lastOutputWatermark` and otherwise leaving `MAX_WATERMARK` to realign
it, and either way `findAndOutputNewMinWatermarkAcrossAlignedSubpartitions`
emits `MAX` as soon as the first source subtask drains. Two smaller things:
**The `isTaskDeployedAsFinished()` guard in `MultipleInputStreamTask`
appears to be dead code.** It's clearly load-bearing in the other two tasks,
where `FinishedOnRestoreMainOperatorOutput#emitWatermarkStatus` throws
unconditionally. But `getChainedSourceOutputs()` returns
`ChainingOutput`/`CopyingChainingOutput`
(`OperatorChain#createChainedSourceOutput`), whose `announcedStatus` is
initialised to `ACTIVE` and is never set to `IDLE` on a finished-on-restore
chain, so `emitWatermarkStatus(ACTIVE)` is deduplicated inside the output and
never reaches the input underneath. The comment also says the *outputs* reject
status events, which holds for the other two tasks but not here: it's
`FinishedOnRestoreInput#processWatermarkStatus` that would throw, and the dedup
keeps it unreachable. I checked this with a scratch test against unmodified
master: `emitWatermarkStatus(ACTIVE)` on a `ChainingOutput` wrapping a
`FinishedOnRestoreInput` does not throw, while the same call with `IDLE` thro
ws `ExceptionInChainedOperatorException` / `IllegalStateException`, so the
input is genuinely reachable, and it is the dedup rather than the guard that
keeps `ACTIVE` safe. `CopyingChainingOutput` behaves identically. Harmless
either way, but the copied comment makes it read as necessary.
**Which of the two new test classes is the regression test?**
`ChainingOutput` isn't touched by this PR, and I ran both tests from
`ChainingOutputIdleMaxWatermarkTest` against unmodified master, both green, so
they pass with and without the fix, useful as characterisation of the idle
gate, but they don't guard the change. `SourceOperatorStreamTaskIdleDrainTest`
is the one that does, I confirmed the red/green locally: without the fix it
reports `[WatermarkStatus(IDLE), EndOfData{mode=DRAIN}]`, with it the assertion
passes. (Applying all three hunks, `SourceOperatorStreamTaskTest`,
`SourceStreamTaskTest`, `MultipleInputStreamTaskTest`, `OneInputStreamTaskTest`
and `TwoInputStreamTaskTest` are 102/102 green here too, for whatever that's
worth alongside CI.) Might be worth saying so in the description, since
"verifies the output-level contract" reads like it guards the fix.
Minor while you're in there: `SourceOperatorStreamTaskTest` already has
`testEmittingMaxWatermarkAfterReadingAllRecords`, which is the exact non-idle
counterpart of the new case on the same harness, the two would read well next
to each other.
--
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]
