[FLINK-40504][runtime] Re-activate idle FLIP-27 outputs when they resume

[Admaing]

What is the purpose of the change

A FLIP-27 source that has been announced as IDLE downstream stays IDLE after it resumes, unless its
next watermark strictly advances the combined watermark. Two mechanisms compound:

1. WatermarkOutputMultiplexer.updateCombinedWatermark() reports the combined idle state via
   underlyingOutput.markIdle(), but has no counterpart for the transition back to active; the
   per-output markActive() implementations only clear internal partial state.
2. WatermarkToDataOutput.emitWatermark() returns at the monotonicity guard
   (newWatermark <= maxWatermarkSoFar) before reaching markActiveInternally(), although the
   WatermarkOutput contract states that emitting a watermark implicitly marks the stream active.

A resumed split with backlog whose watermarks are <= the maximum flushed while all splits were
idle therefore reaches the underlying output with neither a watermark nor an ACTIVE status, and its
records are dropped as late downstream. Every sibling path re-activates eagerly
(StatusWatermarkValve, legacy StreamSourceContexts, the table WatermarkAssignerOperator, and
DataStream V2); the FLIP-27 multiplexer path was the only one without a re-activation signal.

This also covers the new-split special case of the same defect (FLINK-22926): registering a new
output while the combined status is idle now re-activates it right away, instead of only on the
next periodic emit.

Brief change log

• WatermarkOutputMultiplexer keeps track of the idle state last reported to the underlying output
  and reports the idle -> active transition via markActive().
• A watermark on a previously idle multiplexed output triggers the combined update, so the
  re-activation is propagated eagerly instead of only on the next periodic emit.
• registerNewOutput() updates and reports the combined status right away, so registering a new
  split while the source is idle marks the underlying output active immediately (FLINK-22926).
• WatermarkToDataOutput.emitWatermark() marks the output active before the monotonicity guard.
• Tests: 4 new WatermarkOutputMultiplexerTest cases, 1 new WatermarkToDataOutputTest case and
  2 new end-to-end SourceOperatorEventTimeTest cases.

Verifying this change

This change added tests and can be verified as follows:

• mvn -pl flink-core -am -Dtest=WatermarkOutputMultiplexerTest test
• mvn -pl flink-runtime -am -Dtest='WatermarkToDataOutputTest,SourceOperatorEventTimeTest' test
• All 7 new tests were verified to fail without the corresponding production change and to pass
  with it (red/green verification by reverting the three production files, and separately by
  reverting the registerNewOutput() update).
• Broader regression over all watermark-related test classes in flink-core-api, flink-core and
  flink-runtime (20 classes, 137 tests) passes on a clean build.
• SourceOperatorEventTimeTest#testResumingSplitWithoutAdvancingWatermarkEmitsActive reproduces the
  scenario from the ticket: all splits go idle, the combined watermark is flushed to the maximum, a
  split resumes with backlog whose watermarks do not advance the combined watermark, and the edge
  must emit ACTIVE.
• SourceOperatorEventTimeTest#testRegisteringNewSplitWhileIdleEmitsActive covers FLINK-22926. It
  deliberately does not advance the processing time service, because a periodic watermark emit
  would re-activate the output lazily as well and would mask the immediate re-activation under test.

Notes for reviewers:

• The only TimestampsAndWatermarks.WatermarkUpdateListener implementation is SourceOperator.
  emitWatermark() now calls markActiveInternally() before updateCurrentEffectiveWatermark()
  for advancing watermarks (the downstream event order ACTIVE -> watermark is unchanged).
  updateIdle() only writes a field that checkWatermarkAlignment() does not read, so the swapped
  listener callback order has no observable effect.
• unregisterOutput() still does not update the combined status, so removing the last active output
  while others are idle is only reflected on the next periodic emit. This is a pre-existing,
  bounded-delay issue of the same family and is deliberately left out of this change.
• This is complementary to [FLINK-40499][runtime] Emit ACTIVE before final MAX_WATERMARK at drain #29037 (FLINK-40499), which re-activates at drain time in the stream
  tasks; this PR fixes the root cause and does not touch those files. It also covers the new-split
  case described in FLINK-22926, which can be closed together with this change.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API: no (WatermarkOutputMultiplexer and WatermarkToDataOutput are @Internal)
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes, minimally —
  ImmediateOutput.emitWatermark() gains one field read and a branch; the combined update (which
  iterates all splits) still only runs on a watermark advance, an idle -> active transition, or a
  registration
• Anything that affects deployment or recovery: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

• Yes (please specify the tool below)

Generated-by: DeepSeek Harness (deepseek-v4.1-flash)

[flinkbot]

CI report:

• ca8d250 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build