[FLINK-36257][API] Remove easy-to-drop deprecated API

Author: reswqa

flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcColumnarRowFileInputFormat.java

@@ -143,13 +143,6 @@ public static SavepointReader read(
         this.stateBackend = stateBackend;
     }
 
-    /** @deprecated use {@link #readListState(OperatorIdentifier, String, TypeInformation)} */
-    @Deprecated
-    public <T> DataStream<T> readListState(String uid, String name, TypeInformation<T> typeInfo)
-            throws IOException {
-        return readListState(OperatorIdentifier.forUid(uid), name, typeInfo);
-    }
-
     /**
      * Read operator {@code ListState} from a {@code Savepoint}.
      *
@@ -166,17 +159,6 @@ public <T> DataStream<T> readListState(
         return readListState(identifier, typeInfo, new ListStateDescriptor<>(name, typeInfo));
     }
 
-    /**
-     * @deprecated use {@link #readListState(OperatorIdentifier, String, TypeInformation,
-     *     TypeSerializer)}
-     */
-    @Deprecated
-    public <T> DataStream<T> readListState(
-            String uid, String name, TypeInformation<T> typeInfo, TypeSerializer<T> serializer)
-            throws IOException {
-        return readListState(OperatorIdentifier.forUid(uid), name, typeInfo, serializer);
-    }
-
     /**
      * Read operator {@code ListState} from a {@code Savepoint} when a custom serializer was used;
      * e.g., a different serializer than the one returned by {@code
@@ -216,13 +198,6 @@ private <T> DataStream<T> readListState(
         return SourceBuilder.fromFormat(env, inputFormat, typeInfo);
     }
 
-    /** @deprecated use {@link #readUnionState(OperatorIdentifier, String, TypeInformation)} */
-    @Deprecated
-    public <T> DataStream<T> readUnionState(String uid, String name, TypeInformation<T> typeInfo)
-            throws IOException {
-        return readListState(OperatorIdentifier.forUid(uid), name, typeInfo);
-    }
-
     /**
      * Read operator {@code UnionState} from a {@code Savepoint}.
      *
@@ -239,16 +214,6 @@ public <T> DataStream<T> readUnionState(
         return readUnionState(identifier, typeInfo, new ListStateDescriptor<>(name, typeInfo));
     }
 
-    /**
-     * @deprecated use {@link #readUnionState(OperatorIdentifier, String, TypeInformation,
-     *     TypeSerializer)}
-     */
-    public <T> DataStream<T> readUnionState(
-            String uid, String name, TypeInformation<T> typeInfo, TypeSerializer<T> serializer)
-            throws IOException {
-        return readUnionState(OperatorIdentifier.forUid(uid), name, typeInfo, serializer);
-    }
-
     /**
      * Read operator {@code UnionState} from a {@code Savepoint} when a custom serializer was used;
      * e.g., a different serializer than the one returned by {@code
@@ -288,20 +253,6 @@ private <T> DataStream<T> readUnionState(
         return SourceBuilder.fromFormat(env, inputFormat, typeInfo);
     }
 
-    /**
-     * @deprecated use {@link #readBroadcastState(OperatorIdentifier, String, TypeInformation,
-     *     TypeInformation)}
-     */
-    @Deprecated
-    public <K, V> DataStream<Tuple2<K, V>> readBroadcastState(
-            String uid,
-            String name,
-            TypeInformation<K> keyTypeInfo,
-            TypeInformation<V> valueTypeInfo)
-            throws IOException {
-        return readBroadcastState(OperatorIdentifier.forUid(uid), name, keyTypeInfo, valueTypeInfo);
-    }
-
     /**
      * Read operator {@code BroadcastState} from a {@code Savepoint}.
      *
@@ -327,28 +278,6 @@ public <K, V> DataStream<Tuple2<K, V>> readBroadcastState(
                 new MapStateDescriptor<>(name, keyTypeInfo, valueTypeInfo));
     }
 
-    /**
-     * @deprecated use {@link #readBroadcastState(OperatorIdentifier, String, TypeInformation,
-     *     TypeInformation, TypeSerializer, TypeSerializer)}
-     */
-    @Deprecated
-    public <K, V> DataStream<Tuple2<K, V>> readBroadcastState(
-            String uid,
-            String name,
-            TypeInformation<K> keyTypeInfo,
-            TypeInformation<V> valueTypeInfo,
-            TypeSerializer<K> keySerializer,
-            TypeSerializer<V> valueSerializer)
-            throws IOException {
-        return readBroadcastState(
-                OperatorIdentifier.forUid(uid),
-                name,
-                keyTypeInfo,
-                valueTypeInfo,
-                keySerializer,
-                valueSerializer);
-    }
-
     /**
      * Read operator {@code BroadcastState} from a {@code Savepoint} when a custom serializer was
      * used; e.g., a different serializer than the one returned by {@code
@@ -399,13 +328,6 @@ private <K, V> DataStream<Tuple2<K, V>> readBroadcastState(
                 env, inputFormat, new TupleTypeInfo<>(keyTypeInfo, valueTypeInfo));
     }
 
-    /** @deprecated use {@link #readKeyedState(OperatorIdentifier, KeyedStateReaderFunction)} */
-    @Deprecated
-    public <K, OUT> DataStream<OUT> readKeyedState(
-            String uid, KeyedStateReaderFunction<K, OUT> function) throws IOException {
-        return readKeyedState(OperatorIdentifier.forUid(uid), function);
-    }
-
     /**
      * Read keyed state from an operator in a {@code Savepoint}.
      *
@@ -455,20 +377,6 @@ public <K, OUT> DataStream<OUT> readKeyedState(
         return readKeyedState(identifier, function, keyTypeInfo, outType);
     }
 
-    /**
-     * @deprecated use {@link #readKeyedState(OperatorIdentifier, KeyedStateReaderFunction,
-     *     TypeInformation, TypeInformation)}
-     */
-    @Deprecated
-    public <K, OUT> DataStream<OUT> readKeyedState(
-            String uid,
-            KeyedStateReaderFunction<K, OUT> function,
-            TypeInformation<K> keyTypeInfo,
-            TypeInformation<OUT> outTypeInfo)
-            throws IOException {
-        return readKeyedState(OperatorIdentifier.forUid(uid), function, keyTypeInfo, outTypeInfo);
-    }
-
     /**
      * Read keyed state from an operator in a {@code Savepoint}.
      *

flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/avro/ParquetAvroWriters.java

@@ -66,12 +66,6 @@ public class SavepointWriter {
     private final Map<OperatorIdentifier, OperatorIdentifier> uidTransformationMap =
             new HashMap<>();
 
-    /** @deprecated use {@link #fromExistingSavepoint(StreamExecutionEnvironment, String)} */
-    @Deprecated
-    public static SavepointWriter fromExistingSavepoint(String path) throws IOException {
-        return new SavepointWriter(readSavepointMetadata(path), null, null);
-    }
-
     /**
      * Loads an existing savepoint. Useful if you want to modify or extend the state of an existing
      * application. The savepoint will be written using the state backend defined via the clusters
@@ -87,24 +81,13 @@ public static SavepointWriter fromExistingSavepoint(
         return new SavepointWriter(readSavepointMetadata(path), null, executionEnvironment);
     }
 
-    /**
-     * @deprecated use {@link #fromExistingSavepoint(StreamExecutionEnvironment, String,
-     *     StateBackend)}
-     */
-    @Deprecated
-    public static SavepointWriter fromExistingSavepoint(String path, StateBackend stateBackend)
-            throws IOException {
-        return new SavepointWriter(readSavepointMetadata(path), stateBackend, null);
-    }
-
     /**
      * Loads an existing savepoint. Useful if you want to modify or extend the state of an existing
      * application.
      *
      * @param path The path to an existing savepoint on disk.
      * @param stateBackend The state backend of the savepoint.
      * @return A {@link SavepointWriter}.
-     * @see #fromExistingSavepoint(String)
      */
     public static SavepointWriter fromExistingSavepoint(
             StreamExecutionEnvironment executionEnvironment, String path, StateBackend stateBackend)
