[cleanup][broker] Follow up on to restore some behavior in PersistentDispatcherMultipleConsumers class
@@ -93,7 +93,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
protected volatile PositionImpl minReplayedPosition = null;
protected boolean shouldRewindBeforeReadingOrReplaying = false;
protected final String name;
- protected volatile boolean sendInProgress;
+ protected boolean sendInProgress;
protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers>
TOTAL_AVAILABLE_PERMITS_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class,
@@ -248,8 +248,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
* We should not call readMoreEntries() recursively in the same thread as there is a risk of StackOverflowError.
*
*/
- public void readMoreEntiresAsync() {
- topic.getBrokerService().executor().execute(() -> readMoreEntries());
+ public void readMoreEntriesAsync() {
+ topic.getBrokerService().executor().execute(this::readMoreEntries);
}
public synchronized void readMoreEntries() {
@@ -295,7 +295,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
// next entries as readCompletedEntries-callback was never called
if ((messagesToReplayNow.size() - deletedMessages.size()) == 0) {
havePendingReplayRead = false;
- readMoreEntiresAsync();
+ readMoreEntriesAsync();
}
} else if (BLOCKED_DISPATCHER_ON_UNACKMSG_UPDATER.get(this) == TRUE) {
if (log.isDebugEnabled()) {
@@ -550,6 +550,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (serviceConfig.isDispatcherDispatchMessagesInSubscriptionThread()) {
// setting sendInProgress here, because sendMessagesToConsumers will be executed
// in a separate thread, and we want to prevent more reads
+ sendInProgress = true;
dispatchMessagesThread.execute(safeRun(() -> {
if (sendMessagesToConsumers(readType, entries)) {
readMoreEntries();
@@ -557,7 +558,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}));
} else {
if (sendMessagesToConsumers(readType, entries)) {
- readMoreEntiresAsync();
+ readMoreEntriesAsync();
}
}
}
@@ -923,7 +924,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (maxUnackedMessages <= 0 && blockedDispatcherOnUnackedMsgs == TRUE
&& BLOCKED_DISPATCHER_ON_UNACKMSG_UPDATER.compareAndSet(this, TRUE, FALSE)) {
log.info("[{}] Dispatcher is unblocked, since maxUnackedMessagesPerSubscription=0", name);
- readMoreEntiresAsync();
+ readMoreEntriesAsync();
}
int unAckedMessages = TOTAL_UNACKED_MESSAGES_UPDATER.addAndGet(this, numberOfMessages);
@@ -946,7 +947,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
// unblock dispatcher if it acks back enough messages
if (BLOCKED_DISPATCHER_ON_UNACKMSG_UPDATER.compareAndSet(this, TRUE, FALSE)) {
log.info("[{}] Dispatcher is unblocked", name);
- readMoreEntiresAsync();
+ readMoreEntriesAsync();
}
}
// increment broker-level count
