You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "pvary (via GitHub)" <gi...@apache.org> on 2023/12/12 14:56:22 UTC
[PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
pvary opened a new pull request, #23912:
URL: https://github.com/apache/flink/pull/23912
## What is the purpose of the change
Mixin based implementation for [FLIP-372](https://cwiki.apache.org/confluence/display/FLINK/FLIP-372%3A+Allow+TwoPhaseCommittingSink+WithPreCommitTopology+to+alter+the+type+of+the+Committable)
For now, just checking backward compatibility of the API changes
## Brief change log
- Moving inner classes to main classes:
- `CommitterInitContext` from `Sink.CommitterInitContext`
- `CommittingSinkWriter` from `TwoPhaseCommittingSink.PrecommittingSinkWriter`
- `StatefulSinkWriter` from `StatefulSink.StatefulSinkWriter`
- `WriterInitContext` from `Sink.InitContext`
- Moving to mixin interfaces:
- `SupportsCommitter` from `TwoPhaseCommittingSink`
- `SupportsWriterState` from `StatefulSink`
- `SupportsPostCommitTopology` from `WithPostCommitToplogy`
- `SupportsPreCommitTopology` from `WithPreCommitTopology`
- `SupportsPreWriteTopology` from `WithPreWriteTopology`
- Deprecation methods with default implementations
- Few internal changes to make everything work
## Verifying this change
The CI will run all the tests to see if the change breaks anything
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? not documented
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary closed pull request #23912: [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach
URL: https://github.com/apache/flink/pull/23912
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #23912:
URL: https://github.com/apache/flink/pull/23912#issuecomment-1852213829
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64b7d707433f11bb4e1636796fb4c290376ac9cf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "64b7d707433f11bb4e1636796fb4c290376ac9cf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 64b7d707433f11bb4e1636796fb4c290376ac9cf UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1445462324
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/Sink.java:
##########
@@ -52,11 +54,30 @@ public interface Sink<InputT> extends Serializable {
* @param context the runtime context.
* @return A sink writer.
* @throws IOException for any failure during creation.
+ * @deprecated Please implement {@link #createWriter(WriterInitContext)}. For backward
+ * compatibility reasons - to keep {@link Sink} a functional interface - Flink did not
+ * provide a default implementation. New {@link Sink} implementations should implement this
+ * method, but it will not be used, and it will be removed in 1.20.0 release. Do not use
+ * {@link Override} annotation when implementing this method, to prevent compilation errors
+ * when migrating to 1.20.x release.
*/
+ @Deprecated
Review Comment:
do we use the attribute option `since` in other places? it might be useful to have that in the annotation itself
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1445463214
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/StatefulSink.java:
##########
@@ -36,49 +34,43 @@
* @param <WriterStateT> The type of the sink writer's state
*/
@PublicEvolving
-public interface StatefulSink<InputT, WriterStateT> extends Sink<InputT> {
+@Deprecated
Review Comment:
we could add reason for deprecation in the java doc of this interface
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1446081742
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/WithPreWriteTopology.java:
##########
@@ -21,18 +21,9 @@
import org.apache.flink.annotation.Experimental;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
-import org.apache.flink.streaming.api.datastream.DataStream;
/** Allows expert users to implement a custom topology before {@link SinkWriter}. */
@Experimental
-public interface WithPreWriteTopology<InputT> extends Sink<InputT> {
-
- /**
- * Adds an arbitrary topology before the writer. The topology may be used to repartition the
- * data.
- *
- * @param inputDataStream the stream of input records.
- * @return the custom topology before {@link SinkWriter}.
- */
- DataStream<InputT> addPreWriteTopology(DataStream<InputT> inputDataStream);
-}
+@Deprecated
Review Comment:
Added the `@deprecated` in #24022
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #23912:
URL: https://github.com/apache/flink/pull/23912#issuecomment-1864763831
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1446081187
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/StatefulSink.java:
##########
@@ -36,49 +34,43 @@
* @param <WriterStateT> The type of the sink writer's state
*/
@PublicEvolving
-public interface StatefulSink<InputT, WriterStateT> extends Sink<InputT> {
+@Deprecated
Review Comment:
added in #24022
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/SupportsWriterState.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.flink.api.connector.sink2;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.IOException;
+import java.util.Collection;
+
+/**
+ * A mixin interface for a {@link Sink} which supports a stateful {@link StatefulSinkWriter}.
+ *
+ * <p>The {@link Sink} needs to be serializable. All configuration should be validated eagerly. The
+ * respective sink writers are transient and will only be created in the subtasks on the
+ * taskmanagers.
+ *
+ * @param <InputT> The type of the sink's input
+ * @param <WriterStateT> The type of the sink writer's state
+ */
+@PublicEvolving
+public interface SupportsWriterState<InputT, WriterStateT> {
+
+ /**
+ * Create a {@link StatefulSinkWriter} from a recovered state.
+ *
+ * @param context the runtime context.
+ * @return A sink writer.
+ * @throws IOException for any failure during creation.
+ */
+ StatefulSinkWriter<InputT, WriterStateT> restoreWriter(
+ WriterInitContext context, Collection<WriterStateT> recoveredState) throws IOException;
+
+ /**
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * StatefulSinkWriter#snapshotState(long)} properly. The respective state is used in {@link
+ * #restoreWriter(WriterInitContext, Collection)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ SimpleVersionedSerializer<WriterStateT> getWriterStateSerializer();
+
+ /**
+ * A mix-in for {@link SupportsWriterState} that allows users to migrate from a sink with a
+ * compatible state to this sink.
+ */
+ @PublicEvolving
+ interface WithCompatibleState {
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
Review Comment:
Fixed in #24022
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1433804590
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/TwoPhaseCommittingSink.java:
##########
@@ -40,17 +37,9 @@
* @param <CommT> The type of the committables.
*/
@PublicEvolving
-public interface TwoPhaseCommittingSink<InputT, CommT> extends Sink<InputT> {
-
- /**
- * Creates a {@link PrecommittingSinkWriter} that creates committables on checkpoint or end of
- * input.
- *
- * @param context the runtime context.
- * @return A sink writer for the two-phase commit protocol.
- * @throws IOException for any failure during creation.
- */
- PrecommittingSinkWriter<InputT, CommT> createWriter(InitContext context) throws IOException;
Review Comment:
@leonardBang: Thanks for the review!
This was the exact issue which prevented us to evolve the API.
`TwoPhaseCommittingSink` and `StatefulSink` overrides the same `createWriter` method with different return types. If we need to evolve the `createWriter` method for whatever reason in the future, then we need to provide a default implementation for it, and the new method to allow backward compatibility.
If there is a sink which implements both `TwoPhaseCommittingSink` and `StatefulSink`, then it will not implement the new method, so the default implementation will be inherited from both `TwoPhaseCommittingSink` and `StatefulSink` which will cause a java compilation error.
That is why we need to remove this method, and use `instanceof` instead in the implementation, to check the types.
Of course, I am open for any suggestion, but sadly nothing viable come up during the discussions on the dev list [1], [2]
Thanks,
[1] - https://lists.apache.org/thread/344pzbrqbbb4w0sfj67km25msp7hxlyd
[2] - https://lists.apache.org/thread/h6nkgth838dlh5s90sd95zd6hlsxwx57
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1433677957
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/TwoPhaseCommittingSink.java:
##########
@@ -40,17 +37,9 @@
* @param <CommT> The type of the committables.
*/
@PublicEvolving
-public interface TwoPhaseCommittingSink<InputT, CommT> extends Sink<InputT> {
-
- /**
- * Creates a {@link PrecommittingSinkWriter} that creates committables on checkpoint or end of
- * input.
- *
- * @param context the runtime context.
- * @return A sink writer for the two-phase commit protocol.
- * @throws IOException for any failure during creation.
- */
- PrecommittingSinkWriter<InputT, CommT> createWriter(InitContext context) throws IOException;
Review Comment:
We'd better save the method which is better than `SinkWriter<InputT, CommT> createWriter(InitContext context)` as it return a `PrecommittingSinkWriter` which contains more information than `SinkWriter`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1446081006
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/Sink.java:
##########
@@ -52,11 +54,30 @@ public interface Sink<InputT> extends Serializable {
* @param context the runtime context.
* @return A sink writer.
* @throws IOException for any failure during creation.
+ * @deprecated Please implement {@link #createWriter(WriterInitContext)}. For backward
+ * compatibility reasons - to keep {@link Sink} a functional interface - Flink did not
+ * provide a default implementation. New {@link Sink} implementations should implement this
+ * method, but it will not be used, and it will be removed in 1.20.0 release. Do not use
+ * {@link Override} annotation when implementing this method, to prevent compilation errors
+ * when migrating to 1.20.x release.
*/
+ @Deprecated
Review Comment:
`since` is only supported since java9. The code should be compatible with java8, so we can not use `since` yet.
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/StatefulSink.java:
##########
@@ -36,49 +34,43 @@
* @param <WriterStateT> The type of the sink writer's state
*/
@PublicEvolving
-public interface StatefulSink<InputT, WriterStateT> extends Sink<InputT> {
+@Deprecated
Review Comment:
added in #24022 24022
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1433838959
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/TwoPhaseCommittingSink.java:
##########
@@ -40,17 +37,9 @@
* @param <CommT> The type of the committables.
*/
@PublicEvolving
-public interface TwoPhaseCommittingSink<InputT, CommT> extends Sink<InputT> {
-
- /**
- * Creates a {@link PrecommittingSinkWriter} that creates committables on checkpoint or end of
- * input.
- *
- * @param context the runtime context.
- * @return A sink writer for the two-phase commit protocol.
- * @throws IOException for any failure during creation.
- */
- PrecommittingSinkWriter<InputT, CommT> createWriter(InitContext context) throws IOException;
Review Comment:
Thanks @pvary for the explanation, makes sense to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1445465244
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/SupportsWriterState.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.flink.api.connector.sink2;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.IOException;
+import java.util.Collection;
+
+/**
+ * A mixin interface for a {@link Sink} which supports a stateful {@link StatefulSinkWriter}.
+ *
+ * <p>The {@link Sink} needs to be serializable. All configuration should be validated eagerly. The
+ * respective sink writers are transient and will only be created in the subtasks on the
+ * taskmanagers.
+ *
+ * @param <InputT> The type of the sink's input
+ * @param <WriterStateT> The type of the sink writer's state
+ */
+@PublicEvolving
+public interface SupportsWriterState<InputT, WriterStateT> {
+
+ /**
+ * Create a {@link StatefulSinkWriter} from a recovered state.
+ *
+ * @param context the runtime context.
+ * @return A sink writer.
+ * @throws IOException for any failure during creation.
+ */
+ StatefulSinkWriter<InputT, WriterStateT> restoreWriter(
+ WriterInitContext context, Collection<WriterStateT> recoveredState) throws IOException;
+
+ /**
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * StatefulSinkWriter#snapshotState(long)} properly. The respective state is used in {@link
+ * #restoreWriter(WriterInitContext, Collection)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ SimpleVersionedSerializer<WriterStateT> getWriterStateSerializer();
+
+ /**
+ * A mix-in for {@link SupportsWriterState} that allows users to migrate from a sink with a
+ * compatible state to this sink.
+ */
+ @PublicEvolving
+ interface WithCompatibleState {
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
Review Comment:
nit: `A collection of state...`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #23912:
URL: https://github.com/apache/flink/pull/23912#issuecomment-1883046267
Thanks @rodmeneses for the review!
The PR will merged in several parts, to help the reviewers.
The first part is #24022, addressed your comments there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #23912:
URL: https://github.com/apache/flink/pull/23912#issuecomment-1908498601
We merged this in several smaller parts
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1445464969
##########
flink-core/src/main/java/org/apache/flink/api/connector/sink2/SupportsWriterState.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.flink.api.connector.sink2;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.IOException;
+import java.util.Collection;
+
+/**
+ * A mixin interface for a {@link Sink} which supports a stateful {@link StatefulSinkWriter}.
+ *
+ * <p>The {@link Sink} needs to be serializable. All configuration should be validated eagerly. The
+ * respective sink writers are transient and will only be created in the subtasks on the
+ * taskmanagers.
+ *
+ * @param <InputT> The type of the sink's input
+ * @param <WriterStateT> The type of the sink writer's state
+ */
+@PublicEvolving
+public interface SupportsWriterState<InputT, WriterStateT> {
+
+ /**
+ * Create a {@link StatefulSinkWriter} from a recovered state.
+ *
+ * @param context the runtime context.
Review Comment:
nit: Java doc doesn't include second parameter `recoveredState`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33328] Allow TwoPhaseCommittingSink WithPreCommitTopology to alter the type of the Committable - mixin approach [flink]
Posted by "rodmeneses (via GitHub)" <gi...@apache.org>.
rodmeneses commented on code in PR #23912:
URL: https://github.com/apache/flink/pull/23912#discussion_r1445467435
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/WithPreWriteTopology.java:
##########
@@ -21,18 +21,9 @@
import org.apache.flink.annotation.Experimental;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
-import org.apache.flink.streaming.api.datastream.DataStream;
/** Allows expert users to implement a custom topology before {@link SinkWriter}. */
@Experimental
-public interface WithPreWriteTopology<InputT> extends Sink<InputT> {
-
- /**
- * Adds an arbitrary topology before the writer. The topology may be used to repartition the
- * data.
- *
- * @param inputDataStream the stream of input records.
- * @return the custom topology before {@link SinkWriter}.
- */
- DataStream<InputT> addPreWriteTopology(DataStream<InputT> inputDataStream);
-}
+@Deprecated
Review Comment:
we could explain in the javadoc why this interface is deprecated, specially since it's a new addition
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org