weiqingy commented on code in PR #27083:
URL: https://github.com/apache/flink/pull/27083#discussion_r3888391817
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/FinishedOnRestoreInput.java:
##########
@@ -58,7 +59,22 @@ public void processWatermark(Watermark watermark) {
@Override
public void processWatermarkStatus(WatermarkStatus watermarkStatus) throws
Exception {
- throw new IllegalStateException();
+ // Tasks that finished on restore may receive FINISHED watermark
status from upstream.
+ // Aggregate FINISHED status from all inputs and emit once all are
received, similar to
+ // how MAX_WATERMARK is handled. This ensures proper watermark status
propagation.
+ if (watermarkStatus.isFinished()) {
+ if (++finishedStatusSeen == inputCount) {
+ for (RecordWriterOutput<?> streamOutput : streamOutputs) {
+ streamOutput.emitWatermarkStatus(watermarkStatus);
+ }
+ }
+ return;
+ }
+ // Other watermark statuses (ACTIVE, IDLE) should not occur for
finished tasks
Review Comment:
Thanks for the review, and sorry for the long delay in getting back to you.
Both done in `56e06c2471d`. The guard is inverted, so the bad case throws first
and the `return` is gone.
For the identifier, the class had no task reference, so I passed the task
name in. `OperatorChain` already has `containingTask` at that line, so it is
just `containingTask.getName()`. I added it to the `processWatermark` message
too, since that one had the same gap.
The branch is also rebased onto master to pick up FLINK-40475, which had
rewritten the same block in `StatusWatermarkValve`.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -746,6 +748,41 @@ private boolean isCurrentSyncSavepoint(long checkpointId) {
*/
protected void advanceToEndOfEventTime() throws Exception {}
+ /**
+ * Emits FINISHED watermark status to indicate this task has permanently
completed and should be
+ * excluded from watermark aggregation in downstream operators.
+ *
+ * <p>This method is overridden by source tasks to emit FINISHED status
directly to downstream
+ * via network outputs. Processing tasks (e.g., OneInputStreamTask,
TwoInputStreamTask) do not
+ * override this method because they rely on StatusWatermarkValve to
aggregate FINISHED status
+ * from upstream inputs and propagate it downstream automatically.
+ */
+ protected void emitFinishedStatus() {
Review Comment:
Fair point. What stopped me is that `advanceToEndOfEventTime()` sits eleven
lines above with the same shape: a no-op on `StreamTask`, overridden by the
same three source tasks, same "does nothing for other tasks" javadoc.
The three overriders also have no shared base to move to.
`SourceStreamTask`, `SourceOperatorStreamTask` and `MultipleInputStreamTask`
all extend `StreamTask` directly, and the last one is not a pure source task.
So I matched the existing pattern rather than adding a second one next to
it. Would you rather see both methods lifted into a source-specific type as a
separate change?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -746,6 +748,41 @@ private boolean isCurrentSyncSavepoint(long checkpointId) {
*/
protected void advanceToEndOfEventTime() throws Exception {}
+ /**
+ * Emits FINISHED watermark status to indicate this task has permanently
completed and should be
+ * excluded from watermark aggregation in downstream operators.
+ *
+ * <p>This method is overridden by source tasks to emit FINISHED status
directly to downstream
+ * via network outputs. Processing tasks (e.g., OneInputStreamTask,
TwoInputStreamTask) do not
+ * override this method because they rely on StatusWatermarkValve to
aggregate FINISHED status
+ * from upstream inputs and propagate it downstream automatically.
+ */
+ protected void emitFinishedStatus() {
+ // Empty by default - only source tasks override this method.
+ }
+
+ /**
+ * Helper method to emit FINISHED watermark status to all stream outputs.
Subclasses can call
+ * this from their emitFinishedStatus() override if needed.
+ */
+ protected void emitFinishedStatusToOutputs() {
+ try {
+ if (operatorChain != null) {
+ RecordWriterOutput<?>[] streamOutputs =
operatorChain.getStreamOutputs();
Review Comment:
It can be empty. `streamOutputs` comes from
`configuration.getVertexNonChainedOutputs(...)`, so a chain ending in a sink
has none, which is the shape when a source is chained straight to a sink.
In that case there is no downstream task to tell, so nothing is lost, and a
warn would fire on every drain of every such task. Do you see a case where
empty means something actually went wrong?
The `operatorChain != null` warn above is different, since that one is
genuinely unexpected.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/watermarkstatus/WatermarkStatus.java:
##########
@@ -81,21 +81,25 @@ public final class WatermarkStatus extends StreamElement {
public static final int IDLE_STATUS = -1;
public static final int ACTIVE_STATUS = 0;
+ public static final int FINISHED_STATUS = 1;
Review Comment:
It cannot be one here. `WatermarkStatus extends StreamElement`, and a Java
enum already extends `java.lang.Enum`, so it cannot have another parent. Making
it an enum would mean turning `StreamElement` into an interface, which touches
every stream element type.
The int is also the wire format. `StreamElementSerializer` writes
`getStatus()` and reads it back as `new WatermarkStatus(source.readInt())`.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/watermarkstatus/WatermarkStatus.java:
##########
@@ -81,21 +81,25 @@ public final class WatermarkStatus extends StreamElement {
public static final int IDLE_STATUS = -1;
public static final int ACTIVE_STATUS = 0;
+ public static final int FINISHED_STATUS = 1;
public static final WatermarkStatus IDLE = new
WatermarkStatus(IDLE_STATUS);
public static final WatermarkStatus ACTIVE = new
WatermarkStatus(ACTIVE_STATUS);
+ public static final WatermarkStatus FINISHED = new
WatermarkStatus(FINISHED_STATUS);
public final int status;
public WatermarkStatus(int status) {
- if (status != IDLE_STATUS && status != ACTIVE_STATUS) {
+ if (status != IDLE_STATUS && status != ACTIVE_STATUS && status !=
FINISHED_STATUS) {
Review Comment:
Through deserialization. `StreamElementSerializer.deserialize` calls `new
WatermarkStatus(source.readInt())` with whatever int arrives on the wire, so
this is where an unknown value gets rejected instead of becoming a status that
nothing handles.
The check predates this PR. I only added `FINISHED_STATUS` to the allowed
set. It does matter a bit more now, since a mixed-version deployment could send
a value the receiver does not know.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/FinishedOnRestoreInput.java:
##########
@@ -30,6 +30,7 @@ public class FinishedOnRestoreInput<IN> implements Input<IN> {
private final int inputCount;
private int watermarksSeen = 0;
+ private int finishedStatusSeen = 0;
Review Comment:
Good question, and it found something I had missed.
Short version: FINISHED never reaches the DataStream V2 event time path, so
V2 does not get this fix. And if it ever did reach it, it would arrive as
active, which is the state this PR exists to prevent.
The reason is that `AbstractStreamTaskNetworkInput` runs two separate
aggregations over the same channels. `WatermarkStatus` comes in as a
`StreamElement` and goes to `statusWatermarkValve`. V2 watermarks come in as a
`WatermarkEvent` and go to `EventTimeWatermarkCombiner`, which extends the same
valve class but is a different instance with its own state.
`combineWatermark` only feeds two things into that valve, the event time
watermark and the idle flag, so nothing there ever becomes FINISHED. On the way
back out, `WrappedDataOutput.emitWatermarkStatus` converts using `isIdle()`,
and `IDLE_STATUS_WATERMARK_DECLARATION` is a `BoolWatermarkDeclaration`. A
boolean has no room for a third state, so FINISHED would read as not idle,
meaning active.
It is unreachable today, but it is a silent downgrade waiting for someone to
add a FINISHED path, so I made it throw instead, matching the other unsupported
cases in that class.
Doing this properly in V2 needs a third state in `EventTimeExtension`, which
is public API. We can create a follow-up PR for that, or a FLIP if it needs the
wider discussion. Does that split seem right to you?
--
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]
