Skip to content
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

Jump to bottom

[pulsar-broker] Make some member variables of Dispatcher volatile #5193

Merged
merged 1 commit into from
Sep 17, 2019
Merged

[pulsar-broker] Make some member variables of Dispatcher volatile #5193

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
2 changes: 1 addition & 1 deletion ...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,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;
Expand Down
8 changes: 4 additions & 4 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 @@ -82,13 +82,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");
Expand Down
4 changes: 2 additions & 2 deletions ...org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down