This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch release-1.20 in repository https://gitbox.apache.org/repos/asf/flink.git
commit c6825b2ffb7200171317645a3640366599060c64 Author: Martijn Visser <[email protected]> AuthorDate: Thu Jul 2 11:29:31 2026 +0200 [FLINK-39481][task] Execute deferrable mails when finishing an operator Since FLINK-35528 the continuation mail of an interrupted timer-firing chain is deferrable and hidden from tryYield(). The finish drain in StreamOperatorWrapper therefore skipped it, so a task reaching end of input mid-firing forwarded EndOfData before the pending watermark, and downstream operators discarded the state of windows that only fire on that watermark. Restore the pre-FLINK-35528 behaviour by draining with a new yield variant that matches deferrable mails by their original priority, used only in that place. Generated-by: Claude Code (Fable 5) (cherry picked from commit 83055aead2ead294715a59b97fa5b0dc1dd2453c) --- .../runtime/tasks/StreamOperatorWrapper.java | 14 ++- .../streaming/runtime/tasks/mailbox/Mail.java | 5 + .../runtime/tasks/mailbox/MailboxExecutorImpl.java | 13 ++- .../runtime/tasks/mailbox/TaskMailbox.java | 11 ++ .../runtime/tasks/mailbox/TaskMailboxImpl.java | 23 +++-- ...nalignedCheckpointsInterruptibleTimersTest.java | 111 +++++++++++++++++++++ .../runtime/tasks/StreamOperatorWrapperTest.java | 18 ++++ .../tasks/StreamTaskMailboxTestHarness.java | 10 ++ .../tasks/mailbox/MailboxExecutorImplTest.java | 26 +++++ 9 files changed, 221 insertions(+), 10 deletions(-) diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java index 6be1e5cf0c1..4f300ac8974 100755 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java @@ -27,6 +27,7 @@ import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2; import org.apache.flink.streaming.api.operators.BoundedMultiInput; import org.apache.flink.streaming.api.operators.BoundedOneInput; import org.apache.flink.streaming.api.operators.StreamOperator; +import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl; import javax.annotation.Nonnull; @@ -39,6 +40,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import static org.apache.flink.util.Preconditions.checkNotNull; +import static org.apache.flink.util.Preconditions.checkState; /** * This class handles the finish, endInput and other related logic of a {@link StreamOperator}. It @@ -55,7 +57,7 @@ public class StreamOperatorWrapper<OUT, OP extends StreamOperator<OUT>> { @SuppressWarnings("OptionalUsedAsFieldOrParameterType") private final Optional<ProcessingTimeService> processingTimeService; - private final MailboxExecutor mailboxExecutor; + private final MailboxExecutorImpl mailboxExecutor; private final boolean isHead; @@ -73,7 +75,11 @@ public class StreamOperatorWrapper<OUT, OP extends StreamOperator<OUT>> { this.wrapped = checkNotNull(wrapped); this.processingTimeService = checkNotNull(processingTimeService); - this.mailboxExecutor = checkNotNull(mailboxExecutor); + checkState( + mailboxExecutor instanceof MailboxExecutorImpl, + "The mailbox executor must be a MailboxExecutorImpl, so that deferrable mails" + + " can be executed when finishing the operator (FLINK-39481)."); + this.mailboxExecutor = (MailboxExecutorImpl) mailboxExecutor; this.isHead = isHead; } @@ -183,7 +189,9 @@ public class StreamOperatorWrapper<OUT, OP extends StreamOperator<OUT>> { // run the mailbox processing loop until all operations are finished while (!finishedFuture.isDone()) { - while (mailboxExecutor.tryYield()) {} + // FLINK-39481: also execute deferrable mails, so that a deferred watermark + // continuation completes and forwards the watermark before EndOfData is sent + while (mailboxExecutor.tryYieldIgnoringDeferrable()) {} // we wait a little bit to avoid unnecessary CPU occupation due to empty loops, // such as when all mails of the operator have been processed but the closed future diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java index 6afd8918e3d..8335533606e 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java @@ -86,6 +86,11 @@ public class Mail { return mailOptions.isDeferrable() ? TaskMailbox.MIN_PRIORITY : priority; } + /** The priority the mail was created with, ignoring the deferrable demotion. */ + public int getPriorityIgnoringDeferrable() { + return priority; + } + public void tryCancel(boolean mayInterruptIfRunning) { if (runnable instanceof Future) { ((Future<?>) runnable).cancel(mayInterruptIfRunning); diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java index 2987ee21cca..a60e82c4c6d 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java @@ -97,7 +97,18 @@ public final class MailboxExecutorImpl implements MailboxExecutor { @Override public boolean tryYield() { - Optional<Mail> optionalMail = mailbox.tryTake(priority); + return runIfPresent(mailbox.tryTake(priority)); + } + + /** + * Same as {@link #tryYield()}, but also executes deferrable mails (see {@link + * MailboxExecutor.MailOptions#deferrable()}). + */ + public boolean tryYieldIgnoringDeferrable() { + return runIfPresent(mailbox.tryTakeIgnoringDeferrable(priority)); + } + + private static boolean runIfPresent(Optional<Mail> optionalMail) { if (optionalMail.isPresent()) { try { optionalMail.get().run(); diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java index 6915d4b6d1a..0b40c4cbaf2 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java @@ -107,6 +107,17 @@ public interface TaskMailbox { */ Optional<Mail> tryTake(int priority); + /** + * Same as {@link #tryTake(int)}, but deferrable mails are matched against the priority they + * were created with instead of {@link #MIN_PRIORITY} (see {@link + * MailboxExecutor.MailOptions#deferrable()}). + * + * <p>Must be called from the mailbox thread ({@link #isMailboxThread()}. + * + * @throws IllegalStateException if mailbox is already closed. + */ + Optional<Mail> tryTakeIgnoringDeferrable(int priority); + /** * This method returns the oldest mail from the mailbox (head of queue) or blocks until a mail * is available. diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java index 00d19220e6e..38c77b442ba 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java @@ -109,9 +109,18 @@ public class TaskMailboxImpl implements TaskMailbox { @Override public Optional<Mail> tryTake(int priority) { + return tryTake(priority, false); + } + + @Override + public Optional<Mail> tryTakeIgnoringDeferrable(int priority) { + return tryTake(priority, true); + } + + private Optional<Mail> tryTake(int priority, boolean ignoreDeferrable) { checkIsMailboxThread(); checkTakeStateConditions(); - Mail head = takeOrNull(batch, priority); + Mail head = takeOrNull(batch, priority, ignoreDeferrable); if (head != null) { return Optional.of(head); } @@ -121,7 +130,7 @@ public class TaskMailboxImpl implements TaskMailbox { final ReentrantLock lock = this.lock; lock.lock(); try { - final Mail value = takeOrNull(queue, priority); + final Mail value = takeOrNull(queue, priority, ignoreDeferrable); if (value == null) { return Optional.empty(); } @@ -136,7 +145,7 @@ public class TaskMailboxImpl implements TaskMailbox { public @Nonnull Mail take(int priority) throws InterruptedException, IllegalStateException { checkIsMailboxThread(); checkTakeStateConditions(); - Mail head = takeOrNull(batch, priority); + Mail head = takeOrNull(batch, priority, false); if (head != null) { return head; } @@ -144,7 +153,7 @@ public class TaskMailboxImpl implements TaskMailbox { lock.lockInterruptibly(); try { Mail headMail; - while ((headMail = takeOrNull(queue, priority)) == null) { + while ((headMail = takeOrNull(queue, priority, false)) == null) { // to ease debugging notEmpty.await(1, TimeUnit.SECONDS); } @@ -225,7 +234,7 @@ public class TaskMailboxImpl implements TaskMailbox { // ------------------------------------------------------------------------------------------------------------------ @Nullable - private Mail takeOrNull(Deque<Mail> queue, int priority) { + private Mail takeOrNull(Deque<Mail> queue, int priority, boolean ignoreDeferrable) { if (queue.isEmpty()) { return null; } @@ -233,7 +242,9 @@ public class TaskMailboxImpl implements TaskMailbox { Iterator<Mail> iterator = queue.iterator(); while (iterator.hasNext()) { Mail mail = iterator.next(); - if (mail.getPriority() >= priority) { + int mailPriority = + ignoreDeferrable ? mail.getPriorityIgnoringDeferrable() : mail.getPriority(); + if (mailPriority >= priority) { iterator.remove(); return mail; } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java index a5e3712ec76..2defc02ad69 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java @@ -21,6 +21,11 @@ package org.apache.flink.streaming.runtime.io.checkpointing; import org.apache.flink.api.common.operators.MailboxExecutor; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.common.typeutils.base.StringSerializer; +import org.apache.flink.api.java.functions.KeySelector; +import org.apache.flink.configuration.CheckpointingOptions; +import org.apache.flink.runtime.io.network.api.EndOfData; +import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent; +import org.apache.flink.runtime.io.network.api.StopMode; import org.apache.flink.streaming.api.graph.StreamConfig; import org.apache.flink.streaming.api.operators.AbstractStreamOperator; import org.apache.flink.streaming.api.operators.InternalTimer; @@ -135,6 +140,53 @@ class UnalignedCheckpointsInterruptibleTimersTest { } } + /** + * FLINK-39481: an interrupted timer-firing chain defers its continuation (and the downstream + * watermark emission) to a deferrable mail. When EndOfData arrives at that point, finishing the + * operator must still complete the deferred work, otherwise downstream operators never receive + * the watermark and lose the state of windows that only fire on it. + */ + @Test + void testDeferredWatermarkIsEmittedBeforeEndOfData() throws Exception { + final Instant windowEnd = Instant.ofEpochMilli(1000L); + final int numKeys = 2; + + try (final StreamTaskMailboxTestHarness<String> harness = + new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, Types.STRING) + .setCollectNetworkEvents() + .setKeyType(Types.STRING) + .addInput(Types.STRING, 1, (KeySelector<String, String>) value -> value) + .setupOperatorChain( + SimpleOperatorFactory.of(new TimerPerElement(windowEnd))) + .name("first") + .finishForSingletonOperatorChain(StringSerializer.INSTANCE) + .build()) { + harness.setAutoProcess(false); + for (int keyIdx = 0; keyIdx < numKeys; keyIdx++) { + harness.processElement(new StreamRecord<>(String.format("key-%d", keyIdx))); + } + harness.processAll(); + + harness.processElement(asWatermark(windowEnd)); + harness.processEvent(new EndOfData(StopMode.DRAIN), 0, 0); + harness.processEvent(EndOfPartitionEvent.INSTANCE, 0, 0); + + // the mailbox loop lets the default action process EndOfData at a batch boundary, + // while the re-deferred watermark continuation is still pending; the single-step + // processing of processAll() cannot reach that interleaving + harness.runMailboxLoop(); + + assertThat(harness.getOutput()) + .containsExactly( + asFiredRecord("key-0"), + asMailRecord("key-0"), + asFiredRecord("key-1"), + asMailRecord("key-1"), + asWatermark(windowEnd), + new EndOfData(StopMode.DRAIN)); + } + } + private static Watermark asWatermark(Instant timestamp) { return new Watermark(timestamp.toEpochMilli()); } @@ -227,4 +279,63 @@ class UnalignedCheckpointsInterruptibleTimersTest { return new MultipleTimersAtTheSameTimestamp(copy); } } + + /** + * Registers a timer for the current key on every element; every fired timer enqueues a mail, so + * that firing is interrupted after each timer. + */ + private static class TimerPerElement extends AbstractStreamOperator<String> + implements OneInputStreamOperator<String, String>, + Triggerable<String, String>, + YieldingOperator<String> { + + private final Instant timerTimestamp; + private transient @Nullable MailboxExecutor mailboxExecutor; + private transient @Nullable MailboxWatermarkProcessor watermarkProcessor; + + TimerPerElement(Instant timerTimestamp) { + this.timerTimestamp = timerTimestamp; + } + + @Override + public void setMailboxExecutor(MailboxExecutor mailboxExecutor) { + this.mailboxExecutor = mailboxExecutor; + } + + @Override + public void open() throws Exception { + super.open(); + if (getTimeServiceManager().isPresent()) { + this.watermarkProcessor = + new MailboxWatermarkProcessor( + output, mailboxExecutor, getTimeServiceManager().get()); + } + } + + @Override + public void processWatermark(Watermark mark) throws Exception { + if (watermarkProcessor == null) { + super.processWatermark(mark); + } else { + watermarkProcessor.emitWatermarkInsideMailbox(mark); + } + } + + @Override + public void processElement(StreamRecord<String> element) { + final InternalTimerService<String> timers = + getInternalTimerService("timers", StringSerializer.INSTANCE, this); + timers.registerEventTimeTimer("window", timerTimestamp.toEpochMilli()); + } + + @Override + public void onEventTime(InternalTimer<String, String> timer) throws Exception { + mailboxExecutor.execute( + () -> output.collect(asMailRecord(timer.getKey())), "mail-" + timer.getKey()); + output.collect(asFiredRecord(timer.getKey())); + } + + @Override + public void onProcessingTime(InternalTimer<String, String> timer) throws Exception {} + } } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java index 5947a351f2c..e0eaafd5bb7 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java @@ -45,6 +45,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -147,6 +148,23 @@ class StreamOperatorWrapperTest { .containsExactlyElementsOf(expected.subList(2, expected.size())); } + @Test + void testFinishExecutesDeferrableMails() throws Exception { + // FLINK-39481: a deferrable mail (e.g. a deferred watermark continuation) must be + // executed while finishing the operator, before EndOfData is sent downstream + MailboxExecutor localExecutor = + containingTask.getMailboxExecutorFactory().createExecutor(0); + AtomicBoolean deferrableMailExecuted = new AtomicBoolean(); + localExecutor.execute( + MailboxExecutor.MailOptions.deferrable(), + () -> deferrableMailExecuted.set(true), + "deferrable mail"); + + operatorWrappers.get(0).finish(containingTask.getActionExecutor(), StopMode.DRAIN); + + assertThat(deferrableMailExecuted.get()).isTrue(); + } + @Test void testFinishingOperatorWithException() { AbstractStreamOperator<Void> streamOperator = diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java index 4e78f4db011..6925b04f380 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java @@ -136,6 +136,16 @@ public class StreamTaskMailboxTestHarness<OUT> implements AutoCloseable { return false; } + /** + * Runs the production mailbox loop until it is suspended or all actions are completed. Unlike + * {@link #processAll()}, this preserves the production interleaving of mails and the default + * action: the default action can process input at a mail-batch boundary while mails enqueued + * during the batch are still pending. + */ + public void runMailboxLoop() throws Exception { + streamTask.mailboxProcessor.runMailboxLoop(); + } + public MailboxExecutor getExecutor(int priority) { return streamTask.getMailboxExecutorFactory().createExecutor(priority); } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java index 1b6fddc7d91..63a614c1269 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java @@ -130,6 +130,32 @@ class MailboxExecutorImplTest { assertThat(deferrableMailExecuted.get()).isTrue(); } + @Test + void testTryYieldIgnoringDeferrable() throws Exception { + int priority = 42; + MailboxExecutorImpl localExecutor = + (MailboxExecutorImpl) mailboxProcessor.getMailboxExecutor(priority); + + AtomicBoolean deferrableMailExecuted = new AtomicBoolean(); + + localExecutor.execute( + MailboxExecutor.MailOptions.deferrable(), + () -> deferrableMailExecuted.set(true), + "deferrable mail"); + // the deferrable mail keeps its original priority, so it stays hidden from executors + // with a higher priority + assertThat( + ((MailboxExecutorImpl) mailboxProcessor.getMailboxExecutor(priority + 1)) + .tryYieldIgnoringDeferrable()) + .isFalse(); + assertThat(deferrableMailExecuted.get()).isFalse(); + assertThat(localExecutor.tryYield()).isFalse(); + assertThat(deferrableMailExecuted.get()).isFalse(); + + assertThat(localExecutor.tryYieldIgnoringDeferrable()).isTrue(); + assertThat(deferrableMailExecuted.get()).isTrue(); + } + @Test void testClose() throws Exception { final TestRunnable yieldRun = new TestRunnable();
