Skip to content

Implemented AtomicBoolean of havePendingRead in readMoreEntries() #10411

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
devinbost wants to merge 1 commit into from
Closed

Implemented AtomicBoolean of havePendingRead in readMoreEntries() #10411

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions ...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
Expand Down Expand Up @@ -81,7 +82,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul

private Optional<DelayedDeliveryTracker> delayedDeliveryTracker = Optional.empty();

protected volatile boolean havePendingRead = false;
protected volatile AtomicBoolean havePendingRead = new AtomicBoolean(false);
protected volatile boolean havePendingReplayRead = false;
protected boolean shouldRewindBeforeReadingOrReplaying = false;
protected final String name;
Expand Down Expand Up @@ -134,7 +135,7 @@ public synchronized void addConsumer(Consumer consumer) throws BrokerServiceExce
return;
}
if (consumerList.isEmpty()) {
if (havePendingRead || havePendingReplayRead) {
if (havePendingRead.get() || havePendingReplayRead) {
// There is a pending read from previous run. We must wait for it to complete and then rewind
shouldRewindBeforeReadingOrReplaying = true;
} else {
Expand Down Expand Up @@ -251,12 +252,11 @@ public void readMoreEntries() {
} else if (BLOCKED_DISPATCHER_ON_UNACKMSG_UPDATER.get(this) == TRUE) {
log.warn("[{}] Dispatcher read is blocked due to unackMessages {} reached to max {}", name,
totalUnackedMessages, topic.getMaxUnackedMessagesOnSubscription());
} else if (!havePendingRead) {
} else if (havePendingRead.compareAndSet(false, true)) {
if (log.isDebugEnabled()) {
log.debug("[{}] Schedule read of {} messages for {} consumers", name, messagesToRead,
consumerList.size());
}
havePendingRead = true;
cursor.asyncReadEntriesOrWait(messagesToRead, serviceConfig.getDispatcherMaxReadSizeBytes(),
this,
ReadType.Normal, topic.getMaxReadPosition());
Expand Down Expand Up @@ -402,8 +402,8 @@ public synchronized CompletableFuture<Void> disconnectAllConsumers(boolean isRes

@Override
protected void cancelPendingRead() {
if (havePendingRead && cursor.cancelPendingReadRequest()) {
havePendingRead = false;
if (havePendingRead.get() && cursor.cancelPendingReadRequest()) {
havePendingRead.set(false);
}
}

Expand Down Expand Up @@ -432,7 +432,7 @@ public SubType getType() {
public synchronized void readEntriesComplete(List<Entry> entries, Object ctx) {
ReadType readType = (ReadType) ctx;
if (readType == ReadType.Normal) {
havePendingRead = false;
havePendingRead.set(false);
} else {
havePendingReplayRead = false;
}
Expand Down Expand Up @@ -594,7 +594,7 @@ public synchronized void readEntriesFailed(ManagedLedgerException exception, Obj
}

if (readType == ReadType.Normal) {
havePendingRead = false;
havePendingRead.set(false);
} else {
havePendingReplayRead = false;
if (exception instanceof ManagedLedgerException.InvalidReplayPositionException) {
Expand All @@ -611,7 +611,7 @@ public synchronized void readEntriesFailed(ManagedLedgerException exception, Obj

topic.getBrokerService().executor().schedule(() -> {
synchronized (PersistentDispatcherMultipleConsumers.this) {
if (!havePendingRead) {
if (!havePendingRead.get()) {
log.info("[{}] Retrying read operation", name);
readMoreEntries();
} else {
Expand Down Expand Up @@ -839,7 +839,7 @@ public boolean checkAndUnblockIfStuck() {
return false;
}
// consider dispatch is stuck if : dispatcher has backlog, available-permits and there is no pending read
if (totalAvailablePermits > 0 && !havePendingReplayRead && !havePendingRead
if (totalAvailablePermits > 0 && !havePendingReplayRead && !havePendingRead.get()
&& cursor.getNumberOfEntriesInBacklog(false) > 0) {
log.warn("{}-{} Dispatcher is stuck and unblocking by issuing reads", topic.getName(), name);
readMoreEntries();
Expand Down
13 changes: 6 additions & 7 deletions ...ache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ public synchronized void readEntryComplete(Entry entry, PendingReadEntryRequest
if (ctx.isLast()) {
readFailureBackoff.reduceToHalf();
if (readType == ReadType.Normal) {
havePendingRead = false;
havePendingRead.set(false);
} else {
havePendingReplayRead = false;
}
Expand Down Expand Up @@ -99,11 +99,11 @@ public synchronized void readEntryComplete(Entry entry, PendingReadEntryRequest
*/
@Override
public void canReadMoreEntries(boolean withBackoff) {
havePendingRead = false;
havePendingRead.set(false);
topic.getBrokerService().executor().schedule(() -> {
topic.getBrokerService().getTopicOrderedExecutor().executeOrdered(topic.getName(), SafeRun.safeRun(() -> {
synchronized (PersistentStreamingDispatcherMultipleConsumers.this) {
if (!havePendingRead) {
if (!havePendingRead.get()) {
log.info("[{}] Scheduling read operation", name);
readMoreEntries();
} else {
Expand All @@ -129,8 +129,8 @@ public void notifyConsumersEndOfTopic() {

@Override
protected void cancelPendingRead() {
if (havePendingRead && streamingEntryReader.cancelReadRequests()) {
havePendingRead = false;
if (havePendingRead.get() && streamingEntryReader.cancelReadRequests()) {
havePendingRead.set(false);
}
}

Expand Down Expand Up @@ -170,12 +170,11 @@ public void readMoreEntries() {
} else if (BLOCKED_DISPATCHER_ON_UNACKMSG_UPDATER.get(this) == TRUE) {
log.warn("[{}] Dispatcher read is blocked due to unackMessages {} reached to max {}", name,
totalUnackedMessages, topic.getMaxUnackedMessagesOnSubscription());
} else if (!havePendingRead) {
} else if (havePendingRead.compareAndSet(false, true)) {
if (log.isDebugEnabled()) {
log.debug("[{}] Schedule read of {} messages for {} consumers", name, messagesToRead,
consumerList.size());
}
havePendingRead = true;
streamingEntryReader.asyncReadEntries(messagesToRead, serviceConfig.getDispatcherMaxReadSizeBytes(),
ReadType.Normal);
} else {
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1861,7 +1861,7 @@ public CompletableFuture<PersistentTopicInternalStats> getInternalStats(boolean
if (sub.getDispatcher() instanceof PersistentDispatcherMultipleConsumers) {
PersistentDispatcherMultipleConsumers dispatcher = (PersistentDispatcherMultipleConsumers) sub
.getDispatcher();
cs.subscriptionHavePendingRead = dispatcher.havePendingRead;
cs.subscriptionHavePendingRead = dispatcher.havePendingRead.get();
cs.subscriptionHavePendingReplayRead = dispatcher.havePendingReplayRead;
} else if (sub.getDispatcher() instanceof PersistentDispatcherSingleActiveConsumer) {
PersistentDispatcherSingleActiveConsumer dispatcher = (PersistentDispatcherSingleActiveConsumer) sub
Expand Down
4 changes: 2 additions & 2 deletions ...broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ public void testUnblockStuckSubscription() throws Exception {
consumer2.close();

// block sub to read messages
sharedDispatcher.havePendingRead = true;
sharedDispatcher.havePendingRead.set(true);
failOverDispatcher.havePendingRead = true;

producer.newMessage().value("test").eventTime(5).send();
Expand All @@ -146,7 +146,7 @@ public void testUnblockStuckSubscription() throws Exception {
assertNull(msg);

// allow reads but dispatchers are still blocked
sharedDispatcher.havePendingRead = false;
sharedDispatcher.havePendingRead.set(false);
failOverDispatcher.havePendingRead = false;

// run task to unblock stuck dispatcher: first iteration sets the lastReadPosition and next iteration will
Expand Down