ferenc-csaky commented on code in PR #127:
URL:
https://github.com/apache/flink-connector-pulsar/pull/127#discussion_r4004462461
##########
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:
I believe we can improve this logic a bit, to make the flow and also the
comparing logic easier to read. For example by defining a comparator and a
helper method that explaing the `if` conditions better:
```java
private static final Comparator<MessageIdAdv> MESSAGE_ID_COMPARATOR =
(messageId1, messageId2) -> {
int ledgerComparison =
Long.compare(messageId1.getLedgerId(),
messageId2.getLedgerId());
if (ledgerComparison != 0) {
return ledgerComparison;
}
int entryComparison = Long.compare(messageId1.getEntryId(),
messageId2.getEntryId());
if (entryComparison != 0) {
return entryComparison;
}
if (messageId1 instanceof BatchMessageIdImpl
&& messageId2 instanceof BatchMessageIdImpl) {
return messageId1.getBatchIndex() -
messageId2.getBatchIndex();
}
return 0;
};
/**
* Returns whether the previous message ID is at or after the given message
ID.
*/
private static boolean isAtOrAfter(MessageIdAdv prevMessageId, MessageIdAdv
messageId) {
return prevMessageId != null
&& MESSAGE_ID_COMPARATOR.compare(prevMessageId, messageId) >= 0;
}
// then using it like:
if (isAtOrAfter(latestMessageIdInTheCurrentFetch, msgId)
|| isAtOrAfter((MessageIdAdv) registeredSplit.getLatestConsumedId(),
msgId)) {
continue;
}
```
##########
flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE:
##########
@@ -6,15 +6,11 @@ The Apache Software Foundation (http://www.apache.org/).
This project bundles the following dependencies under the Apache Software
License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
-- com.fasterxml.jackson.core:jackson-annotations:2.13.4
-- org.apache.pulsar:pulsar-client-admin-api:3.0.5
-- org.apache.pulsar:pulsar-client-all:3.0.5
-- org.apache.pulsar:pulsar-client-api:3.0.5
-
-This project bundles the following dependencies under the Bouncy Castle
license.
-See bundled license files for details.
-
-- org.bouncycastle:bcpkix-jdk15on:1.69
-- org.bouncycastle:bcprov-ext-jdk15on:1.69
-- org.bouncycastle:bcprov-jdk15on:1.69
-- org.bouncycastle:bcutil-jdk15on:1.69
Review Comment:
These are still included in the SQL connector JAR, so we should not remove
it from the NOTICE
--
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]
