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
The following commit(s) were added to refs/heads/release-2.3 by this push:
new be45361034c [FLINK-40571][table-planner] Fix flaky
process-stateful-multi-input-with-timeout PTF test (#29352)
be45361034c is described below
commit be45361034c9caed02a6a714fc5757662e67659d
Author: Purushottam Sinha <[email protected]>
AuthorDate: Mon Oct 12 15:08:38 2026 +0530
[FLINK-40571][table-planner] Fix flaky
process-stateful-multi-input-with-timeout PTF test (#29352)
* [FLINK-40571][table-planner] Fix flaky
process-stateful-multi-input-with-timeout PTF test
The two inputs of the PTF are processed in arrival order, not event-time
order. TimedJoinFunction keeps a single score per key, so if the city row
arrived after several scores, the earlier scores were overwritten and only
one join row was emitted at the city row's timestamp. The pinned output
only matched the city-first interleaving.
Restructure the test data so the output is independent of the
interleaving: a key with a city has a single score with the same timestamp
as its city, while keys without a city may have multiple scores. Timeouts
stay deterministic as timers only fire once both inputs have finished.
Generated-by: Claude Code (claude-opus-5-5)
* [FLINK-40571][table-planner] Add single-input PTF test for clearAllTimers
keeping state
The reworked multi-input timeout test no longer has a key receiving an
event after its join, so nothing verified that ctx.clearAllTimers()
leaves state intact. Add a dedicated single-input program for it and
inline the city source into the only program that uses it.
Generated-by: Claude Code (claude-opus-5-5)
(cherry picked from commit 7bd618a7910b21f010f107fb5c8347ac1c1f4e40)
---
.../stream/ProcessTableFunctionSemanticTests.java | 1 +
.../stream/ProcessTableFunctionTestPrograms.java | 69 +++++++++++++++++++---
.../exec/stream/ProcessTableFunctionTestUtils.java | 31 ++++++----
3 files changed, 82 insertions(+), 19 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
index b113671cbe1..a6d44797eb3 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
@@ -101,6 +101,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT,
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_WITH_SCALAR_ARGS,
ProcessTableFunctionTestPrograms.PROCESS_STATEFUL_MULTI_INPUT_WITH_TIMEOUT,
+
ProcessTableFunctionTestPrograms.PROCESS_CLEAR_ALL_TIMERS_KEEPS_STATE,
ProcessTableFunctionTestPrograms.PROCESS_UPDATING_MULTI_INPUT,
ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY,
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
index 0749d8fe7a3..9224dad790f 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
@@ -24,6 +24,7 @@ import
org.apache.flink.table.api.config.ExecutionConfigOptions;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.AtomicTypeWrappingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedReceivingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedSendingFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ClearAllTimersKeepsStateFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ClearStateFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ContextFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
@@ -96,7 +97,6 @@ import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTable
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.MULTI_VALUES_SOURCE_SCHEMA;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.PASS_THROUGH_BASE_SINK_SCHEMA;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_BASE_SINK_SCHEMA;
-import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_CITY_SOURCE;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_MULTI_BASE_SINK_SCHEMA;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_SOURCE;
import static
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.TIMED_SOURCE_LATE_EVENTS;
@@ -1567,19 +1567,42 @@ public class ProcessTableFunctionTestPrograms {
"process-stateful-multi-input-with-timeout",
"joins two tables and emits the left side after a
timeout if there is no right side")
.setupTemporarySystemFunction("f", TimedJoinFunction.class)
- .setupTableSource(TIMED_SOURCE)
- .setupTableSource(TIMED_CITY_SOURCE)
+ // The two inputs may reach eval() in any interleaving. To
keep the output
+ // independent of it, a key with a city has a single score
with the same
+ // timestamp as its city. Keys without a city may have
multiple scores.
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(TIMED_SOURCE_SCHEMA)
+ .producedValues(
+ Row.of("Bob", 1,
Instant.ofEpochMilli(0)),
+ Row.of("Alice", 1,
Instant.ofEpochMilli(1)),
+ Row.of("Charly", 3,
Instant.ofEpochMilli(2)),
+ Row.of("Alice", 2,
Instant.ofEpochMilli(3)),
+ Row.of("Dave", 4,
Instant.ofEpochMilli(4)),
+ Row.of("Frank", 5,
Instant.ofEpochMilli(6)))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("city")
+ .addSchema(
+ "name STRING",
+ "city STRING",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'0.001' SECOND")
+ .producedValues(
+ Row.of("Bob", "London",
Instant.ofEpochMilli(0)),
+ Row.of("Charly", "Paris",
Instant.ofEpochMilli(2)),
+ Row.of("Dave", "Berlin",
Instant.ofEpochMilli(4)),
+ Row.of("Eve", "Rome",
Instant.ofEpochMilli(5)))
+ .build())
.setupTableSink(
SinkTestStep.newBuilder("sink")
.addSchema(TIMED_MULTI_BASE_SINK_SCHEMA)
.consumedValues(
"+I[Bob, Bob, 1 score in city
London, 1970-01-01T00:00:00Z]",
- "+I[Bob, Bob, 2 score in city
London, 1970-01-01T00:00:00.002Z]",
- "+I[Bob, Bob, 3 score in city
London, 1970-01-01T00:00:00.003Z]",
- "+I[Bob, Bob, 4 score in city
London, 1970-01-01T00:00:00.004Z]",
- "+I[Bob, Bob, 5 score in city
London, 1970-01-01T00:00:00.005Z]",
- "+I[Bob, Bob, 6 score in city
London, 1970-01-01T00:00:00.006Z]",
- "+I[Alice, Alice, no city found
for score 1, 1970-01-01T00:00:01.001Z]")
+ "+I[Charly, Charly, 3 score in
city Paris, 1970-01-01T00:00:00.002Z]",
+ "+I[Dave, Dave, 4 score in city
Berlin, 1970-01-01T00:00:00.004Z]",
+ "+I[Alice, Alice, no city found
for score 2, 1970-01-01T00:00:01.003Z]",
+ "+I[Frank, Frank, no city found
for score 5, 1970-01-01T00:00:01.006Z]")
.build())
.runSql(
"INSERT INTO sink SELECT * FROM f("
@@ -1588,6 +1611,34 @@ public class ProcessTableFunctionTestPrograms {
+ "on_time => DESCRIPTOR(ts))")
.build();
+ public static final TableTestProgram PROCESS_CLEAR_ALL_TIMERS_KEEPS_STATE =
+ TableTestProgram.of(
+ "process-clear-all-timers-keeps-state",
+ "clearing all timers does not clear the state")
+ .setupTemporarySystemFunction("f",
ClearAllTimersKeepsStateFunction.class)
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(TIMED_SOURCE_SCHEMA)
+ .producedValues(
+ Row.of("Bob", 1,
Instant.ofEpochMilli(0)),
+ Row.of("Alice", 1,
Instant.ofEpochMilli(1)),
+ Row.of("Bob", 2,
Instant.ofEpochMilli(2)),
+ Row.of("Bob", 3,
Instant.ofEpochMilli(3)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(KEYED_TIMED_BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[Bob, count 1,
1970-01-01T00:00:00Z]",
+ "+I[Alice, count 1,
1970-01-01T00:00:00.001Z]",
+ "+I[Bob, count 2,
1970-01-01T00:00:00.002Z]",
+ "+I[Bob, count 3,
1970-01-01T00:00:00.003Z]",
+ "+I[Alice, timeout with count 1,
1970-01-01T00:00:01.001Z]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM f(r => TABLE t
PARTITION BY name, on_time => DESCRIPTOR(ts))")
+ .build();
+
public static final TableTestProgram PROCESS_UPDATING_MULTI_INPUT =
TableTestProgram.of(
"process-updating-multi-input",
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
index c8f41318731..69c610abf55 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
@@ -95,16 +95,6 @@ public class ProcessTableFunctionTestUtils {
"CREATE VIEW city AS SELECT * FROM "
+ "(VALUES ('Bob', 'London'), ('Alice', 'Berlin'),
('Charly', 'Paris')) AS T(name, city)";
- public static final SourceTestStep TIMED_CITY_SOURCE =
- SourceTestStep.newBuilder("city")
- .addSchema(
- "name STRING",
- "city STRING",
- "ts TIMESTAMP_LTZ(3)",
- "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND")
- .producedValues(Row.of("Bob", "London",
Instant.ofEpochMilli(0)))
- .build();
-
public static final String UPDATING_VALUES =
"CREATE VIEW t AS SELECT name, COUNT(*) FROM "
+ "(VALUES ('Bob', 12), ('Alice', 42), ('Bob', 14)) AS
T(name, score) "
@@ -1044,6 +1034,27 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Testing function. */
+ public static class ClearAllTimersKeepsStateFunction extends
AppendProcessTableFunctionBase {
+ public void eval(
+ Context ctx,
+ @StateHint Tuple1<Integer> count,
+ @ArgumentHint({SET_SEMANTIC_TABLE, REQUIRE_ON_TIME}) Row r) {
+ final TimeContext<Instant> timeCtx =
ctx.timeContext(Instant.class);
+ count.f0 = count.f0 == null ? 1 : count.f0 + 1;
+ if (count.f0 == 1) {
+ timeCtx.registerOnTime("timeout",
timeCtx.time().plusMillis(1000));
+ } else if (count.f0 == 2) {
+ ctx.clearAllTimers();
+ }
+ collect(Row.of("count " + count.f0));
+ }
+
+ public void onTimer(OnTimerContext ctx, Tuple1<Integer> count) {
+ collect(Row.of("timeout with count " + count.f0));
+ }
+ }
+
/**
* Implements a custom join that acts like kind of an outer join and never
produces deletions.
* Both the score and city can change at any time. The join will output an
update if a matching
