[flink] Support remove_orphan_files action for Flink 1.18

fluss-flink/fluss-flink-1.18/pom.xml

@@ -161,6 +161,14 @@
             <scope>test</scope>
             <type>test-jar</type>
         </dependency>
+
+        <!-- Flink 1.18 DataStream batch mode requires flink-scala for Kryo serialization -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-scala_2.12</artifactId>
+            <version>${flink.minor.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -219,14 +227,11 @@
                                     <include>org.apache.fluss:fluss-client</include>
                                 </includes>
                             </artifactSet>
-                            <filters combine.children="append">
-                                <filter>
-                                    <artifact>org.apache.fluss:fluss-flink-common</artifact>
-                                    <excludes>
-                                        <exclude>org/apache/fluss/flink/action/**</exclude>
-                                    </excludes>
-                                </filter>
-                            </filters>
+                            <transformers combine.children="append">
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                                    <mainClass>org.apache.fluss.flink.action.FlussActionEntrypoint</mainClass>
+                                </transformer>
+                            </transformers>
                         </configuration>
                     </execution>
                 </executions>

fluss-flink/fluss-flink-1.18/src/main/java/org/apache/fluss/flink/adapter/ProcessFunctionAdapter.java

@@ -0,0 +1,44 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+
+/**
+ * Flink 1.18 variant of the process function adapter.
+ *
+ * <p>In Flink 1.18, only {@code open(Configuration)} is available (OpenContext does not exist).
+ * This adapter overrides that method and delegates to {@link #doOpen()}.
+ *
+ * <p>TODO: remove this class when no longer support flink 1.18.
+ *
+ * @param <I> the input element type
+ * @param <O> the output element type
+ */
+public abstract class ProcessFunctionAdapter<I, O> extends ProcessFunction<I, O> {
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        doOpen();
+    }
+
+    /** Subclass initialization logic, called exactly once when the function is opened. */
+    protected abstract void doOpen() throws Exception;
+}

fluss-flink/fluss-flink-1.18/src/main/resources/META-INF/services/org.apache.fluss.flink.action.ActionFactory

@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.apache.fluss.flink.action.orphan.OrphanFilesCleanActionFactory

fluss-flink/fluss-flink-1.18/src/test/java/org/apache/fluss/flink/action/orphan/Flink118OrphanFilesCleanITCase.java

@@ -0,0 +1,22 @@
+/*
+ * 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.fluss.flink.action.orphan;
+
+/** The IT case for orphan files cleanup in Flink 1.18. */
+class Flink118OrphanFilesCleanITCase extends OrphanFilesCleanITCase {}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java

@@ -26,12 +26,14 @@
 import org.apache.fluss.flink.action.orphan.rule.FileMeta;
 import org.apache.fluss.flink.action.orphan.rule.FileRule;
 import org.apache.fluss.flink.action.orphan.rule.RuleDispatcher;
+import org.apache.fluss.flink.adapter.ProcessFunctionAdapter;
+import org.apache.fluss.flink.adapter.RuntimeContextAdapter;
 import org.apache.fluss.fs.FileStatus;
 import org.apache.fluss.fs.FileSystem;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.shaded.guava32.com.google.common.util.concurrent.RateLimiter;
 
-import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
 import org.apache.flink.util.Collector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -60,7 +62,7 @@
  * processing within each subtask guarantees no concurrent throttler access.
  */
 @Internal
-public final class ScanAndCleanFunction extends ProcessFunction<CleanTask, CleanStats> {
+public final class ScanAndCleanFunction extends ProcessFunctionAdapter<CleanTask, CleanStats> {
 
     private static final long serialVersionUID = 1L;
     private static final Logger LOG = LoggerFactory.getLogger(ScanAndCleanFunction.class);
@@ -78,15 +80,14 @@ public ScanAndCleanFunction(
     }
 
     @Override
-    public void open(org.apache.flink.api.common.functions.OpenContext openContext)
-            throws Exception {
-        super.open(openContext);
+    protected void doOpen() throws Exception {
         if (!extraConfigs.isEmpty()) {
             FileSystem.initialize(Configuration.fromMap(extraConfigs), null);
         }
         audit = new AuditLogger();
-        int parallelism = getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks();
-        int subtaskIndex = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
+        StreamingRuntimeContext ctx = (StreamingRuntimeContext) getRuntimeContext();
+        int parallelism = RuntimeContextAdapter.getNumberOfParallelSubtasks(ctx);
+        int subtaskIndex = RuntimeContextAdapter.getIndexOfThisSubtask(ctx);
         // Distribute the configured rate as base + 1 extra for the first `remainder` subtasks.
         // Flink does not provide a cross-JVM limiter here, so this is a best-effort job-level
         // target. Each subtask gets at least 1/s; if parallelism exceeds the configured rate, the

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/ProcessFunctionAdapter.java

@@ -0,0 +1,45 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+
+/**
+ * Adapter for {@link ProcessFunction} which hides the different version of {@code open} method.
+ *
+ * <p>In Flink 1.19+ and 2.x, the entry point is {@code open(OpenContext)}. In Flink 1.18, only
+ * {@code open(Configuration)} exists. This adapter overrides the correct lifecycle method and
+ * delegates to {@link #doOpen()} so that subclasses remain version-agnostic.
+ *
+ * <p>TODO: remove this class when no longer support flink 1.18.
+ *
+ * @param <I> the input element type
+ * @param <O> the output element type
+ */
+public abstract class ProcessFunctionAdapter<I, O> extends ProcessFunction<I, O> {
+
+    @Override
+    public void open(OpenContext openContext) throws Exception {
+        super.open(openContext);
+        doOpen();
+    }
+
+    /** Subclass initialization logic, called exactly once when the function is opened. */
+    protected abstract void doOpen() throws Exception;
+}