[GitHub] [flink-connector-pulsar] syhily commented on a diff in pull request #54: [FLINK-32612] Upgrade Pulsar version to 3.0.0

["syhily (via GitHub)" <gi...@apache.org>]

syhily commented on code in PR #54:
URL: https://github.com/apache/flink-connector-pulsar/pull/54#discussion_r1265759429


##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/stop/MessageIdStopCursor.java:
##########
@@ -43,8 +45,12 @@ public MessageIdStopCursor(MessageId messageId, boolean inclusive) {
         checkArgument(!earliest.equals(messageId), "MessageId.earliest is not supported.");
         checkArgument(!latest.equals(messageId), "Use LatestMessageStopCursor instead.");
 
-        this.messageId = convertToMessageIdImpl(messageId);
         this.inclusive = inclusive;
+        try {
+            this.messageId = MessageId.fromByteArray(messageId.toByteArray());
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }

Review Comment:
   We don't need that acker field indeed. BTW, this is just used for checking this is a valid message id. Flink will distribute the application to the downstream TMs which requires the serialization on messge id. The acker will be omitted finally.



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