This is an automated email from the ASF dual-hosted git repository. fpaul pushed a commit to branch release-1.14 in repository https://gitbox.apache.org/repos/asf/flink.git
commit b0f56390b5ae768851c58b811c3b0f3dec96f294 Author: Fabian Paul <[email protected]> AuthorDate: Fri Nov 26 11:42:36 2021 +0100 [FLINK-24596][core] Introduce SerializableFunction and unify usages --- .../streaming/connectors/elasticsearch/table/KeyExtractor.java | 3 ++- .../org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java | 3 ++- .../java/org/apache/flink/util/function}/SerializableFunction.java | 2 +- .../runtime/operators/coordination/TestingOperatorCoordinator.java | 2 +- .../java/org/apache/flink/runtime/util/SerializableFunction.java | 7 ++++++- .../flink/connector/datagen/table/types/DataGeneratorMapper.java | 7 +------ 6 files changed, 13 insertions(+), 11 deletions(-) diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractor.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractor.java index ae7c522..ca9c297 100644 --- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractor.java +++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/KeyExtractor.java @@ -24,6 +24,7 @@ import org.apache.flink.table.api.TableSchema; import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.logical.DistinctType; import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.function.SerializableFunction; import java.io.Serializable; import java.time.Duration; @@ -37,7 +38,7 @@ import java.util.function.Function; /** An extractor for a Elasticsearch key from a {@link RowData}. */ @Internal -class KeyExtractor implements Function<RowData, String>, Serializable { +class KeyExtractor implements SerializableFunction<RowData, String> { private final FieldFormatter[] fieldFormatters; private final String keyDelimiter; diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java index 121800e..e7b7514 100644 --- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java +++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcOutputFormat.java @@ -40,6 +40,7 @@ import org.apache.flink.connector.jdbc.utils.JdbcUtils; import org.apache.flink.types.Row; import org.apache.flink.util.Preconditions; import org.apache.flink.util.concurrent.ExecutorThreadFactory; +import org.apache.flink.util.function.SerializableFunction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -99,7 +100,7 @@ public class JdbcOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStatementExe * @param <T> The type of instance. */ public interface StatementExecutorFactory<T extends JdbcBatchStatementExecutor<?>> - extends Function<RuntimeContext, T>, Serializable {} + extends SerializableFunction<RuntimeContext, T> {} private static final long serialVersionUID = 1L; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializableFunction.java b/flink-core/src/main/java/org/apache/flink/util/function/SerializableFunction.java similarity index 96% copy from flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializableFunction.java copy to flink-core/src/main/java/org/apache/flink/util/function/SerializableFunction.java index 75d7f91..2eee42e 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializableFunction.java +++ b/flink-core/src/main/java/org/apache/flink/util/function/SerializableFunction.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.flink.runtime.util; +package org.apache.flink.util.function; import org.apache.flink.annotation.Public; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/TestingOperatorCoordinator.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/TestingOperatorCoordinator.java index 50b3046..15d0dc6 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/TestingOperatorCoordinator.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/TestingOperatorCoordinator.java @@ -19,7 +19,7 @@ package org.apache.flink.runtime.operators.coordination; import org.apache.flink.runtime.jobgraph.OperatorID; -import org.apache.flink.runtime.util.SerializableFunction; +import org.apache.flink.util.function.SerializableFunction; import javax.annotation.Nullable; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializableFunction.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializableFunction.java index 75d7f91..00d2326 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializableFunction.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializableFunction.java @@ -23,7 +23,12 @@ import org.apache.flink.annotation.Public; import java.io.Serializable; import java.util.function.Function; -/** A {@link Function} that is also {@link Serializable}. */ +/** + * A {@link Function} that is also {@link Serializable}. + * + * @deprecated Please use {@link org.apache.flink.util.function.SerializableFunction} + */ +@Deprecated @Public @FunctionalInterface public interface SerializableFunction<T, R> extends Function<T, R>, Serializable {} diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/types/DataGeneratorMapper.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/types/DataGeneratorMapper.java index 19fb9b4..4e1c61d 100644 --- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/types/DataGeneratorMapper.java +++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/connector/datagen/table/types/DataGeneratorMapper.java @@ -22,9 +22,7 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.functions.RuntimeContext; import org.apache.flink.runtime.state.FunctionInitializationContext; import org.apache.flink.streaming.api.functions.source.datagen.DataGenerator; - -import java.io.Serializable; -import java.util.function.Function; +import org.apache.flink.util.function.SerializableFunction; /** Utility for mapping the output of a {@link DataGenerator}. */ @Internal @@ -55,7 +53,4 @@ public class DataGeneratorMapper<A, B> implements DataGenerator<B> { public B next() { return mapper.apply(generator.next()); } - - /** A simple serializable function. */ - public interface SerializableFunction<A, B> extends Function<A, B>, Serializable {} }
