[FLINK-39378][table] Add support for Context, timers and on_time to PTF test harness by autophagy · Pull Request #28326 · apache/flink

autophagy · 2026-06-05T09:25:35Z

What is the purpose of the change

This PR adds support for Context, OnTimerContext, TimeContext, timer registration and firing, watermark management, and rowtime support to the PTF Test Harness.

When a PTF registers a timer, the timer is stored in the TimerManager and keyed by the partition row. When the user advances the watermark using
either setWatermark or setWatermarkForTable, the manager then fires all pending timers below or equal to the watermark in a deterministic order.

I moved per-invocation state stuff into an InvocationContext to make separation of concerns a little easier to follow.

Rows emitted from test harness setups that configure an on_time column also contain the rowtime column, and the PTF rejects at the point of registration if the user tries to register a timer with PASS_COLUMNS_THROUGH
enabled (similar to live, per the PROCESS_INVALID_PASS_THROUGH_TIMERS test on live.)

Some open questions remain. One is that there's an edge case where if a user's onTimer registers a new timer at or before the current watermark, it then fires, and then registers,
then fires, etc. Should there be some sort of max depth check here?

The second is that when defining the on time parameter in SQL, you pass in DESCRIPTOR(ts), but in the harness you just pass in the string name of the column. Would it instead be better to support withOnTime("DESCRIPTOR(ts)") to better mirror SQL, rather than withOnTimeColumn("ts")?

Brief change log

Added support for Context, OnTimerContext, TimeContext in ProcessTableFunctionTestHarness
Added support for timer registration, watermark tracking and timer firing in ProcessTableFunctionTestHarness
Added an InvocationContext to capture per-invocation state to simplify the collector logic
Reworked derriving the output type from the system inference to account for on_time and uid
Fixed an issue in createStateConverter where the incorrect state converters were being used for Map/List state.

Verifying this change

This change added tests and can be verified as follows:

Added tests to ProcessFunctionTestHarnessesTest to cover timer firing, watermark advancement and context.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (docs)

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

2.1.156 (Claude Code)

flinkbot · 2026-06-05T09:36:08Z

CI report:

2cfcafb Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

…rocessTableFunctionTestHarness

fhueske

Thanks for the PR @autophagy!

I've reviewed the changes except for the test (will do that later this week).
Overall, the changes are good. Left a bunch of comments with suggestions and questions.

Cheers, Fabian

fhueske · 2026-06-08T13:06:04Z

+            Optional<List<StaticArgument>> staticArgsOpt = systemTypeInference.getStaticArguments();
+            List<StaticArgument> staticArgs =
+                    staticArgsOpt.orElseThrow(
+                            () ->
+                                    new IllegalStateException(
+                                            "SystemTypeInference has no static arguments"));


nit

Suggested change

Optional<List<StaticArgument>> staticArgsOpt = systemTypeInference.getStaticArguments();

List<StaticArgument> staticArgs =

staticArgsOpt.orElseThrow(

() ->

new IllegalStateException(

"SystemTypeInference has no static arguments"));

List<StaticArgument> staticArgs =

systemTypeInference.getStaticArguments().orElseThrow(

() ->

new IllegalStateException(

"SystemTypeInference has no static arguments"));

Can be folded?

fhueske · 2026-06-08T13:32:37Z

