poorbarcode commented on code in PR #127:
URL:
https://github.com/apache/flink-connector-pulsar/pull/127#discussion_r4006848267
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarPartitionSplitReader.java:
##########
@@ -129,25 +131,71 @@ public RecordsWithSplitIds<Message<byte[]>> fetch()
throws IOException {
Deadline deadline =
Deadline.fromNow(sourceConfiguration.getMaxFetchTime());
// Consume messages from pulsar until it was woken up by flink reader.
+ CompletableFuture<Message<byte[]>> msgFuture = null;
+ MessageIdAdv latestMessageIdInTheCurrentFetch = null;
for (int messageNum = 0;
- messageNum < sourceConfiguration.getMaxFetchRecords() &&
deadline.hasTimeLeft();
- messageNum++) {
+ messageNum < sourceConfiguration.getMaxFetchRecords() &&
deadline.hasTimeLeft(); ) {
try {
int fetchTime = sourceConfiguration.getFetchOneMessageTime();
if (fetchTime <= 0) {
fetchTime = (int) deadline.timeLeftIfAny().toMillis();
}
-
- Message<byte[]> message = pulsarConsumer.receive(fetchTime,
TimeUnit.MILLISECONDS);
+ // (Highlight) The synchronised API "receive(Duration)" has a
bug, which may throw
+ // an error: "Try to
+ // reserve/release memory failed, the param memorySize is a
negative value".
+ // Here we use an asynchronous API.
+ msgFuture = pulsarConsumer.receiveAsync();
+ Message<byte[]> message = null;
+ try {
+ message = msgFuture.get(fetchTime, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ if (msgFuture.completeExceptionally(e)) {
+ throw e;
+ } else if (!msgFuture.isCompletedExceptionally()) {
+ message = msgFuture.get();
+ } else {
+ // throws error.
+ msgFuture.get();
+ }
+ }
if (message == null) {
break;
}
+ MessageIdAdv msgId = (MessageIdAdv) message.getMessageId();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "[{}] [{}] received a message {}:{}:{}/{}.",
+ pulsarConsumer.getTopic(),
+ pulsarConsumer.getSubscription(),
+ msgId.getLedgerId(),
+ msgId.getEntryId(),
+ msgId.getBatchIndex(),
+ msgId.getBatchSize());
+ }
+ // (Highlight) Since the connector will not acknowledge
messages immediately, when
+ // the pulsar consumer
+ // reconnects, it may receive repeated messages. We use the
following two mechanism
+ // to solve the
+ // repeated receiving messages issue.
+ if (latestMessageIdInTheCurrentFetch != null
+ && compareMessageIds(latestMessageIdInTheCurrentFetch,
msgId) >= 0) {
+ continue;
+ }
+ if (registeredSplit.getLatestConsumedId() != null
+ && compareMessageIds(
+ (MessageIdAdv)
registeredSplit.getLatestConsumedId(), msgId)
+ >= 0) {
+ continue;
+ }
Review Comment:
Thanks, addressed in 49e9bf8. I extracted the MessageId comparator and
isAtOrAfter helper, and simplified the duplicate-check condition.
--
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]
