[FLINK-12169] [javadocs] improve javadoc of MessageAcknowledgingSourc…

…eBase This closes apache#8155.
deepfield · Apr 12, 2019 · 4c1311e · 4c1311e
1 parent 264a0c9
commit 4c1311e
Showing 1 changed file with 5 additions and 3 deletions.
diff --git a/.../java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java b/.../java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java
@@ -70,8 +70,9 @@
  * while (running) {
  * Message msg = queue.retrieve();
  * synchronized (ctx.getCheckpointLock()) {
- * ctx.collect(msg.getMessageData());
- * addId(msg.getMessageId());
+ * if (addId(msg.getMessageId())) {
+ * ctx.collect(msg.getMessageData());
+ * }
  * }
  * }
  * }
@@ -187,7 +188,8 @@ public void close() throws Exception {
  protected abstract void acknowledgeIDs(long checkpointId, Set<UId> uIds);
 
  /**
- * Adds an ID to be stored with the current checkpoint.
+ * Adds an ID to be stored with the current checkpoint. In order to achieve exactly-once guarantees, implementing
+ * classes should only emit records with IDs for which this method return true.
  * @param uid The ID to add.
  * @return True if the id has not been processed previously.
  */