+ * eligible to fire.
+ */
+@Internal
+class TestHarnessTimerManager {


It might make sense to add a separate test for this class.

fhueske · 2026-06-08T13:35:44Z

+        if (name != null) {
+            timerSet.removeIf(t -> name.equals(t.name));
+        }


I think we should also remove unnamed timers.
There can only be one timer per timestamp and name (also if name = null).

…tables

…timers per partition key

fhueske

Hi @autophagy, thanks for addressing most of my earlier comments!
I've had a look at the tests and they are pretty complete.
Left a few comments with suggestions to increase the coverage.

Cheers, Fabian

fhueske · 2026-06-11T11:06:37Z

+                Context ctx,
+                @ArgumentHint({ArgumentTrait.SET_SEMANTIC_TABLE, ArgumentTrait.REQUIRE_ON_TIME})
+                        Row input) {
+            TimeContext<LocalDateTime> timeCtx = ctx.timeContext(LocalDateTime.class);


Suggested change

TimeContext<LocalDateTime> timeCtx = ctx.timeContext(LocalDateTime.class);

TimeContext<Instant> timeCtx = ctx.timeContext(Instant.class);

to check another codepath?

Also check with Long and java.sql.Timestamp?

fhueske · 2026-06-11T11:09:53Z

+                Context ctx,
+                @ArgumentHint({ArgumentTrait.SET_SEMANTIC_TABLE, ArgumentTrait.REQUIRE_ON_TIME})
+                        Row input) {
+            TableSemantics semantics = ctx.tableSemanticsFor("input");


would be good to do the same check also for ctx.tableSemanticsFor() being called from the onTimer() method.

fhueske · 2026-06-11T11:11:19Z

    @DataTypeHint("ROW<value INT>")
    public static class PTFWithContext extends ProcessTableFunction<Row> {
        public void eval(Context ctx, @ArgumentHint(ArgumentTrait.ROW_SEMANTIC_TABLE) Row input) {
            collect(input);
        }
    }


I don't think we need this PTF (and the corresponding test) anymore.
Everything it covers is covered many times by all the new tests.

fhueske · 2026-06-11T11:13:17Z

+        private final Class<TimeType> conversionClass;
+
+        TestTimeContext(Class<TimeType> conversionClass) {
+            this.conversionClass = conversionClass;


Should we eagerly check here if conversionClass is a supported class?

fhueske · 2026-06-11T11:16:38Z

+
+            // Advance only left table past the timer — global watermark is still min(left, right)
+            // Since right has no watermark yet, global won't advance enough to fire
+            harness.setWatermarkForTable("rightTable", LocalDateTime.of(2025, 1, 1, 0, 0, 3));


Do we also have tests that use the other supported WM data types?

fhueske · 2026-06-11T11:34:59Z

+                                    LocalDateTime.of(2025, 1, 1, 0, 0, 6)));
+
+            assertThat(harness.getPendingTimers()).isEmpty();
+            assertThat(harness.getFiredTimers()).hasSize(1);


Also assert the specific fired timer (name + ts)?

fhueske · 2026-06-11T11:37:00Z

+                        .withOnTimeColumn("ts")
+                        .build()) {
+
+            harness.setWatermark(LocalDateTime.of(2025, 1, 1, 0, 0, 5));


also check for setWatermarkForTable()?

fhueske · 2026-06-11T11:37:43Z

+
+            harness.processElement(Row.of("P1", LocalDateTime.of(2025, 1, 1, 0, 0, 1)));
+
+            assertThrows(


check if error message is meaningful for users?

fhueske · 2026-06-11T11:38:00Z

+                        .build()) {
+
+            assertThrows(
+                    TableRuntimeException.class,


check if error message is meaningful?

fhueske · 2026-06-11T11:38:31Z

+            harness.clearOutput();
+
+            harness.setWatermark(LocalDateTime.of(2025, 1, 1, 0, 0, 7));
+            assertThat(harness.getOutput()).hasSize(1);


check that the right onTimer() method was called by checking the output row?

autophagy marked this pull request as ready for review June 5, 2026 11:28

fhueske changed the title ~~Timers~~ [FLINK-39378][table] Add support for Context, timers and on_time to PTF test harness Jun 5, 2026

[FLINK-39378][table] Add support for Context, timers and on_time to P…

efe9d3a

…rocessTableFunctionTestHarness

autophagy force-pushed the flink-39378-timers-2 branch from b05a727 to efe9d3a Compare June 8, 2026 09:59

fhueske reviewed Jun 8, 2026

View reviewed changes

autophagy added 12 commits June 9, 2026 17:27

Add example for asserting on timestamp for a timer

c3124c8

Clarify harness output during timer example

90fd9ee

Show multiple timer firing with one watermark advancement

5542a5f

Clarify what is set on invocationcontext

ae83622

Clarify timer firing conditions when setting watermarks for specific …

5eac6fc

…tables

Give getTimers a better name, and add public api methods for getting …

9be4712

…timers per partition key

Tweak error message for timer firing with no valid ontimer

dd64482

Be more specific about try/catch blocks

3399110

Improve error message in tableSemanticsFor

c2d04dc

Rename timeColumn

57b0581

nit: whitespace about testontimercontext constructor

2cfb45c

Remove initial watermark methods in builder

2cfcafb

fhueske reviewed Jun 11, 2026

View reviewed changes

	TimeContext<LocalDateTime> timeCtx = ctx.timeContext(LocalDateTime.class);
	TimeContext<Instant> timeCtx = ctx.timeContext(Instant.class);


		harness.processElement(Row.of("P1", LocalDateTime.of(2025, 1, 1, 0, 0, 1)));

		assertThrows(

Conversation

autophagy commented Jun 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Jun 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

fhueske left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fhueske left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

autophagy commented Jun 5, 2026 •

edited

Loading

flinkbot commented Jun 5, 2026 •

edited

Loading