[refactor][client] Remove useless locks in batch receiving.
@@ -38,8 +38,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
import lombok.Getter;
import lombok.Setter;
import org.apache.pulsar.client.api.BatchReceivePolicy;
@@ -95,7 +93,6 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
.newUpdater(ConsumerBase.class, "incomingMessagesSize");
protected volatile long incomingMessagesSize = 0;
protected volatile Timeout batchReceiveTimeout = null;
- protected final Lock reentrantLock = new ReentrantLock();
protected static final AtomicLongFieldUpdater<ConsumerBase> CONSUMER_EPOCH =
AtomicLongFieldUpdater.newUpdater(ConsumerBase.class, "consumerEpoch");
@@ -910,13 +907,7 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
if (opBatchReceive == null) {
return;
}
-
- reentrantLock.lock();
- try {
notifyPendingBatchReceivedCallBack(opBatchReceive.future);
- } finally {
- reentrantLock.unlock();
- }
}
private boolean hasNextBatchReceive() {
