This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch release-2.3 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 017c72e72b37eb801277d3265816387b03143239 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 +++-- .../runtime/tasks/StreamOperatorWrapperTest.java | 18 ++++ .../tasks/mailbox/MailboxExecutorImplTest.java | 26 ++++++ ...nalignedCheckpointsInterruptibleTimersTest.java | 101 +++++++++++++++++++++ .../tasks/StreamTaskMailboxTestHarness.java | 10 ++ 9 files changed, 211 insertions(+), 10 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java index 6be1e5cf0c1..4f300ac8974 100755 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapper.java +++ b/flink-runtime/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-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java index 016ae6be2cf..b68c8483eff 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java @@ -103,6 +103,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-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java index 4fef813b109..2ad3d7b7e86 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java @@ -98,7 +98,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-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java index df3a6f065de..62745e8eaa7 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java @@ -108,6 +108,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-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java index 56188c3167f..18bff55c588 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java @@ -115,12 +115,21 @@ 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(); moveUrgentMailsToBatchIfNeeded(true); - Mail head = takeOrNull(batch, priority); + Mail head = takeOrNull(batch, priority, ignoreDeferrable); if (head != null) { return Optional.of(head); } @@ -130,7 +139,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(); } @@ -148,7 +157,7 @@ public class TaskMailboxImpl implements TaskMailbox { moveUrgentMailsToBatchIfNeeded(true); - Mail head = takeOrNull(batch, priority); + Mail head = takeOrNull(batch, priority, false); if (head != null) { return head; } @@ -156,7 +165,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); } @@ -333,7 +342,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; } @@ -341,7 +350,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-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java index 09c5c3100c7..99a37642102 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamOperatorWrapperTest.java +++ b/flink-runtime/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; @@ -146,6 +147,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-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java index 1b6fddc7d91..63a614c1269 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java +++ b/flink-runtime/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(); 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 d116130d9f7..47bfbe84592 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,8 +21,12 @@ 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.configuration.ReadableConfig; +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; @@ -147,6 +151,58 @@ 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) + .addJobConfig( + CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(1)) + .addJobConfig(CheckpointingOptions.ENABLE_UNALIGNED, true) + .addJobConfig( + CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS, true) + .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()); } @@ -223,4 +279,49 @@ 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; + + TimerPerElement(Instant timerTimestamp) { + this.timerTimestamp = timerTimestamp; + } + + @Override + public boolean useInterruptibleTimers(ReadableConfig config) { + return true; + } + + @Override + public void setMailboxExecutor(MailboxExecutor mailboxExecutor) { + super.setMailboxExecutor(mailboxExecutor); + this.mailboxExecutor = mailboxExecutor; + } + + @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/StreamTaskMailboxTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskMailboxTestHarness.java index 4e733481536..be1934336c9 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); }
