[runtime] Add cross-release state migration test for ActionExecutionOperator

runtime/pom.xml

@@ -246,6 +246,51 @@ under the License.
     </build>
 
     <profiles>
+        <!--
+          Writes the reference snapshots of migration tests for one release. Run it on the
+          release branch of that version:
+          mvn clean package -Pgenerate-migration-test-data -Dgenerate.version=0.4 -DskipTests -pl runtime -am
+        -->
+        <profile>
+            <id>generate-migration-test-data</id>
+            <properties>
+                <generate.classes>org.apache.flink.agents.runtime.operator.ActionExecutionOperatorMigrationTest</generate.classes>
+            </properties>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-antrun-plugin</artifactId>
+                        <executions>
+                            <execution>
+                                <id>generate-migration-test-data</id>
+                                <phase>package</phase>
+                                <goals>
+                                    <goal>run</goal>
+                                </goals>
+                                <configuration>
+                                    <target>
+                                        <java classname="org.apache.flink.agents.runtime.migration.MigrationTestsSnapshotGenerator"
+                                              fork="true" failonerror="true" dir="${project.basedir}">
+                                            <classpath refid="maven.test.classpath"/>
+                                            <!-- Same flags as the surefire argLine in the root pom. -->
+                                            <jvmarg value="--add-opens=java.base/java.util=ALL-UNNAMED"/>
+                                            <jvmarg value="--add-opens=java.base/java.nio=ALL-UNNAMED"/>
+                                            <jvmarg value="--add-exports=java.base/jdk.internal.vm=ALL-UNNAMED"/>
+                                            <arg value="--version"/>
+                                            <arg value="${generate.version}"/>
+                                            <arg value="--classes"/>
+                                            <arg value="${generate.classes}"/>
+                                        </java>
+                                    </target>
+                                </configuration>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+
         <!-- Profile for building JDK 21 specific classes when running on JDK 21+ -->
         <profile>
             <id>java-21</id>

runtime/src/test/java/org/apache/flink/agents/runtime/migration/FlinkAgentsVersion.java

@@ -0,0 +1,69 @@
+/*
+ * 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.agents.runtime.migration;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Flink Agents releases that have reference snapshots for migration tests.
+ *
+ * <p>Flink's {@code FlinkVersion} can't name Flink Agents releases, so migration tests are
+ * parameterized over this enum instead. After a release, add its constant here and generate its
+ * snapshots on the release branch with the {@code generate-migration-test-data} profile.
+ */
+public enum FlinkAgentsVersion {
+    v0_4("0.4");
+
+    private static final Pattern VERSION_PATTERN = Pattern.compile("v?(\\d+)[._](\\d+)");
+
+    private final String versionStr;
+
+    FlinkAgentsVersion(String versionStr) {
+        this.versionStr = versionStr;
+    }
+
+    @Override
+    public String toString() {
+        return versionStr;
+    }
+
+    /** Returns every version with reference snapshots, oldest first. */
+    public static List<FlinkAgentsVersion> publishedVersions() {
+        return Arrays.asList(values());
+    }
+
+    /** Parses names like {@code 0.4}, {@code 0_4} or {@code v0_4}. */
+    public static FlinkAgentsVersion parse(String name) {
+        Matcher matcher = VERSION_PATTERN.matcher(name);
+        if (matcher.matches()) {
+            String versionStr = matcher.group(1) + "." + matcher.group(2);
+            for (FlinkAgentsVersion version : values()) {
+                if (version.versionStr.equals(versionStr)) {
+                    return version;
+                }
+            }
+        }
+        throw new IllegalArgumentException(
+                "Unknown Flink Agents version "
+                        + name
+                        + ". Add it to FlinkAgentsVersion before generating its snapshots.");
+    }
+}

runtime/src/test/java/org/apache/flink/agents/runtime/migration/MigrationTestsSnapshotGenerator.java

@@ -0,0 +1,102 @@
+/*
+ * 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.agents.runtime.migration;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Writes the reference snapshots of migration tests for one Flink Agents version.
+ *
+ * <p>Run it through the {@code generate-migration-test-data} profile from the repository root, on
+ * the release branch of the version being generated. The profile runs the generator in the {@code
+ * runtime} module directory, so the snapshots land in {@code runtime/src/test/resources}:
+ *
+ * <pre>
+ * mvn clean package -Pgenerate-migration-test-data -Dgenerate.version=0.4 -DskipTests -pl runtime -am
+ * </pre>
+ *
+ * <p>Arguments: {@code --version <version> --classes <class>[,<class>...]}.
+ */
+public class MigrationTestsSnapshotGenerator {
+
+    private static final String USAGE =
+            "Usage: MigrationTestsSnapshotGenerator --version <version> --classes <class>[,<class>...]";
+
+    public static void main(String[] args) throws Throwable {
+        String versionName = null;
+        String classes = null;
+        if (args.length % 2 != 0) {
+            throw new IllegalArgumentException(USAGE);
+        }
+        for (int i = 0; i < args.length; i += 2) {
+            switch (args[i]) {
+                case "--version":
+                    versionName = args[i + 1];
+                    break;
+                case "--classes":
+                    classes = args[i + 1];
+                    break;
+                default:
+                    throw new IllegalArgumentException("Unknown argument " + args[i]);
+            }
+        }
+        List<String> classNames = new ArrayList<>();
+        if (classes != null) {
+            for (String className : classes.split(",")) {
+                if (!className.trim().isEmpty()) {
+                    classNames.add(className.trim());
+                }
+            }
+        }
+        if (versionName == null || classNames.isEmpty()) {
+            throw new IllegalArgumentException(USAGE);
+        }
+
+        FlinkAgentsVersion version = FlinkAgentsVersion.parse(versionName);
+        for (String className : classNames) {
+            generate(Class.forName(className), version);
+        }
+    }
+
+    private static void generate(Class<?> testClass, FlinkAgentsVersion version) throws Throwable {
+        Constructor<?> constructor = testClass.getDeclaredConstructor();
+        constructor.setAccessible(true);
+        Object test = constructor.newInstance();
+        int generated = 0;
+        for (Method method : testClass.getDeclaredMethods()) {
+            if (!method.isAnnotationPresent(SnapshotsGenerator.class)) {
+                continue;
+            }
+            method.setAccessible(true);
+            try {
+                method.invoke(test, version);
+            } catch (InvocationTargetException e) {
+                throw e.getCause();
+            }
+            generated++;
+        }
+        if (generated == 0) {
+            throw new IllegalArgumentException(
+                    testClass.getName() + " has no @SnapshotsGenerator method");
+        }
+    }
+}

runtime/src/test/java/org/apache/flink/agents/runtime/migration/SnapshotsGenerator.java

@@ -0,0 +1,34 @@
+/*
+ * 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.agents.runtime.migration;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Marks a method that writes the reference snapshots of a migration test.
+ *
+ * <p>The method must take a single {@link FlinkAgentsVersion} argument, the version the snapshots
+ * are written for. {@link MigrationTestsSnapshotGenerator} calls it on a new instance of the test
+ * class.
+ */
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface SnapshotsGenerator {}