wenbingshen commented on code in PR #95:
URL:
https://github.com/apache/flink-connector-pulsar/pull/95#discussion_r1738486579
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/CursorPosition.java:
##########
@@ -104,6 +105,19 @@ private MessageId getActualMessageId(MessageIdAdv
messageIdImpl) {
if (include) {
return messageIdImpl;
} else {
+ // if the message is batched, should return next single message in
current batch.
+ if (messageIdImpl.getBatchIndex() >= 0
+ && messageIdImpl.getBatchSize() > 0
+ && messageIdImpl.getBatchIndex() !=
messageIdImpl.getBatchSize() - 1) {
+ return new BatchMessageIdImpl(
+ messageIdImpl.getLedgerId(),
+ messageIdImpl.getEntryId(),
+ messageIdImpl.getPartitionIndex(),
+ messageIdImpl.getBatchIndex() + 1,
+ messageIdImpl.getBatchSize(),
+ messageIdImpl.getAckSet());
Review Comment:
> Batch Message is a quite complex feature. How could we verify the
acknowledge is acked cumulately in batch message? It could be acked
individually.
1. In flink-pulsar-connector, the receive queue of the consumer is set to 1,
will the message before the current batchIndex not be confirmed?
2. Even if a single message is confirmed, the current connector does not
support BatchIndexAck.
3. The seek operation during task failure recovery cannot guarantee that
AckSet will work, as I said above.
4. The current connector's ack behavior is cumulative confirmation
Finally, if we don't change the seeking behavior of CursorPosition, we won't
be able to recover from AckSet regardless of the AckSet in the state.
The changes in this PR are valid under the current cumulative acknowledgment
behavior.
--
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]
