diff --git a/fluss-flink/fluss-flink-1.18/pom.xml b/fluss-flink/fluss-flink-1.18/pom.xml index 9f67b6ce9b..45ab3fd356 100644 --- a/fluss-flink/fluss-flink-1.18/pom.xml +++ b/fluss-flink/fluss-flink-1.18/pom.xml @@ -161,6 +161,14 @@ test test-jar + + + + org.apache.flink + flink-scala_2.12 + ${flink.minor.version} + test + @@ -219,14 +227,11 @@ org.apache.fluss:fluss-client - - - org.apache.fluss:fluss-flink-common - - org/apache/fluss/flink/action/** - - - + + + org.apache.fluss.flink.action.FlussActionEntrypoint + + diff --git a/fluss-flink/fluss-flink-1.18/src/main/java/org/apache/fluss/flink/adapter/ProcessFunctionAdapter.java b/fluss-flink/fluss-flink-1.18/src/main/java/org/apache/fluss/flink/adapter/ProcessFunctionAdapter.java new file mode 100644 index 0000000000..edde1d2cbb --- /dev/null +++ b/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. + * + *

In Flink 1.18, only {@code open(Configuration)} is available (OpenContext does not exist). + * This adapter overrides that method and delegates to {@link #doOpen()}. + * + *

TODO: remove this class when no longer support flink 1.18. + * + * @param the input element type + * @param the output element type + */ +public abstract class ProcessFunctionAdapter extends ProcessFunction { + + @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; +} diff --git a/fluss-flink/fluss-flink-1.18/src/main/resources/META-INF/services/org.apache.fluss.flink.action.ActionFactory b/fluss-flink/fluss-flink-1.18/src/main/resources/META-INF/services/org.apache.fluss.flink.action.ActionFactory new file mode 100644 index 0000000000..c30c9dd5ab --- /dev/null +++ b/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 diff --git a/fluss-flink/fluss-flink-1.18/src/test/java/org/apache/fluss/flink/action/orphan/Flink118OrphanFilesCleanITCase.java b/fluss-flink/fluss-flink-1.18/src/test/java/org/apache/fluss/flink/action/orphan/Flink118OrphanFilesCleanITCase.java new file mode 100644 index 0000000000..e32e6edef6 --- /dev/null +++ b/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 {} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java index 2b54174643..e43810543c 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/action/orphan/job/ScanAndCleanFunction.java +++ b/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 { +public final class ScanAndCleanFunction extends ProcessFunctionAdapter { 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 diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/ProcessFunctionAdapter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/ProcessFunctionAdapter.java new file mode 100644 index 0000000000..7f1b10649d --- /dev/null +++ b/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. + * + *

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. + * + *

TODO: remove this class when no longer support flink 1.18. + * + * @param the input element type + * @param the output element type + */ +public abstract class ProcessFunctionAdapter extends ProcessFunction { + + @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; +}