[Flink] Add RuntimeContextAdapter

fluss-flink/fluss-flink-1.18/src/main/java/org/apache/fluss/flink/adapter/RuntimeContextAdapter.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.fluss.flink.adapter;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+
+/**
+ * An adapter for Flink {@link RuntimeContext} class. The {@link RuntimeContext} class added the
+ * `getJobInfo` and `getTaskInfo` methods in version 1.19 and deprecated many methods, such as
+ * `getAttemptNumber`.
+ *
+ * <p>TODO: remove this class when no longer support flink 1.18.
+ */
+public class RuntimeContextAdapter {
+
+    public static int getAttemptNumber(RuntimeContext runtimeContext) {
+        return runtimeContext.getAttemptNumber();
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/RuntimeContextAdapter.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.fluss.flink.adapter;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+
+/**
+ * An adapter for Flink {@link RuntimeContext} class. The {@link RuntimeContext} class added the
+ * `getJobInfo` and `getTaskInfo` methods in version 1.19 and deprecated many methods, such as
+ * `getAttemptNumber`.
+ *
+ * <p>TODO: remove this class when no longer support flink 1.18.
+ */
+public class RuntimeContextAdapter {
+
+    public static int getAttemptNumber(RuntimeContext runtimeContext) {
+        return runtimeContext.getTaskInfo().getAttemptNumber();
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java

@@ -23,6 +23,7 @@
 import org.apache.fluss.client.metadata.LakeSnapshot;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
+import org.apache.fluss.flink.adapter.RuntimeContextAdapter;
 import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
 import org.apache.fluss.flink.tiering.event.FinishedTieringEvent;
 import org.apache.fluss.flink.tiering.event.TieringFailOverEvent;
@@ -132,7 +133,7 @@ public void setup(
             StreamConfig config,
             Output<StreamRecord<CommittableMessage<Committable>>> output) {
         super.setup(containingTask, config, output);
-        int attemptNumber = getRuntimeContext().getAttemptNumber();
+        int attemptNumber = RuntimeContextAdapter.getAttemptNumber(getRuntimeContext());
         if (attemptNumber > 0) {
             LOG.info("Send TieringFailoverEvent, current attempt number: {}", attemptNumber);
             // attempt number is greater than zero, the job must failover