[GitHub] [flink-statefun] igalshilman commented on a change in pull request #105: [FLINK-17533] Add support for multiple concurrent checkpoints

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

igalshilman commented on a change in pull request #105:
URL: https://github.com/apache/flink-statefun/pull/105#discussion_r425656646



##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/feedback/FeedbackUnionOperator.java
##########
@@ -84,14 +90,30 @@ public void processFeedback(T element) {
     if (closedOrDisposed) {
       return;
     }
-    if (isBarrierMessage.test(element)) {
-      feedbackLogger.commit();
+    OptionalLong maybeCheckpoint = isBarrierMessage.apply(element);
+    if (maybeCheckpoint.isPresent()) {
+      commitCheckpoint(maybeCheckpoint.getAsLong());
     } else {
       sendDownstream(element);
-      feedbackLogger.append(element);
+      appendToCheckpoint(element);
+    }
+  }
+
+  private void appendToCheckpoint(T element) {
+    for (UnboundedFeedbackLogger<T> logger : uncompletedCheckpoints.values()) {
+      logger.append(element);
     }
   }
 
+  private void commitCheckpoint(final long checkpointId) {
+    UnboundedFeedbackLogger<T> logger = uncompletedCheckpoints.remove(checkpointId);

Review comment:
       It would eventually be called when a `checkpointId` succeed with a greater value.
   You can check for a reference any of the `TwoPhaseCommit` operators that Flink has for a similar logic.
   (for example StreamingFlieSink)




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

For queries about this service, please contact Infrastructure at:
users@infra.apache.org