This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.2 by this push:
new 73e8970f6f7 [FLINK-40571][table-planner] Fix flaky
process-stateful-multi-input-with-timeout PTF test (#29352)
73e8970f6f7 is described below
commit 73e8970f6f71ef2e639a6b436d6e11fed6cb3384
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 72a95b43696..3b260344978 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
@@ -97,6 +97,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_MAP_STATE,
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT,
ProcessTableFunctionTestPrograms.PROCESS_STATEFUL_MULTI_INPUT_WITH_TIMEOUT,
+
ProcessTableFunctionTestPrograms.PROCESS_CLEAR_ALL_TIMERS_KEEPS_STATE,
ProcessTableFunctionTestPrograms.PROCESS_UPDATING_MULTI_INPUT);
}
}
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 728a25e8e88..ad776fd0898 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;
@@ -88,7 +89,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;
@@ -1425,19 +1425,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("
@@ -1446,6 +1469,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 d20dc3fd70e..7b1c9a511a7 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
@@ -90,16 +90,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) "
@@ -955,6 +945,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
