[GitHub] [flink] syhily commented on a diff in pull request #19972: [FLINK-27399][Connector/Pulsar] Change initial consuming position setting logic for better handle the checkpoint.

[GitBox]

syhily commented on code in PR #19972:
URL: https://github.com/apache/flink/pull/19972#discussion_r932145488


##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/stop/MessageIdStopCursor.java:
##########
@@ -32,21 +38,23 @@ public class MessageIdStopCursor implements StopCursor {
 
     private final MessageId messageId;
 
-    private final boolean exclusive;
+    private final boolean inclusive;
 
-    public MessageIdStopCursor(MessageId messageId) {
-        this(messageId, true);
-    }
+    public MessageIdStopCursor(MessageId messageId, boolean inclusive) {
+        MessageIdImpl idImpl = unwrapMessageId(messageId);
+        checkArgument(!earliest.equals(idImpl), "MessageId.earliest is not 
supported.");
+        checkArgument(
+                !latest.equals(idImpl),
+                "MessageId.latest is not supported, use 
LatestMessageStopCursor instead.");
 
-    public MessageIdStopCursor(MessageId messageId, boolean exclusive) {
-        this.messageId = messageId;
-        this.exclusive = exclusive;
+        this.messageId = idImpl;
+        this.inclusive = inclusive;
     }
 
     @Override
     public boolean shouldStop(Message<?> message) {
         MessageId id = message.getMessageId();
-        if (exclusive) {
+        if (inclusive) {
             return id.compareTo(messageId) > 0;

Review Comment:
   Nope, this is a wrong use in `PulsarPartitionSplitReaderBase`, we just fix 
the bugs.



-- 
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-unsubscr...@flink.apache.org

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