[GitHub] [flink] AHeise commented on a change in pull request #17536: [FLINK-24530][datastream] GlobalCommitter might not commit all records on drain

[GitBox <gi...@apache.org>]

AHeise commented on a change in pull request #17536:
URL: https://github.com/apache/flink/pull/17536#discussion_r735863018



##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/ForwardCommittingHandler.java
##########
@@ -17,15 +17,19 @@
 
 package org.apache.flink.streaming.runtime.operators.sink;
 
+import org.apache.flink.api.connector.sink.Committer;
+
 import java.io.IOException;
 import java.util.List;
 
 /**
  * This committer handler simply forwards all committables downstream. It's used in {@link
  * SinkOperator} without committers but with downstream operators (in streaming, only global
  * committer on sink; in batch, committer or global committer present).
+ *
+ * @param <CommT> The input and output type of the {@link Committer}.
  */
-class ForwardCommittingHandler<CommT> extends AbstractCommitterHandler<CommT, CommT, CommT> {
+class ForwardCommittingHandler<CommT> extends AbstractCommitterHandler<CommT, Void> {

Review comment:
       They are still needed in the current `SinkOperator`:
   - In batch with committer, we need `ForwardingCommittingHandler`.
   - In stream/batch without any committer, we need `NoopCommittingHandler`.
   
   We could replace them by booleans but it's getting a bit ugly:
   - In streaming, we only emit on `notifyCheckpointCompleted`.
   - In batch, we only emit on `preSnapshotBarrier`.
   So you'd need two booleans afaik. It's certainly less code but I'm not sure if it's easier to understand.




-- 
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