[flink] Avoid deprecated APIs removed in Flink 2.0 Preview

paimon-benchmark/paimon-cluster-benchmark/src/main/java/org/apache/paimon/benchmark/QueryRunner.java

@@ -77,7 +77,7 @@ public Result run() {
 
             String sinkPathConfig =
                     BenchmarkGlobalConfiguration.loadConfiguration()
-                            .getString(BenchmarkOptions.SINK_PATH);
+                            .get(BenchmarkOptions.SINK_PATH);
             if (sinkPathConfig == null) {
                 throw new IllegalArgumentException(
                         BenchmarkOptions.SINK_PATH.key() + " must be set");

paimon-flink/paimon-flink-1.15/src/main/java/org/apache/flink/api/common/JobInfo.java

@@ -0,0 +1,40 @@
+/*
+ * 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.api.common;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/** The {@link JobInfo} represents the meta information of current job. */
+@PublicEvolving
+public interface JobInfo {
+
+    /**
+     * Get the ID of the job.
+     *
+     * @return the ID of the job
+     */
+    JobID getJobId();
+
+    /**
+     * Get the name of the job.
+     *
+     * @return the name of the job
+     */
+    String getJobName();
+}

paimon-flink/paimon-flink-1.15/src/main/java/org/apache/flink/api/common/SupportsConcurrentExecutionAttempts.java

@@ -0,0 +1,31 @@
+/*
+ * 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.api.common;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * The interface indicates that it supports multiple attempts executing at the same time.
+ *
+ * <p>Currently, the interface is used for speculative execution. If a sink implementation (SinkV2,
+ * OutputFormat or SinkFunction) inherits this interface, the sink operator would be considered to
+ * support speculative execution.
+ */
+@PublicEvolving
+public interface SupportsConcurrentExecutionAttempts {}

paimon-flink/paimon-flink-1.15/src/main/java/org/apache/flink/api/common/functions/DefaultOpenContext.java

@@ -0,0 +1,28 @@
+/*
+ * 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.api.common.functions;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/** The default implementation of {@link OpenContext}. */
+@PublicEvolving
+public class DefaultOpenContext implements OpenContext {
+
+    public static final OpenContext INSTANCE = new DefaultOpenContext();
+}

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/flink/table/catalog/CatalogMaterializedTable.java → paimon-flink/paimon-flink-1.15/src/main/java/org/apache/flink/api/common/functions/OpenContext.java

@@ -16,19 +16,14 @@
  * limitations under the License.
  */
 
-package org.apache.flink.table.catalog;
+package org.apache.flink.api.common.functions;
+
+import org.apache.flink.annotation.PublicEvolving;
 
 /**
- * Dummy placeholder to resolve compatibility issue of CatalogMaterializedTable(introduced in flink
- * 1.20).
+ * The {@link OpenContext} interface provides necessary information required by the {@link
+ * RichFunction} when it is opened. The {@link OpenContext} is currently empty because it can be
+ * used to add more methods without affecting the signature of {@code RichFunction#open}.
  */
-public interface CatalogMaterializedTable extends CatalogBaseTable {
-    /** Dummy LogicalRefreshMode placeholder. */
-    enum LogicalRefreshMode {}
-
-    /** Dummy RefreshMode placeholder. */
-    enum RefreshMode {}
-
-    /** Dummy RefreshStatus placeholder. */
-    enum RefreshStatus {}
-}
+@PublicEvolving
+public interface OpenContext {}

paimon-flink/paimon-flink-1.15/src/main/java/org/apache/flink/api/common/functions/RichFunction.java

@@ -0,0 +1,171 @@
+/*
+ * 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.api.common.functions;
+
+import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.Configuration;
+
+/**
+ * An base interface for all rich user-defined functions. This class defines methods for the life
+ * cycle of the functions, as well as methods to access the context in which the functions are
+ * executed.
+ */
+@Public
+public interface RichFunction extends Function {
+
+    /**
+     * Initialization method for the function. It is called before the actual working methods (like
+     * <i>map</i> or <i>join</i>) and thus suitable for one time setup work. For functions that are
+     * part of an iteration, this method will be invoked at the beginning of each iteration
+     * superstep.
+     *
+     * <p>The configuration object passed to the function can be used for configuration and
+     * initialization. The configuration contains all parameters that were configured on the
+     * function in the program composition.
+     *
+     * <pre>{@code
+     * public class MyFilter extends RichFilterFunction<String> {
+     *
+     *     private String searchString;
+     *
+     *     public void open(Configuration parameters) {
+     *         this.searchString = parameters.getString("foo");
+     *     }
+     *
+     *     public boolean filter(String value) {
+     *         return value.equals(searchString);
+     *     }
+     * }
+     * }</pre>
+     *
+     * <p>By default, this method does nothing.
+     *
+     * @param parameters The configuration containing the parameters attached to the contract.
+     * @throws Exception Implementations may forward exceptions, which are caught by the runtime.
+     *     When the runtime catches an exception, it aborts the task and lets the fail-over logic
+     *     decide whether to retry the task execution.
+     * @see org.apache.flink.configuration.Configuration
+     * @deprecated This method is deprecated since Flink 1.19. The users are recommended to
+     *     implement {@code open(OpenContext openContext)} and implement {@code open(Configuration
+     *     parameters)} with an empty body instead. 1. If you implement {@code open(OpenContext
+     *     openContext)}, the {@code open(OpenContext openContext)} will be invoked and the {@code
+     *     open(Configuration parameters)} won't be invoked. 2. If you don't implement {@code
+     *     open(OpenContext openContext)}, the {@code open(Configuration parameters)} will be
+     *     invoked in the default implementation of the {@code open(OpenContext openContext)}.
+     * @see <a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=263425231">
+     *     FLIP-344: Remove parameter in RichFunction#open </a>
+     */
+    @Deprecated
+    void open(Configuration parameters) throws Exception;
+
+    /**
+     * Initialization method for the function. It is called before the actual working methods (like
+     * <i>map</i> or <i>join</i>) and thus suitable for one time setup work. For functions that are
+     * part of an iteration, this method will be invoked at the beginning of each iteration
+     * superstep.
+     *
+     * <p>The openContext object passed to the function can be used for configuration and
+     * initialization. The openContext contains some necessary information that were configured on
+     * the function in the program composition.
+     *
+     * <pre>{@code
+     * public class MyFilter extends RichFilterFunction<String> {
+     *
+     *     private String searchString;
+     *
+     *     public void open(OpenContext openContext) {
+     *         // initialize the value of searchString
+     *     }
+     *
+     *     public boolean filter(String value) {
+     *         return value.equals(searchString);
+     *     }
+     * }
+     * }</pre>
+     *
+     * <p>By default, this method does nothing.
+     *
+     * <p>1. If you implement {@code open(OpenContext openContext)}, the {@code open(OpenContext
+     * openContext)} will be invoked and the {@code open(Configuration parameters)} won't be
+     * invoked. 2. If you don't implement {@code open(OpenContext openContext)}, the {@code
+     * open(Configuration parameters)} will be invoked in the default implementation of the {@code
+     * open(OpenContext openContext)}.
+     *
+     * @param openContext The context containing information about the context in which the function
+     *     is opened.
+     * @throws Exception Implementations may forward exceptions, which are caught by the runtime.
+     *     When the runtime catches an exception, it aborts the task and lets the fail-over logic
+     *     decide whether to retry the task execution.
+     */
+    @PublicEvolving
+    default void open(OpenContext openContext) throws Exception {
+        open(new Configuration());
+    }
+
+    /**
+     * Tear-down method for the user code. It is called after the last call to the main working
+     * methods (e.g. <i>map</i> or <i>join</i>). For functions that are part of an iteration, this
+     * method will be invoked after each iteration superstep.
+     *
+     * <p>This method can be used for clean up work.
+     *
+     * @throws Exception Implementations may forward exceptions, which are caught by the runtime.
+     *     When the runtime catches an exception, it aborts the task and lets the fail-over logic
+     *     decide whether to retry the task execution.
+     */
+    void close() throws Exception;
+
+    // ------------------------------------------------------------------------
+    //  Runtime context
+    // ------------------------------------------------------------------------
+
+    /**
+     * Gets the context that contains information about the UDF's runtime, such as the parallelism
+     * of the function, the subtask index of the function, or the name of the task that executes the
+     * function.
+     *
+     * <p>The RuntimeContext also gives access to the {@link
+     * org.apache.flink.api.common.accumulators.Accumulator}s and the {@link
+     * org.apache.flink.api.common.cache.DistributedCache}.
+     *
+     * @return The UDF's runtime context.
+     */
+    RuntimeContext getRuntimeContext();
+
+    /**
+     * Gets a specialized version of the {@link RuntimeContext}, which has additional information
+     * about the iteration in which the function is executed. This IterationRuntimeContext is only
+     * available if the function is part of an iteration. Otherwise, this method throws an
+     * exception.
+     *
+     * @return The IterationRuntimeContext.
+     * @throws java.lang.IllegalStateException Thrown, if the function is not executed as part of an
+     *     iteration.
+     */
+    IterationRuntimeContext getIterationRuntimeContext();
+
+    /**
+     * Sets the function's runtime context. Called by the framework when creating a parallel
+     * instance of the function.
+     *
+     * @param t The runtime context.
+     */
+    void setRuntimeContext(RuntimeContext t);
+}