You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2026/07/06 15:48:30 UTC
(flink) branch release-2.3 updated (f73dec412d7 -> fcd20f29b87)
This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a change to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
from f73dec412d7 [FLINK-40062][table] Allow `BridgingSqlAggFunction` being instantiated from `BuiltInFunctionDefinitions`, fix `$WELFORD_M2$1` function deserialization (#28647)
new 017c72e72b3 [FLINK-39481][task] Execute deferrable mails when finishing an operator
new fcd20f29b87 [FLINK-39481][table] Add stress ITCase for watermarks deferred at end of input
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../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 +
.../sql/WindowDistinctAggregateStressITCase.java | 217 +++++++++++++++++++++
10 files changed, 428 insertions(+), 10 deletions(-)
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/WindowDistinctAggregateStressITCase.java
(flink) 01/02: [FLINK-39481][task] Execute deferrable mails when finishing an operator
Posted by ma...@apache.org.
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 <29...@users.noreply.github.com>
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);
}
(flink) 02/02: [FLINK-39481][table] Add stress ITCase for watermarks deferred at end of input
Posted by ma...@apache.org.
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 fcd20f29b87d09de2d55ee44e8c32224d265bda8
Author: Martijn Visser <29...@users.noreply.github.com>
AuthorDate: Fri Jul 3 15:07:00 2026 +0200
[FLINK-39481][table] Add stress ITCase for watermarks deferred at end of input
Exercises the full SQL pipeline (two-phase split-distinct window aggregation
on RocksDB) under backpressure with the configuration in which all CI
failures of WindowDistinctAggregateITCase occurred, comparing repeated
checkpoint/failure/restore/end-of-input rounds against a golden run within a
fixed time budget. Without the FLINK-39481 fix this loses the trailing
window family in roughly one of three executions.
Generated-by: Claude Code (Fable 5)
(cherry picked from commit bbc688cef98bab45a8885efc4b3e67bb47b89cc9)
---
.../sql/WindowDistinctAggregateStressITCase.java | 217 +++++++++++++++++++++
1 file changed, 217 insertions(+)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/WindowDistinctAggregateStressITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/WindowDistinctAggregateStressITCase.java
new file mode 100644
index 00000000000..f7ca534fcee
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/WindowDistinctAggregateStressITCase.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.runtime.stream.sql;
+
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.core.execution.CheckpointingMode;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.runtime.utils.FailingCollectionSource;
+import org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase;
+import org.apache.flink.types.Row;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Stress test for FLINK-39481: with unaligned checkpoints and interruptible timers, a watermark
+ * whose timer-firing chain was interrupted must still be forwarded downstream before EndOfData,
+ * otherwise windows that only fire on that watermark lose their state. The deterministic coverage
+ * lives in {@code UnalignedCheckpointsInterruptibleTimersTest}; this test additionally exercises
+ * the full SQL pipeline (two-phase split-distinct window aggregation on RocksDB) under
+ * backpressure, where the interruption races end of input, and compares against a golden run of the
+ * identical topology.
+ */
+class WindowDistinctAggregateStressITCase extends StreamingWithStateTestBase {
+
+ private static final Duration TEST_DURATION = Duration.ofSeconds(20);
+ private static final long SINK_OPEN_DELAY_MS = 300;
+ private static final long SINK_ROW_DELAY_MS = 1;
+
+ private static final int ROWS_PER_SECOND = 20;
+ private static final int LAST_SECOND = 34;
+ private static final int NUM_KEYS = 200;
+ private static final int STRING_CARDINALITY = 17;
+
+ private int sinkCount;
+
+ WindowDistinctAggregateStressITCase() {
+ super(StreamingWithStateTestBase.ROCKSDB_BACKEND());
+ }
+
+ @BeforeEach
+ @Override
+ public void before() {
+ super.before();
+ env().enableCheckpointing(100, CheckpointingMode.EXACTLY_ONCE);
+ final Configuration configuration = new Configuration();
+ configuration.set(RestartStrategyOptions.RESTART_STRATEGY, "fixeddelay");
+ configuration.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 1);
+ configuration.set(
+ RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY, Duration.ofMillis(0));
+ // the configuration under which every FLINK-39481 CI failure occurred: every checkpoint
+ // is unaligned and may interrupt firing timers; also set on the table config, which
+ // builds the job configuration for table programs
+ configuration.set(CheckpointingOptions.ENABLE_UNALIGNED, true);
+ configuration.set(CheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, Duration.ZERO);
+ configuration.set(CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS, true);
+ configuration.set(CheckpointingOptions.FILE_MERGING_ENABLED, true);
+ configuration.set(ExecutionOptions.SNAPSHOT_COMPRESSION, true);
+ configuration.setString("state.backend.rocksdb.use-ingest-db-restore-mode", "true");
+ env().configure(configuration, Thread.currentThread().getContextClassLoader());
+ configuration.toMap().forEach((k, v) -> tEnv().getConfig().set(k, v));
+
+ final String dataId = TestValuesTableFactory.registerData(generatedRows());
+ tEnv().executeSql(
+ "CREATE TABLE T1 (\n"
+ + " `ts` STRING,\n"
+ + " `int` INT,\n"
+ + " `double` DOUBLE,\n"
+ + " `float` FLOAT,\n"
+ + " `bigdec` DECIMAL(10, 2),\n"
+ + " `string` STRING,\n"
+ + " `name` STRING,\n"
+ + " `rowtime` AS TO_TIMESTAMP(`ts`),\n"
+ + " WATERMARK for `rowtime` AS `rowtime` - INTERVAL '1' SECOND\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'data-id' = '"
+ + dataId
+ + "',\n"
+ + " 'failing-source' = 'true'\n"
+ + ")");
+
+ tEnv().getConfig()
+ .set(OptimizerConfigOptions.TABLE_OPTIMIZER_DISTINCT_AGG_SPLIT_ENABLED, true);
+ tEnv().createTemporarySystemFunction("throttle", ThrottlingFunction.class);
+ }
+
+ /** Many keys per 5s slice, so that each slice boundary fires a long chain of timers. */
+ private static List<Row> generatedRows() {
+ final List<Row> rows = new ArrayList<>();
+ for (int sec = 0; sec <= LAST_SECOND; sec++) {
+ for (int i = 0; i < ROWS_PER_SECOND; i++) {
+ final int idx = sec * ROWS_PER_SECOND + i;
+ rows.add(
+ Row.of(
+ String.format("2020-10-10 00:00:%02d", sec),
+ idx % 10,
+ (double) (idx % 7),
+ (float) (idx % 5),
+ new BigDecimal(String.format("%d.%02d", idx % 9, idx % 100)),
+ idx % 19 == 0 ? null : "s" + (idx % STRING_CARDINALITY),
+ idx % 23 == 0 ? null : String.format("k%03d", idx % NUM_KEYS)));
+ }
+ }
+ return rows;
+ }
+
+ @Test
+ void testCumulateWindowRollupUnderBackpressure() throws Exception {
+ // golden run: identical topology, but the source does not fail (failedBefore is true)
+ final List<String> expected = runQuery();
+ assertThat(expected).isNotEmpty();
+
+ final Deadline deadline = Deadline.fromNow(TEST_DURATION);
+ do {
+ // each round: checkpoint under backpressure, artificial failure, restore, end of
+ // input
+ FailingCollectionSource.reset();
+ final List<String> actual = runQuery();
+ assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
+ } while (deadline.hasTimeLeft());
+ }
+
+ private List<String> runQuery() throws Exception {
+ final String sinkName = "sink_" + sinkCount++;
+ tEnv().executeSql(
+ "CREATE TABLE "
+ + sinkName
+ + " (\n"
+ + " g BIGINT,\n"
+ + " name STRING,\n"
+ + " window_start TIMESTAMP(3),\n"
+ + " window_end TIMESTAMP(3),\n"
+ + " cnt BIGINT,\n"
+ + " sum_dec DECIMAL(38, 2),\n"
+ + " max_d DOUBLE,\n"
+ + " min_f FLOAT,\n"
+ + " distinct_cnt BIGINT\n"
+ + ") WITH ('connector' = 'values')");
+ tEnv().executeSql(
+ "INSERT INTO "
+ + sinkName
+ + "\n"
+ + "SELECT\n"
+ + " throttle(GROUPING_ID(`name`)),\n"
+ + " `name`,\n"
+ + " window_start,\n"
+ + " window_end,\n"
+ + " COUNT(*),\n"
+ + " SUM(`bigdec`),\n"
+ + " MAX(`double`),\n"
+ + " MIN(`float`),\n"
+ + " COUNT(DISTINCT `string`)\n"
+ + "FROM TABLE(\n"
+ + " CUMULATE(\n"
+ + " TABLE T1,\n"
+ + " DESCRIPTOR(rowtime),\n"
+ + " INTERVAL '5' SECOND,\n"
+ + " INTERVAL '15' SECOND))\n"
+ + "GROUP BY ROLLUP(`name`), window_start, window_end")
+ .await();
+ return TestValuesTableFactory.getResultsAsStrings(sinkName);
+ }
+
+ /**
+ * Chained in front of the sink, so it lives inside the final aggregation task: the open delay
+ * keeps that task INITIALIZING while the source emits (checkpoint 1 then captures the records
+ * as unaligned channel state), the per-row delay sustains backpressure while checkpoints race
+ * the end-of-input watermark.
+ */
+ public static class ThrottlingFunction extends ScalarFunction {
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ Thread.sleep(SINK_OPEN_DELAY_MS);
+ }
+
+ public Long eval(Long value) {
+ try {
+ Thread.sleep(SINK_ROW_DELAY_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return value;
+ }
+ }
+}