You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2019/04/12 14:27:30 UTC
[flink] branch release-1.7 updated: [FLINK-12169] [javadocs]
improve javadoc of MessageAcknowledgingSourceBase
This is an automated email from the ASF dual-hosted git repository.
jincheng pushed a commit to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.7 by this push:
new c1ca910 [FLINK-12169] [javadocs] improve javadoc of MessageAcknowledgingSourceBase
c1ca910 is described below
commit c1ca910c62f97e06fa76ef548f4084277c483905
Author: Konstantin Knauf <kn...@gmail.com>
AuthorDate: Fri Apr 12 10:20:35 2019 +0200
[FLINK-12169] [javadocs] improve javadoc of MessageAcknowledgingSourceBase
This closes #8155.
---
.../api/functions/source/MessageAcknowledgingSourceBase.java | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java
index a0b3129..36a8c9f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java
@@ -70,8 +70,9 @@ import java.util.Set;
* 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 abstract class MessageAcknowledgingSourceBase<Type, UId>
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.
*/