MartijnVisser commented on code in PR #27083:
URL: https://github.com/apache/flink/pull/27083#discussion_r4427541348
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/watermarkstatus/StatusWatermarkValve.java:
##########
@@ -212,84 +243,214 @@ public void inputWatermark(Watermark watermark, int
channelIndex, DataOutput<?>
public void inputWatermarkStatus(
WatermarkStatus watermarkStatus, int channelIndex, DataOutput<?>
output)
throws Exception {
- // Shared input channel is only enabled in batch jobs, which do not
have watermark status
- // events.
- Preconditions.checkState(!isInputChannelShared);
+ // Shared input channel is only enabled in batch jobs. Batch jobs do
not perform
+ // watermark-based processing, so watermark status events can be
safely ignored.
+ // Note: Tasks may emit FINISHED status on completion regardless of
execution mode.
+ if (isInputChannelShared) {
+ LOG.debug(
+ "Ignoring watermark status {} on channel {} (shared input
channels indicate batch mode)",
+ watermarkStatus,
+ channelIndex);
+ return;
+ }
+
SubpartitionStatus subpartitionStatus =
subpartitionStatuses.get(channelIndex).get(subpartitionIndexes[channelIndex]);
// It is supposed that WatermarkStatus will not appear in jobs where
one input channel
// consumes multiple subpartitions, so we do not need to map
channelIndex into
// subpartitionStatusIndex for now, like what is done on Watermarks.
- // only account for watermark status inputs that will result in a
status change for the
- // subpartition
- if (watermarkStatus.isIdle() &&
subpartitionStatus.watermarkStatus.isActive()) {
- // handle active -> idle toggle for the subpartition
- subpartitionStatus.watermarkStatus = WatermarkStatus.IDLE;
-
- // the subpartition is now idle, therefore not aligned
- markWatermarkUnaligned(subpartitionStatus);
-
- // if all subpartitions of the valve are now idle, we need to
output an idle stream
- // status from the valve (this also marks the valve as idle)
- if
(!SubpartitionStatus.hasActiveSubpartitions(subpartitionStatuses)) {
- // now that all subpartitions are idle and no subpartition
will continue to advance
- // its watermark, we should "flush" all watermarks across all
subpartitions;
- // effectively, this means emitting the max watermark across
all subpartitions as
- // the new watermark, so that the eventual watermark is
independent of the order in
- // which the subpartitions became idle. This also keeps the
valve consistent with
- // the unconditional flush in CombinedWatermarkStatus.
- findAndOutputMaxWatermarkAcrossAllSubpartitions(output);
-
- lastOutputWatermarkStatus = WatermarkStatus.IDLE;
- output.emitWatermarkStatus(lastOutputWatermarkStatus);
+ WatermarkStatus currentSubpartitionStatus =
subpartitionStatus.watermarkStatus;
+
+ // Ignore if no change
+ if (watermarkStatus.equals(currentSubpartitionStatus)) {
+ return;
+ }
+
+ // Handle all valid status transitions
+ if (currentSubpartitionStatus.isActive()) {
+ if (watermarkStatus.isIdle()) {
+ subpartitionStatus.watermarkStatus = WatermarkStatus.IDLE;
+ markWatermarkUnaligned(subpartitionStatus);
+
+ // Emit watermark first (final progression before going idle).
This is
+ // unconditional: the subpartition just left the aligned set,
so the aggregation
+ // may yield a larger watermark even when this subpartition
did not hold the
+ // previous minimum. tryEmitNewWatermark only emits on strict
advance, so
+ // monotonicity is preserved.
+ tryEmitNewWatermark(output);
+ // Then update and emit global watermark status
+ tryEmitNewGlobalWatermarkStatus(output);
+ } else if (watermarkStatus.isFinished()) {
+ subpartitionStatus.watermarkStatus = WatermarkStatus.FINISHED;
+ markWatermarkUnaligned(subpartitionStatus);
+
+ // Unconditional for the same reason as the idle transition
above.
+ tryEmitNewWatermark(output);
+ // Then update and emit global watermark status
+ tryEmitNewGlobalWatermarkStatus(output);
} else {
- // the min watermark across the remaining aligned
subpartitions may be larger than
- // the last output watermark, now that the subpartition that
just became idle is no
- // longer part of the aligned set
- findAndOutputNewMinWatermarkAcrossAlignedSubpartitions(output);
+ throw new IllegalStateException(
+ "Invalid ACTIVE -> "
+ + watermarkStatus
+ + " transition for channel "
+ + channelIndex
+ + " subpartition "
+ + subpartitionIndexes[channelIndex]);
}
- } else if (watermarkStatus.isActive() &&
subpartitionStatus.watermarkStatus.isIdle()) {
- // handle idle -> active toggle for the subpartition
- subpartitionStatus.watermarkStatus = WatermarkStatus.ACTIVE;
-
- // if the last watermark of the subpartition, before it was marked
idle, is still
- // larger than the overall last output watermark of the valve,
then we can set the
- // subpartition to be aligned already.
- if (subpartitionStatus.watermark >= lastOutputWatermark) {
- markWatermarkAligned(subpartitionStatus);
+ } else if (currentSubpartitionStatus.isIdle()) {
+ if (watermarkStatus.isActive()) {
+ subpartitionStatus.watermarkStatus = WatermarkStatus.ACTIVE;
+ // Check if watermark has caught up
+ if (subpartitionStatus.watermark >= lastOutputWatermark) {
+ markWatermarkAligned(subpartitionStatus);
+ }
+ // Update and emit global watermark status
+ tryEmitNewGlobalWatermarkStatus(output);
+ // A subpartition that just realigned may hold the new minimum
of the aligned set,
+ // for instance when it is the only aligned subpartition. Its
watermark has to be
+ // emitted now, otherwise it stalls until an even larger
watermark arrives on that
+ // subpartition. This must follow the ACTIVE status above so
downstream inputs do
+ // not drop the watermark while they still consider this input
idle.
+ tryEmitNewWatermark(output);
+ } else if (watermarkStatus.isFinished()) {
+ subpartitionStatus.watermarkStatus = WatermarkStatus.FINISHED;
+ markWatermarkUnaligned(subpartitionStatus);
+ // Update and emit global watermark status
+ tryEmitNewGlobalWatermarkStatus(output);
+ } else {
+ throw new IllegalStateException(
+ "Invalid IDLE -> "
+ + watermarkStatus
+ + " transition for channel "
+ + channelIndex
+ + " subpartition "
+ + subpartitionIndexes[channelIndex]);
}
+ } else if (currentSubpartitionStatus.isFinished()) {
+ LOG.debug(
+ "Channel {} subpartition {} is in FINISHED state. Ignoring
transition to {}.",
+ channelIndex,
+ subpartitionIndexes[channelIndex],
+ watermarkStatus);
+ } else {
+ throw new IllegalStateException(
+ "Invalid status transition for channel "
+ + channelIndex
+ + " subpartition "
+ + subpartitionIndexes[channelIndex]
+ + ": currentStatus="
+ + currentSubpartitionStatus
+ + ", newStatus="
+ + watermarkStatus);
+ }
+ }
+
+ // Calculate and emit new watermark if it progresses.
+ private void tryEmitNewWatermark(DataOutput<?> output) throws Exception {
+ Long newWatermark = calculateWatermarkByAggregationRules();
+ if (newWatermark != null && newWatermark > lastOutputWatermark) {
+ lastOutputWatermark = newWatermark;
+ output.emitWatermark(new Watermark(lastOutputWatermark));
+ }
+ }
+
+ // Calculate and emit new operator-level (global) status.
+ private void tryEmitNewGlobalWatermarkStatus(DataOutput<?> output) throws
Exception {
+ WatermarkStatus newGlobalStatus = calculateGlobalWatermarkStatus();
+ if (!newGlobalStatus.equals(lastOutputWatermarkStatus)) {
+ lastOutputWatermarkStatus = newGlobalStatus;
+ output.emitWatermarkStatus(newGlobalStatus);
+ }
+ }
- // if the valve was previously marked to be idle, mark it as
active and output an active
- // stream status because at least one of the subpartitions is now
active
- if (lastOutputWatermarkStatus.isIdle()) {
- lastOutputWatermarkStatus = WatermarkStatus.ACTIVE;
- output.emitWatermarkStatus(lastOutputWatermarkStatus);
+ /**
+ * Calculates watermark based on the clear aggregation rules.
+ *
+ * <ol>
+ * <li>If there are ACTIVE subpartitions: watermark =
min(active_subpartitions)
+ * <li>Else if there are IDLE subpartitions: watermark =
max(idle_subpartitions)
+ * <li>Else (all subpartitions FINISHED): watermark = Long.MAX_VALUE if
all have received it
+ * </ol>
+ */
+ private Long calculateWatermarkByAggregationRules() {
Review Comment:
Have you considered treating a subpartition that received `MAX_WATERMARK` as
finished inside the valve? It can't advance anymore, and locally that fixes the
ticket's scenario without a new status.
--
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]
