Make some member variables of Dispatcher volatile (apache#5193)

(cherry picked from commit b9e7bf3)
sarvex · Nov 19, 2019 · 90ec8e1 · 90ec8e1
1 parent 850b123
commit 90ec8e1
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 7 deletions.
diff --git a/...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java b/...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java
@@ -39,7 +39,7 @@ public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDi
 
     protected final CopyOnWriteArrayList<Consumer> consumerList = new CopyOnWriteArrayList<>();
     protected final ObjectSet<Consumer> consumerSet = new ObjectHashSet<>();
-    protected int currentConsumerRoundRobinIndex = 0;
+    protected volatile int currentConsumerRoundRobinIndex = 0;
 
     protected static final int FALSE = 0;
     protected static final int TRUE = 1;

diff --git a/...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -80,13 +80,13 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
     private Optional<DelayedDeliveryTracker> delayedDeliveryTracker = Optional.empty();
     private final boolean isDelayedDeliveryEnabled;
 
-    private boolean havePendingRead = false;
-    private boolean havePendingReplayRead = false;
+    private volatile boolean havePendingRead = false;
+    private volatile boolean havePendingReplayRead = false;
     private boolean shouldRewindBeforeReadingOrReplaying = false;
     protected final String name;
 
-    protected int totalAvailablePermits = 0;
-    private int readBatchSize;
+    protected volatile int totalAvailablePermits = 0;
+    private volatile int readBatchSize;
     private final Backoff readFailureBackoff = new Backoff(15, TimeUnit.SECONDS, 1, TimeUnit.MINUTES, 0, TimeUnit.MILLISECONDS);
     private static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers> TOTAL_UNACKED_MESSAGES_UPDATER =
             AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class, "totalUnackedMessages");

diff --git a/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -60,11 +60,11 @@ public final class PersistentDispatcherSingleActiveConsumer extends AbstractDisp
     private final PersistentTopic topic;
     private final ManagedCursor cursor;
     private final String name;
-    private Optional<DispatchRateLimiter> dispatchRateLimiter = Optional.empty();;
+    private Optional<DispatchRateLimiter> dispatchRateLimiter = Optional.empty();
 
     private volatile boolean havePendingRead = false;
 
-    private int readBatchSize;
+    private volatile int readBatchSize;
     private final Backoff readFailureBackoff = new Backoff(15, TimeUnit.SECONDS, 1, TimeUnit.MINUTES, 0, TimeUnit.MILLISECONDS);
     private final ServiceConfiguration serviceConfig;
     private volatile ScheduledFuture<?> readOnActiveConsumerTask = null;