@@ -128,12 +111,6 @@ private static SavepointMetadataV2 readSavepointMetadata(String path) throws IOE
                 maxParallelism, metadata.getMasterStates(), metadata.getOperatorStates());
     }
 
-    /** @deprecated use {@link #newSavepoint(StreamExecutionEnvironment, int)} */
-    @Deprecated
-    public static SavepointWriter newSavepoint(int maxParallelism) {
-        return new SavepointWriter(createSavepointMetadata(maxParallelism), null, null);
-    }
-
     /**
      * Creates a new savepoint. The savepoint will be written using the state backend defined via
      * the clusters configuration.
@@ -149,12 +126,6 @@ public static SavepointWriter newSavepoint(
                 createSavepointMetadata(maxParallelism), null, executionEnvironment);
     }
 
-    /** @deprecated use {@link #newSavepoint(StreamExecutionEnvironment, StateBackend, int)} */
-    @Deprecated
-    public static SavepointWriter newSavepoint(StateBackend stateBackend, int maxParallelism) {
-        return new SavepointWriter(createSavepointMetadata(maxParallelism), stateBackend, null);
-    }
-
     /**
      * Creates a new savepoint.
      *
@@ -205,12 +176,6 @@ private SavepointWriter(
         this.executionEnvironment = executionEnvironment;
     }
 
-    /** @deprecated use {@link #removeOperator(OperatorIdentifier)} */
-    @Deprecated
-    public SavepointWriter removeOperator(String uid) {
-        return removeOperator(OperatorIdentifier.forUid(uid));
-    }
-
     /**
      * Drop an existing operator from the savepoint.
      *
@@ -222,12 +187,6 @@ public SavepointWriter removeOperator(OperatorIdentifier identifier) {
         return this;
     }
 
-    @Deprecated
-    public <T> SavepointWriter withOperator(
-            String uid, StateBootstrapTransformation<T> transformation) {
-        return withOperator(OperatorIdentifier.forUid(uid), transformation);
-    }
-
     /**
      * Adds a new operator to the savepoint.
      *