Vanlightly commented on a change in pull request #11691:
URL: https://github.com/apache/pulsar/pull/11691#discussion_r690942099
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -204,26 +205,11 @@ protected ConsumerBase(PulsarClientImpl client, String
topic, ConsumerConfigurat
}
}
- protected CompletableFuture<Message<T>> peekPendingReceive() {
- CompletableFuture<Message<T>> receivedFuture = null;
- while (receivedFuture == null) {
- receivedFuture = pendingReceives.peek();
- if (receivedFuture == null) {
- break;
- }
- // skip done futures (cancelling a future could mark it done)
- if (receivedFuture.isDone()) {
- CompletableFuture<Message<T>> removed = pendingReceives.poll();
- if (removed != receivedFuture) {
- log.error("Bug! Removed future wasn't the expected one.
expected={} removed={}", receivedFuture, removed);
- }
- receivedFuture = null;
- }
- }
- return receivedFuture;
+ protected boolean hasNextPendingReceive() {
Review comment:
I left these checks without a lock as they are only an indicator that
there may be receives to be notified, but in the end, it is a call to poll()
that counts, which is thread safe and null checks are always performed.
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
##########
@@ -790,31 +778,20 @@ protected void notifyPendingBatchReceivedCallBack() {
}
}
- private OpBatchReceive<T> peekNextBatchReceive() {
- OpBatchReceive<T> opBatchReceive = null;
- while (opBatchReceive == null) {
- opBatchReceive = pendingBatchReceives.peek();
- // no entry available
- if (opBatchReceive == null) {
- return null;
- }
- // remove entries where future is null or has been completed
(cancel / timeout)
- if (opBatchReceive.future == null ||
opBatchReceive.future.isDone()) {
- OpBatchReceive<T> removed = pendingBatchReceives.poll();
- if (removed != opBatchReceive) {
- log.error("Bug: Removed entry wasn't the expected one.
expected={}, removed={}", opBatchReceive, removed);
- }
- opBatchReceive = null;
- }
- }
- return opBatchReceive;
+ private boolean hasNextBatchReceive() {
+ return !pendingBatchReceives.isEmpty();
}
- private OpBatchReceive<T> pollNextBatchReceive() {
+ private OpBatchReceive<T> nextBatchReceive() {
OpBatchReceive<T> opBatchReceive = null;
while (opBatchReceive == null) {
- opBatchReceive = pendingBatchReceives.poll();
+ try {
+ pendingBatchLock.lock();
Review comment:
Yes, any call to `pendingBatchReceives.poll()` requires a lock in order
to protect the peek and poll that caused the bug.
--
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: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
