AHeise commented on a change in pull request #17536:
URL: https://github.com/apache/flink/pull/17536#discussion_r735869185
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -72,4 +78,32 @@ default void snapshotState(StateSnapshotContext context)
throws Exception {}
* @return successfully retried committables that is sent downstream.
*/
Collection<CommT> retry() throws IOException, InterruptedException;
+
+ /**
+ * The serializable factory of a committer handler such that the stateful
implementations of
+ * {@link CommitterHandler} do not need to be {@link Serializable}
themselves.
+ */
+ interface Factory<Sink, CommT> extends Serializable {
+ CommitterHandler<CommT> create(Sink sink) throws IOException;
+
+ default <T> T checkSerializerPresent(Optional<T> optional, boolean
global) {
+ String scope = global ? " global" : "";
+ checkState(
+ optional.isPresent(),
+ "Internal error: a%s committer should only be created if
the sink has a%s committable serializer.",
+ scope,
+ scope);
+ return optional.get();
+ }
+
+ default <T> T checkCommitterPresent(Optional<T> optional, boolean
global) {
+ String scope = global ? " global" : "";
+ checkState(
+ optional.isPresent(),
+ "Expected a%s committer because%s committable serializer
is set.",
+ scope,
+ scope);
+ return optional.get();
+ }
+ }
Review comment:
How about I change the Factory interface to an abstract class? It might
be weird because they are default methods?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
