[FLINK-36366][core] Remove deprecate API in flink-core exclude connector and state part

Author: reswqa

flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java

@@ -429,7 +429,7 @@ private static OutputFormat<RowData> createBulkWriterOutputFormat(
             public void configure(Configuration parameters) {}
 
             @Override
-            public void open(int taskNumber, int numTasks) throws IOException {
+            public void open(InitializationContext context) throws IOException {
                 this.stream = path.getFileSystem().create(path, FileSystem.WriteMode.OVERWRITE);
                 this.writer = factory.create(stream);
             }
@@ -460,7 +460,7 @@ private static OutputFormat<RowData> createEncoderOutputFormat(
             public void configure(Configuration parameters) {}
 
             @Override
-            public void open(int taskNumber, int numTasks) throws IOException {
+            public void open(InitializationContext context) throws IOException {
                 this.output = path.getFileSystem().create(path, FileSystem.WriteMode.OVERWRITE);
             }
 

flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionWriter.java

@@ -58,7 +58,23 @@ OutputFormat<T> createNewOutputFormat(Path path) throws IOException {
             OutputFormat<T> format = factory.createOutputFormat(path);
             format.configure(conf);
             // Here we just think of it as a single file format, so there can only be a single task.
-            format.open(0, 1);
+            format.open(
+                    new OutputFormat.InitializationContext() {
+                        @Override
+                        public int getNumTasks() {
+                            return 0;
+                        }
+
+                        @Override
+                        public int getTaskNumber() {
+                            return 1;
+                        }
+
+                        @Override
+                        public int getAttemptNumber() {
+                            return 0;
+                        }
+                    });
             return format;
         }
     }

flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/testutils/TestingFileSystem.java

@@ -23,7 +23,6 @@
 import org.apache.flink.core.fs.FSDataOutputStream;
 import org.apache.flink.core.fs.FileStatus;
 import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.core.fs.FileSystemKind;
 import org.apache.flink.core.fs.Path;
 
 import javax.annotation.Nullable;
@@ -254,12 +253,6 @@ public boolean rename(Path src, Path dst) throws IOException {
     public boolean isDistributedFS() {
         return false;
     }
-
-    @Override
-    public FileSystemKind getKind() {
-        return FileSystemKind.FILE_SYSTEM;
-    }
-
     // ------------------------------------------------------------------------
 
     /** Test implementation of a {@link FileStatus}. */

flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/PartitionWriterTest.java

@@ -71,7 +71,7 @@ void after() {
                         public void configure(Configuration parameters) {}
 
                         @Override
-                        public void open(int taskNumber, int numTasks) {
+                        public void open(InitializationContext context) {
                             records.put(getKey(), new ArrayList<>());
                         }
 

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/write/HiveOutputFormatFactory.java

@@ -62,7 +62,7 @@ private HiveOutputFormat(RecordWriter recordWriter, Function<Row, Writable> rowC
         public void configure(Configuration parameters) {}
 
         @Override
-        public void open(int taskNumber, int numTasks) {}
+        public void open(InitializationContext context) {}
 
         @Override
         public void writeRecord(Row record) throws IOException {

flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatBase.java

@@ -113,7 +113,7 @@ public void configure(Configuration parameters) {
      * @throws java.io.IOException
      */
     @Override
-    public void open(int taskNumber, int numTasks) throws IOException {
+    public void open(InitializationContext context) throws IOException {
 
         // enforce sequential open() calls
         synchronized (OPEN_MUTEX) {
@@ -178,7 +178,7 @@ public void close() throws IOException {
     }
 
     @Override
-    public void finalizeGlobal(int parallelism) throws IOException {
+    public void finalizeGlobal(FinalizationContext context) throws IOException {
 
         try {
             JobContext jobContext = new JobContextImpl(this.jobConf, new JobID());

flink-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java

@@ -98,12 +98,10 @@ public void configure(Configuration parameters) {
     /**
      * create the temporary output file for hadoop RecordWriter.
      *
-     * @param taskNumber The number of the parallel instance.
-     * @param numTasks The number of parallel tasks.
      * @throws java.io.IOException
      */
     @Override
-    public void open(int taskNumber, int numTasks) throws IOException {
+    public void open(InitializationContext context) throws IOException {
 
         // enforce sequential open() calls
         synchronized (OPEN_MUTEX) {
@@ -207,7 +205,7 @@ public void close() throws IOException {
     }
 
     @Override
-    public void finalizeGlobal(int parallelism) throws IOException {
+    public void finalizeGlobal(FinalizationContext context) throws IOException {
 
         JobContext jobContext;
         TaskAttemptContext taskContext;

flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatTest.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.java.hadoop.mapred;
 
+import org.apache.flink.api.common.io.FinalizeOnMaster;
+import org.apache.flink.api.common.io.FirstAttemptInitializationContext;
 import org.apache.flink.api.java.tuple.Tuple2;
 
 import org.apache.hadoop.conf.Configurable;
@@ -61,7 +63,7 @@ void testOpen() throws Exception {
         HadoopOutputFormat<String, Long> outputFormat =
                 new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
 
-        outputFormat.open(1, 1);
+        outputFormat.open(FirstAttemptInitializationContext.of(1, 1));
 
         verify(jobConf, times(2)).getOutputCommitter();
         verify(outputCommitter, times(1)).setupJob(any(JobContext.class));
@@ -163,7 +165,18 @@ void testFinalizeGlobal() throws Exception {
         HadoopOutputFormat<String, Long> outputFormat =
                 new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
 
-        outputFormat.finalizeGlobal(1);
+        outputFormat.finalizeGlobal(
+                new FinalizeOnMaster.FinalizationContext() {
+                    @Override
+                    public int getParallelism() {
+                        return 1;
+                    }
+
+                    @Override
+                    public int getFinishedAttempt(int subtaskIndex) {
+                        return 0;
+                    }
+                });
 
         verify(outputCommitter, times(1)).commitJob(any(JobContext.class));
     }

flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatTest.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.java.hadoop.mapreduce;
 
+import org.apache.flink.api.common.io.FinalizeOnMaster;
+import org.apache.flink.api.common.io.FirstAttemptInitializationContext;
 import org.apache.flink.api.java.tuple.Tuple2;
 
 import org.apache.hadoop.conf.Configurable;
@@ -80,7 +82,7 @@ void testOpen() throws Exception {
                         setupOutputCommitter(true),
                         new Configuration());
 
-        hadoopOutputFormat.open(1, 4);
+        hadoopOutputFormat.open(FirstAttemptInitializationContext.of(1, 4));
 
         verify(hadoopOutputFormat.outputCommitter, times(1)).setupJob(any(JobContext.class));
         verify(hadoopOutputFormat.mapreduceOutputFormat, times(1))
@@ -152,7 +154,18 @@ void testFinalizedGlobal() throws Exception {
                         null,
                         new Configuration());
 
-        hadoopOutputFormat.finalizeGlobal(1);
+        hadoopOutputFormat.finalizeGlobal(
+                new FinalizeOnMaster.FinalizationContext() {
+                    @Override
+                    public int getParallelism() {
+                        return 1;
+                    }
+
+                    @Override
+                    public int getFinishedAttempt(int subtaskIndex) {
+                        return 0;
+                    }
+                });
 
         verify(hadoopOutputFormat.outputCommitter, times(1)).commitJob(any(JobContext.class));
     }

flink-core/src/main/java/org/apache/flink/api/common/ArchivedExecutionConfig.java

@@ -33,8 +33,6 @@
 public class ArchivedExecutionConfig implements Serializable {
 
     private static final long serialVersionUID = 2907040336948181163L;
-
-    private final String executionMode;
     private final String restartStrategyDescription;
     private final int parallelism;
     private final int maxParallelism;
@@ -43,7 +41,6 @@ public class ArchivedExecutionConfig implements Serializable {
     private final Map<String, String> globalJobParameters;
 
     public ArchivedExecutionConfig(ExecutionConfig ec) {
-        executionMode = ec.getExecutionMode().name();
         restartStrategyDescription =
                 RestartStrategyDescriptionUtils.getRestartStrategyDescription(ec.toConfiguration());
 
@@ -59,14 +56,12 @@ public ArchivedExecutionConfig(ExecutionConfig ec) {
     }
 
     public ArchivedExecutionConfig(
-            String executionMode,
             String restartStrategyDescription,
             int maxParallelism,
             int parallelism,
             boolean objectReuseEnabled,
             long periodicMaterializeIntervalMillis,
             Map<String, String> globalJobParameters) {
-        this.executionMode = executionMode;
         this.restartStrategyDescription = restartStrategyDescription;
         this.maxParallelism = maxParallelism;
         this.parallelism = parallelism;
@@ -75,10 +70,6 @@ public ArchivedExecutionConfig(
         this.globalJobParameters = globalJobParameters;
     }
 
-    public String getExecutionMode() {
-        return executionMode;
-    }
-
     public String getRestartStrategyDescription() {
         return restartStrategyDescription;
     }