You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2026/04/22 13:49:44 UTC
(flink) branch master updated: [FLINK-39437][table] Support interruptible timers in PTFs (#27962)
This is an automated email from the ASF dual-hosted git repository.
fhueske pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5106d15e83d [FLINK-39437][table] Support interruptible timers in PTFs (#27962)
5106d15e83d is described below
commit 5106d15e83d4497824c33b71fd0021846bd1ec09
Author: Fabian Hueske <fh...@confluent.io>
AuthorDate: Wed Apr 22 15:49:37 2026 +0200
[FLINK-39437][table] Support interruptible timers in PTFs (#27962)
Override `useInterruptibleTimers()` in `AbstractProcessTableOperator` to
return `true`, activating the `MailboxWatermarkProcessor` for PTF operators.
This allows timer firing to be interrupted between mailbox iterations,
improving throughput by not blocking mailbox processing during large timer
batches.
Also removes `currentWatermark` from `ProcessTableRunner` and redirect reads to
`AbstractStreamOperatorV2.combinedWatermark`.
Add a test (`ProcessSetTableOperatorInterruptibleTimersTest`) to assert correct behavior
of PTF framework when timer processing is interrupted.
---
.../runtime/generated/ProcessTableRunner.java | 9 -
.../process/AbstractProcessTableOperator.java | 10 +-
.../operators/process/ProcessRowTableOperator.java | 13 +-
...essSetTableOperatorInterruptibleTimersTest.java | 352 +++++++++++++++++++++
4 files changed, 371 insertions(+), 13 deletions(-)
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
index b90a201649e..5477fc4523f 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/ProcessTableRunner.java
@@ -65,7 +65,6 @@ public abstract class ProcessTableRunner extends AbstractRichFunction {
// Current time
private long tableWatermark = Long.MIN_VALUE;
- private long currentWatermark = Long.MIN_VALUE;
private @Nullable Long rowtime;
private @Nullable StringData timerName;
@@ -131,10 +130,6 @@ public abstract class ProcessTableRunner extends AbstractRichFunction {
tableWatermark = Long.MIN_VALUE;
}
- public void ingestCurrentWatermarkEvent(long watermark) {
- currentWatermark = watermark;
- }
-
public void clearAllState() {
Arrays.fill(stateCleared, true);
}
@@ -143,10 +138,6 @@ public abstract class ProcessTableRunner extends AbstractRichFunction {
stateCleared[statePos] = true;
}
- public long getCurrentWatermark() {
- return currentWatermark;
- }
-
public long getTableWatermark() {
return tableWatermark;
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
index cf3a6149af6..d3b58b5c886 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java
@@ -31,6 +31,7 @@ import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2;
@@ -145,11 +146,14 @@ public abstract class AbstractProcessTableOperator extends AbstractStreamOperato
FunctionUtils.openFunction(processTableRunner, DefaultOpenContext.INSTANCE);
}
+ @Override
+ public final boolean useInterruptibleTimers(ReadableConfig config) {
+ return true;
+ }
+
@Override
public void processWatermark(Watermark mark) throws Exception {
super.processWatermark(mark);
- // TODO this line has issues with interruptible timers, see FLINK-39437
- processTableRunner.ingestCurrentWatermarkEvent(mark.getTimestamp());
}
@Override
@@ -220,7 +224,7 @@ public abstract class AbstractProcessTableOperator extends AbstractStreamOperato
internalTimeContext.setTime(
processTableRunner.getTableWatermark(),
- processTableRunner.getCurrentWatermark(),
+ combinedWatermark.getCombinedWatermark(),
processTableRunner.getTime());
return (TimeContext<TimeType>)
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java
index 1964b2d2376..80528c1e63f 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.operators.process;
import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
@@ -82,11 +83,21 @@ public class ProcessRowTableOperator extends AbstractProcessTableOperator
0,
element.getValue(),
inputSemantics.timeColumn(),
- processTableRunner.getCurrentWatermark());
+ combinedWatermark.getCombinedWatermark());
}
processTableRunner.processEval();
}
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ // Advance combinedWatermark before super.processWatermark(mark): super.processWatermark()
+ // fires due timers, which read the watermark via timeContext(). V2's OneInputStreamOperator
+ // watermark path doesn't update combinedWatermark itself — only reportWatermark()
+ // (multi-input) does — so we do it here.
+ combinedWatermark.updateWatermark(0, mark.getTimestamp());
+ super.processWatermark(mark);
+ }
+
@Override
public void processWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception {
super.processWatermarkStatus(watermarkStatus, 1);
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
new file mode 100644
index 00000000000..be390ab5f55
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorInterruptibleTimersTest.java
@@ -0,0 +1,352 @@
+/*
+ * 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.runtime.operators.process;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.InternalTimer;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.MultipleInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.functions.ProcessTableFunction.TimeContext;
+import org.apache.flink.table.runtime.generated.HashFunction;
+import org.apache.flink.table.runtime.generated.ProcessTableRunner;
+import org.apache.flink.table.runtime.generated.RecordComparator;
+import org.apache.flink.table.runtime.generated.RecordEqualiser;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.utils.HandwrittenSelectorUtil;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that {@link ProcessSetTableOperator} drives {@link ProcessTableRunner} correctly when user
+ * timers are fired under unaligned checkpoints, both with and without interruptible timers.
+ *
+ * <p>A minimal {@link ProcessTableRunner} is used: {@link ProcessTableRunner#callEval()} emits a
+ * "record-$ts@wm$wm" row via {@code evalCollector} and registers an unnamed event-time timer at the
+ * row's event time (plus an additional named timer for the row at ts=1000); {@link
+ * ProcessTableRunner#callOnTimer()} emits a "fired-[name-]$ts@wm$wm" row via {@code
+ * onTimerCollector}. The operator is wrapped so that after each timer firing a non-deferrable
+ * "mail-[name-]$ts" mailbox mail is scheduled — the PTF user API does not expose the mailbox, so
+ * this is the only way to observe externally whether the mailbox is drained between timer firings.
+ *
+ * <p>With interruptible timers enabled, the mailbox is drained between successive timer firings, so
+ * each "fired" row is immediately followed by its "mail" row. With interruptible timers disabled,
+ * all timers in a batch fire first (and the watermark is emitted downstream) before any of the
+ * scheduled mails are processed. The watermark values on "record" and "fired" rows also validate
+ * that {@link TimeContext#currentWatermark()} advances consistently across input processing and
+ * timer firing.
+ */
+class ProcessSetTableOperatorInterruptibleTimersTest {
+
+ private static final DataType INPUT_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("k", DataTypes.BIGINT()),
+ DataTypes.FIELD("ts", DataTypes.TIMESTAMP(3)));
+
+ // Output layout is driven by PassPartitionKeysCollector / PassAllCollector in
+ // AbstractProcessTableOperator: [partition-key, PTF-emitted label, rowtime].
+ private static final DataType OUTPUT_TYPE =
+ DataTypes.ROW(
+ DataTypes.FIELD("k", DataTypes.BIGINT()),
+ DataTypes.FIELD("label", DataTypes.STRING()),
+ DataTypes.FIELD("ts", DataTypes.TIMESTAMP(3)));
+
+ private static final InternalTypeInfo<RowData> INPUT_TYPE_INFO =
+ InternalTypeInfo.of(INPUT_TYPE.getLogicalType());
+
+ private static final InternalTypeInfo<RowData> OUTPUT_TYPE_INFO =
+ InternalTypeInfo.of(OUTPUT_TYPE.getLogicalType());
+
+ private static final RowDataKeySelector KEY_SELECTOR =
+ HandwrittenSelectorUtil.getRowDataSelector(
+ new int[] {0}, INPUT_TYPE_INFO.toRowFieldTypes());
+
+ private static final long KEY = 42L;
+
+ private static final String NAMED_TIMER = "n";
+
+ @ParameterizedTest(name = "interruptibleTimers={0}")
+ @ValueSource(booleans = {true, false})
+ void testTimersThroughProcessTableRunner(boolean interruptibleTimers) throws Exception {
+ try (StreamTaskMailboxTestHarness<RowData> harness = createHarness(interruptibleTimers)) {
+ // Each input row registers one event-time timer at its own timestamp. The row at
+ // ts=1000 additionally registers a named timer at the same timestamp, so the first
+ // batch contains three unnamed firings and one named firing at identical timestamps
+ // (ts=1000) — exercising both timer services advancing together.
+ for (long ts = 1000L; ts <= 3000L; ts += 1000L) {
+ harness.processElement(
+ new StreamRecord<>(
+ GenericRowData.of(KEY, TimestampData.fromEpochMillis(ts)), ts));
+ }
+ // Advance watermark past registered timers (first batch: three unnamed + one named).
+ harness.processElement(new Watermark(5000L));
+
+ // One more input row followed by a second watermark advance (second batch: one
+ // unnamed timer only).
+ harness.processElement(
+ new StreamRecord<>(
+ GenericRowData.of(KEY, TimestampData.fromEpochMillis(6000L)), 6000L));
+ harness.processElement(new Watermark(7000L));
+
+ List<String> labels =
+ harness.getOutput().stream()
+ .map(ProcessSetTableOperatorInterruptibleTimersTest::describe)
+ .collect(Collectors.toList());
+
+ // Records emitted from callEval carry the watermark visible during eval, which is null
+ // before any watermark has arrived and 5000 for the record processed between the two
+ // watermarks.
+ if (interruptibleTimers) {
+ // With interruptible timers, the mailbox is drained between successive timer
+ // firings, so each "fired-[name-]$ts" is immediately followed by its matching
+ // "mail-[name-]$ts". The watermark is only emitted downstream once all firings in
+ // the batch have completed.
+ assertThat(labels)
+ .containsExactly(
+ recordLabel(1000L, null),
+ recordLabel(2000L, null),
+ recordLabel(3000L, null),
+ firedLabel(null, 1000L, 5000L),
+ mailLabel(null, 1000L),
+ firedLabel(null, 2000L, 5000L),
+ mailLabel(null, 2000L),
+ firedLabel(null, 3000L, 5000L),
+ mailLabel(null, 3000L),
+ firedLabel(NAMED_TIMER, 1000L, 5000L),
+ mailLabel(NAMED_TIMER, 1000L),
+ watermarkLabel(5000L),
+ recordLabel(6000L, 5000L),
+ firedLabel(null, 6000L, 7000L),
+ mailLabel(null, 6000L),
+ watermarkLabel(7000L));
+ } else {
+ // Without interruptible timers, all timers in a batch fire synchronously (emitting
+ // their "fired" rows) before the watermark is emitted downstream. Only afterwards
+ // does the mailbox drain the scheduled "mail" rows in FIFO order.
+ assertThat(labels)
+ .containsExactly(
+ recordLabel(1000L, null),
+ recordLabel(2000L, null),
+ recordLabel(3000L, null),
+ firedLabel(null, 1000L, 5000L),
+ firedLabel(null, 2000L, 5000L),
+ firedLabel(null, 3000L, 5000L),
+ firedLabel(NAMED_TIMER, 1000L, 5000L),
+ watermarkLabel(5000L),
+ mailLabel(null, 1000L),
+ mailLabel(null, 2000L),
+ mailLabel(null, 3000L),
+ mailLabel(NAMED_TIMER, 1000L),
+ recordLabel(6000L, 5000L),
+ firedLabel(null, 6000L, 7000L),
+ watermarkLabel(7000L),
+ mailLabel(null, 6000L));
+ }
+ }
+ }
+
+ private StreamTaskMailboxTestHarness<RowData> createHarness(boolean interruptibleTimers)
+ throws Exception {
+ return new StreamTaskMailboxTestHarnessBuilder<>(
+ MultipleInputStreamTask::new, OUTPUT_TYPE_INFO)
+ .addJobConfig(CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(1))
+ .addJobConfig(CheckpointingOptions.ENABLE_UNALIGNED, true)
+ .addJobConfig(
+ CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS,
+ interruptibleTimers)
+ .setKeyType(KEY_SELECTOR.getProducedType())
+ .addInput(INPUT_TYPE_INFO, 1, KEY_SELECTOR)
+ .setupOperatorChain(new TestOperatorFactory())
+ .name("ptf")
+ .finishForSingletonOperatorChain(
+ OUTPUT_TYPE_INFO.createSerializer(new SerializerConfigImpl()))
+ .build();
+ }
+
+ private static String describe(Object obj) {
+ if (obj instanceof Watermark) {
+ return watermarkLabel(((Watermark) obj).getTimestamp());
+ } else if (obj instanceof StreamRecord) {
+ Object value = ((StreamRecord<?>) obj).getValue();
+ if (value instanceof RowData) {
+ // Field 1 is the PTF-emitted label (see OUTPUT_TYPE); fields 0 and 2 are the
+ // pass-through partition key and rowtime.
+ return ((RowData) value).getString(1).toString();
+ }
+ return value.toString();
+ }
+ return obj.toString();
+ }
+
+ private static String recordLabel(long ts, @Nullable Long watermark) {
+ return "record-" + ts + "@wm" + (watermark == null ? "null" : watermark);
+ }
+
+ private static String firedLabel(@Nullable String name, long ts, long watermark) {
+ return "fired-" + (name == null ? "" : name + "-") + ts + "@wm" + watermark;
+ }
+
+ private static String mailLabel(@Nullable String name, long ts) {
+ return "mail-" + (name == null ? "" : name + "-") + ts;
+ }
+
+ private static String watermarkLabel(long ts) {
+ return "Watermark@" + ts;
+ }
+
+ private static RuntimeTableSemantics tableSemantics() {
+ return new RuntimeTableSemantics(
+ "r",
+ 0,
+ INPUT_TYPE,
+ new int[] {0},
+ new int[0],
+ new RuntimeTableSemantics.SortDirection[0],
+ RuntimeChangelogMode.serialize(ChangelogMode.insertOnly()),
+ /* passColumnsThrough */ false,
+ /* hasSetSemantics */ true,
+ /* timeColumn */ 1);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Factory and operator
+ // --------------------------------------------------------------------------------------------
+
+ private static class TestOperatorFactory extends AbstractStreamOperatorFactory<RowData> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<RowData>> T createStreamOperator(
+ StreamOperatorParameters<RowData> parameters) {
+ return (T) new TestProcessSetTableOperator(parameters);
+ }
+
+ @Override
+ @SuppressWarnings("rawtypes")
+ public Class<? extends StreamOperator> getStreamOperatorClass(ClassLoader classLoader) {
+ return TestProcessSetTableOperator.class;
+ }
+ }
+
+ /**
+ * {@link ProcessSetTableOperator} with a test-only override of {@link #onEventTime} that
+ * schedules a non-deferrable "mail-[name-]$ts" emission after each timer firing. The PTF user
+ * API does not expose the mailbox; this test-only shortcut is the only way to observe whether
+ * interruption happens between timer firings driven by {@link
+ * ProcessTableRunner#callOnTimer()}.
+ */
+ private static class TestProcessSetTableOperator extends ProcessSetTableOperator {
+
+ private final MailboxExecutor mailboxExecutor;
+
+ TestProcessSetTableOperator(StreamOperatorParameters<RowData> parameters) {
+ super(
+ parameters,
+ List.of(tableSemantics()),
+ List.of(),
+ // No ORDER BY on the input, so the comparator slot is null (the operator
+ // short-circuits sort-buffer setup when orderByColumns is empty). The array
+ // size must still match the number of inputs.
+ new RecordComparator[] {null},
+ new TestProcessTableRunner(),
+ new HashFunction[0],
+ new RecordEqualiser[0],
+ RuntimeChangelogMode.serialize(ChangelogMode.insertOnly()),
+ List.of());
+ this.mailboxExecutor = parameters.getMailboxExecutor();
+ }
+
+ @Override
+ public void onEventTime(InternalTimer<RowData, Object> timer) throws Exception {
+ // Runs the full PTF path: ingestTimerEvent -> processOnTimer -> callOnTimer, emitting
+ // "fired-[name-]$ts" via onTimerCollector.
+ super.onEventTime(timer);
+ long ts = timer.getTimestamp();
+ long keyValue = timer.getKey().getLong(0);
+ final Object ns = timer.getNamespace();
+ final String name = (ns instanceof StringData) ? ns.toString() : null;
+ mailboxExecutor.execute(
+ () ->
+ output.collect(
+ new StreamRecord<>(
+ GenericRowData.of(
+ keyValue,
+ StringData.fromString(mailLabel(name, ts)),
+ TimestampData.fromEpochMillis(ts)))),
+ mailLabel(name, ts));
+ }
+ }
+
+ /**
+ * Minimal {@link ProcessTableRunner}: {@link #callEval} emits a "record-$ts@wm$wm" row and
+ * registers an unnamed event-time timer at the input row's event time (plus a named timer for
+ * the row at ts=1000); {@link #callOnTimer} emits a "fired-[name-]$ts@wm$wm" row.
+ */
+ private static class TestProcessTableRunner extends ProcessTableRunner {
+
+ @Override
+ public void callEval() {
+ final TimeContext<Long> tc = runnerContext.timeContext(Long.class);
+ final long ts = Objects.requireNonNull(tc.time(), "input row event time");
+ final Long wm = tc.currentWatermark();
+ tc.registerOnTime(ts);
+ if (ts == 1000L) {
+ tc.registerOnTime(NAMED_TIMER, ts);
+ }
+ evalCollector.collect(GenericRowData.of(StringData.fromString(recordLabel(ts, wm))));
+ }
+
+ @Override
+ public void callOnTimer() {
+ final TimeContext<Long> tc = runnerOnTimerContext.timeContext(Long.class);
+ final long ts = Objects.requireNonNull(tc.time(), "timer fire time");
+ final long wm =
+ Objects.requireNonNull(tc.currentWatermark(), "watermark at timer fire");
+ final String name = runnerOnTimerContext.currentTimer();
+ onTimerCollector.collect(
+ GenericRowData.of(StringData.fromString(firedLabel(name, ts, wm))));
+ }
+ }
+}