You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2024/01/15 03:28:12 UTC
(flink) branch master updated: [FLINK-32978] Revert the breaking change to the public implementations of RichFunction
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 1d6150f386d [FLINK-32978] Revert the breaking change to the public implementations of RichFunction
1d6150f386d is described below
commit 1d6150f386d9c9ec61f4ab30853b915de7712047
Author: Wencong Liu <li...@163.com>
AuthorDate: Tue Jan 9 20:56:53 2024 +0800
[FLINK-32978] Revert the breaking change to the public implementations of RichFunction
This closes #24067
---
.../functions/sink/OutputFormatSinkFunction.java | 21 ++++++++++++++++++---
.../api/functions/sink/PrintSinkFunction.java | 22 +++++++++++++++++++---
.../api/functions/sink/SocketClientSink.java | 17 ++++++++++++++---
.../source/FromSplittableIteratorFunction.java | 20 ++++++++++++++++++--
.../MultipleIdsMessageAcknowledgingSourceBase.java | 22 +++++++++++++++++++---
5 files changed, 88 insertions(+), 14 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java
index f7c8b2c55e9..c0894e52414 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java
@@ -19,7 +19,6 @@ package org.apache.flink.streaming.api.functions.sink;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.io.CleanupWhenUnsuccessful;
import org.apache.flink.api.common.io.OutputFormat;
@@ -58,10 +57,26 @@ public class OutputFormatSinkFunction<IN> extends RichSinkFunction<IN>
this.format = format;
}
+ /**
+ * Initialization method for the {@link OutputFormatSinkFunction}.
+ *
+ * @param parameters The configuration containing the parameters attached to the contract.
+ * @throws Exception if an error happens.
+ * @deprecated This method is deprecated since Flink 1.19. The users are recommended to
+ * implement {@code open(OpenContext openContext)} and override {@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
@Override
- public void open(OpenContext openContext) throws Exception {
+ public void open(Configuration parameters) throws Exception {
RuntimeContext context = getRuntimeContext();
- format.configure(new Configuration());
+ format.configure(parameters);
int indexInSubtaskGroup = context.getTaskInfo().getIndexOfThisSubtask();
int currentNumberOfSubtasks = context.getTaskInfo().getNumberOfParallelSubtasks();
format.open(
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java
index 6f8c6ac46e2..1843b46fc4c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java
@@ -19,8 +19,8 @@ package org.apache.flink.streaming.api.functions.sink;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.SupportsConcurrentExecutionAttempts;
-import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.util.PrintSinkOutputWriter;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
/**
@@ -66,9 +66,25 @@ public class PrintSinkFunction<IN> extends RichSinkFunction<IN>
writer = new PrintSinkOutputWriter<>(sinkIdentifier, stdErr);
}
+ /**
+ * Initialization method for the {@link PrintSinkFunction}.
+ *
+ * @param parameters The configuration containing the parameters attached to the contract.
+ * @throws Exception if an error happens.
+ * @deprecated This method is deprecated since Flink 1.19. The users are recommended to
+ * implement {@code open(OpenContext openContext)} and override {@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
@Override
- public void open(OpenContext openContext) throws Exception {
- super.open(openContext);
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
StreamingRuntimeContext context = (StreamingRuntimeContext) getRuntimeContext();
writer.open(
context.getTaskInfo().getIndexOfThisSubtask(),
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
index 930a0bf7cb7..e31c33fce07 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
@@ -18,8 +18,8 @@
package org.apache.flink.streaming.api.functions.sink;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.SerializableObject;
import org.slf4j.Logger;
@@ -130,10 +130,21 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
/**
* Initialize the connection with the Socket in the server.
*
- * @param openContext the context.
+ * @param parameters The configuration containing the parameters attached to the contract.
+ * @throws Exception if an error happens.
+ * @deprecated This method is deprecated since Flink 1.19. The users are recommended to
+ * implement {@code open(OpenContext openContext)} and override {@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
@Override
- public void open(OpenContext openContext) throws Exception {
+ public void open(Configuration parameters) throws Exception {
try {
synchronized (lock) {
createConnection();
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java
index a44e60c7e09..78962212951 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java
@@ -18,7 +18,7 @@
package org.apache.flink.streaming.api.functions.source;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.SplittableIterator;
import java.util.Iterator;
@@ -46,8 +46,24 @@ public class FromSplittableIteratorFunction<T> extends RichParallelSourceFunctio
this.fullIterator = iterator;
}
+ /**
+ * Initialization method for the {@link FromSplittableIteratorFunction}.
+ *
+ * @param parameters The configuration containing the parameters attached to the contract.
+ * @throws Exception if an error happens.
+ * @deprecated This method is deprecated since Flink 1.19. The users are recommended to
+ * implement {@code open(OpenContext openContext)} and override {@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
@Override
- public void open(OpenContext openContext) throws Exception {
+ public void open(Configuration parameters) throws Exception {
int numberOfSubTasks = getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks();
int indexofThisSubTask = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
iterator = fullIterator.split(numberOfSubTasks)[indexofThisSubTask];
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java
index c0262775aaf..aec109eeb58 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java
@@ -19,9 +19,9 @@
package org.apache.flink.streaming.api.functions.source;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.slf4j.Logger;
@@ -92,9 +92,25 @@ public abstract class MultipleIdsMessageAcknowledgingSourceBase<Type, UId, Sessi
super(idTypeInfo);
}
+ /**
+ * Initialization method for the {@link MultipleIdsMessageAcknowledgingSourceBase}.
+ *
+ * @param parameters The configuration containing the parameters attached to the contract.
+ * @throws Exception if an error happens.
+ * @deprecated This method is deprecated since Flink 1.19. The users are recommended to
+ * implement {@code open(OpenContext openContext)} and override {@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
@Override
- public void open(OpenContext openContext) throws Exception {
- super.open(openContext);
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
sessionIds = new ArrayList<>(64);
sessionIdsPerSnapshot = new ArrayDeque<>();
}