tisonkun commented on code in PR #19972:
URL: https://github.com/apache/flink/pull/19972#discussion_r932127024
##########
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:
It seems we somehow revert the boolean logic here?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
