[sangkyoonnam]

Linked issue: #1332

Purpose of change

This PR adds ActionExecutionOperatorMigrationTest, which restores ActionExecutionOperator state written by an earlier release, modeled on Flink's migration tests (WindowOperatorMigrationTest, CEPMigrationTest). The committed reference snapshot comes from the 0.4.0 release. A state-layout or serializer change that breaks restoring that snapshot now fails the test on the PR that makes it. The first scenario covers state at rest only.

Runtime flow

Generation runs once per release, on that release's branch. mvn clean package -Pgenerate-migration-test-data -Dgenerate.version=0.4 -DskipTests -pl runtime -am runs MigrationTestsSnapshotGenerator in the runtime module directory. It calls each @SnapshotsGenerator method of ActionExecutionOperatorMigrationTest, which drives the operator through a keyed test harness and writes the snapshot with OperatorSnapshotUtil to src/test/resources/action-execution-operator-migration-at-rest-flink-agents<version>-snapshot.

The test runs in the normal suite, parameterized over FlinkAgentsVersion.publishedVersions(). For each version it restores the committed snapshot into a new harness, checks the restored state, then processes more input.

Key decisions

• A FlinkAgentsVersion enum instead of Flink's MigrationTest utilities, which are typed on FlinkVersion and can't name Flink Agents releases. The generator and the annotation are a small subset of Flink's: no classpath scanning, the profile names the classes.
• The test uses its own agent, MigrationTestAgent, with one action that keeps a running total per key in short-term memory. Its actions and the generated input must stay the same across releases, so the expected values don't move with the code under test. Inputs are keyed by parity so one key has seen two inputs.
• The 0.4 snapshot is an early warning, not a guarantee: in Planning Flink Agents 1.0 - Brainstorming #1327 you scoped the guarantee to upgrades within 1.x, excluding 0.4 to 1.0. If a 1.0 change breaks restoring it on purpose, that PR can remove the snapshot. It was generated with the Maven profile after adding these test files to the release-0.4.0 tree.

Behavioral Semantics

Interaction decisions

Not applicable: the change crosses no independent conditions.

Behavioral contracts

None for users. This PR adds test code, a test resource and a Maven profile; production code is unchanged.

Failure behavior

The generator fails the build when an argument is missing or unknown, when --classes names no class, when --version doesn't match a FlinkAgentsVersion constant, when a listed class has no @SnapshotsGenerator method, or when a generator method throws.

Tests

What the test pins after restoring a 0.4 snapshot	Assertion in restoresStateAtRest
Short-term memory keeps each key's value	Processing continues from the saved totals: outputs 5, 4, 10
Per-key sequence number keeps its value	messageSequenceNumber is 1 and 0, then 3 and 1 after more input
The operator keeps the persisted job identifier	jobIdentifier equals the restored identifier_state and differs from the new harness's job ID
No action or input is pending	actionTasks, pendingInputEvents and currentProcessingKeys are empty
The generator fails on bad input	No automated test; I ran it by hand for each case above

Renaming the short-term memory, sequence number or job identifier state makes the test fail. The pending checks are weaker: a renamed pending state also reads as empty, which matches an at-rest snapshot.

Ran mvn -pl api,plan,runtime test, the migration test on JDK 11, 17 and 21 (11 and 17 use the parallel engine, so each input waits on waitInFlightEventsFinished()), ./tools/lint.sh -f and -c on JDK 11, the generation profile on the release-0.4.0 tree, and tools/check-license.sh.

Not verified: pending actions and events, Python actions, an Action State Store, and Flink versions other than the root build's 2.3.0. Restoring a snapshot with a pending action failed on earlier upgrade paths (#1327); that scenario and any serializer change are follow-ups.

API

No public API changes. FlinkAgentsVersion, @SnapshotsGenerator and the generator live in runtime test sources.

Documentation

• doc-needed
• doc-not-needed
• doc-included

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: Claude Code 2.1.312 (Claude Opus 5.5)

As a follow-up, I'd add snapshot generation for each minor release to the release guide, as Flink does. Would you rather keep the 0.4 snapshot as an early warning, or drop it and start from 1.0.0?

[wenjin272]

Thanks for adding this! Looks good to me as a first step for migration testing. We can cover pending actions/events and Action State Store recovery in follow-ups.