Fix non-atomic volatile update

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -53,6 +53,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -106,7 +107,13 @@ public class ManagedCursorImpl implements ManagedCursor {
     private final BookKeeper.DigestType digestType;
 
     protected volatile PositionImpl markDeletePosition;
+
+    protected static final AtomicReferenceFieldUpdater<ManagedCursorImpl, PositionImpl> READ_POSITION_UPDATER =
+            AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, PositionImpl.class, "readPosition");
     protected volatile PositionImpl readPosition;
+
+    protected static final AtomicReferenceFieldUpdater<ManagedCursorImpl, MarkDeleteEntry> LAST_MARK_DELETE_ENTRY_UPDATER =
+            AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, MarkDeleteEntry.class, "lastMarkDeleteEntry");
     protected volatile MarkDeleteEntry lastMarkDeleteEntry;
 
     protected static final AtomicReferenceFieldUpdater<ManagedCursorImpl, OpReadEntry> WAITING_READ_OP_UPDATER =
@@ -125,6 +132,8 @@ public class ManagedCursorImpl implements ManagedCursor {
     @SuppressWarnings("unused")
     private volatile int pendingReadOps = 0;
 
+    private static final AtomicLongFieldUpdater<ManagedCursorImpl> MSG_CONSUMED_COUNTER_UPDATER =
+            AtomicLongFieldUpdater.newUpdater(ManagedCursorImpl.class, "messagesConsumedCounter");
     // This counters are used to compute the numberOfEntries and numberOfEntriesInBacklog values, without having to look
     // at the list of ledgers in the ml. They are initialized to (-backlog) at opening, and will be incremented each
     // time a message is read or deleted.
@@ -884,11 +893,11 @@ public void operationComplete() {
                     PositionImpl newMarkDeletePosition = ledger.getPreviousPosition(newPosition);
 
                     if (markDeletePosition.compareTo(newMarkDeletePosition) >= 0) {
-                        messagesConsumedCounter -= getNumberOfEntries(
-                                Range.closedOpen(newMarkDeletePosition, markDeletePosition));
+                        MSG_CONSUMED_COUNTER_UPDATER.addAndGet(cursorImpl(), -getNumberOfEntries(
+                                Range.closedOpen(newMarkDeletePosition, markDeletePosition)));
                     } else {
-                        messagesConsumedCounter += getNumberOfEntries(
-                                Range.closedOpen(markDeletePosition, newMarkDeletePosition));
+                        MSG_CONSUMED_COUNTER_UPDATER.addAndGet(cursorImpl(), getNumberOfEntries(
+                                Range.closedOpen(markDeletePosition, newMarkDeletePosition)));
                     }
                     markDeletePosition = newMarkDeletePosition;
                     lastMarkDeleteEntry = new MarkDeleteEntry(newMarkDeletePosition, Collections.emptyMap(),
@@ -1428,7 +1437,7 @@ PositionImpl setAcknowledgedPosition(PositionImpl newMarkDeletePosition) {
                 log.debug("[{}] Moved ack position from: {} to: {} -- skipped: {}", ledger.getName(),
                         oldMarkDeletePosition, newMarkDeletePosition, skippedEntries);
             }
-            messagesConsumedCounter += skippedEntries;
+            MSG_CONSUMED_COUNTER_UPDATER.addAndGet(this, skippedEntries);
         }
 
         // markDelete-position and clear out deletedMsgSet
@@ -1720,7 +1729,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
                 PositionImpl previousPosition = ledger.getPreviousPosition(position);
                 individualDeletedMessages.addOpenClosed(previousPosition.getLedgerId(), previousPosition.getEntryId(),
                         position.getLedgerId(), position.getEntryId());
-                ++messagesConsumedCounter;
+                MSG_CONSUMED_COUNTER_UPDATER.incrementAndGet(this);
 
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] [{}] Individually deleted messages: {}", ledger.getName(), name,
@@ -1766,8 +1775,9 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
 
         // Apply rate limiting to mark-delete operations
         if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) {
-            lastMarkDeleteEntry = new MarkDeleteEntry(newMarkDeletePosition, lastMarkDeleteEntry.properties, null,
-                    null);
+            PositionImpl finalNewMarkDeletePosition = newMarkDeletePosition;
+            LAST_MARK_DELETE_ENTRY_UPDATER.updateAndGet(this,
+                    last -> new MarkDeleteEntry(finalNewMarkDeletePosition, last.properties, null, null));
             callback.deleteComplete(ctx);
             return;
         }
@@ -2666,5 +2676,9 @@ public void trimDeletedEntries(List<Entry> entries) {
                 || individualDeletedMessages.contains(entry.getLedgerId(), entry.getEntryId()));
     }
 
+    private ManagedCursorImpl cursorImpl() {
+        return this;
+    }
+
     private static final Logger log = LoggerFactory.getLogger(ManagedCursorImpl.class);
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyCursorImpl.java

@@ -54,7 +54,8 @@ public ReadOnlyCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, Man
     @Override
     public void skipEntries(int numEntriesToSkip) {
         log.info("[{}] Skipping {} entries on read-only cursor {}", ledger.getName(), numEntriesToSkip);
-        readPosition = ledger.getPositionAfterN(readPosition, numEntriesToSkip, PositionBound.startIncluded).getNext();
+        READ_POSITION_UPDATER.getAndUpdate(this, lastRead ->
+                ledger.getPositionAfterN(lastRead, numEntriesToSkip, PositionBound.startIncluded).getNext());
     }
 
     @Override

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -41,6 +41,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.stream.Collectors;
 
 import javax.naming.AuthenticationException;
@@ -161,6 +162,8 @@ public class ServerCnx extends PulsarHandler {
     private FeatureFlags features;
     // Flag to manage throttling-publish-buffer by atomically enable/disable read-channel.
     private volatile boolean autoReadDisabledPublishBufferLimiting = false;
+    private static final AtomicLongFieldUpdater<ServerCnx> MSG_PUBLISH_BUFFER_SIZE_UPDATER =
+            AtomicLongFieldUpdater.newUpdater(ServerCnx.class, "messagePublishBufferSize");
     private volatile long messagePublishBufferSize = 0;
 
     enum State {
@@ -1759,7 +1762,7 @@ public boolean isWritable() {
     }
 
     public void startSendOperation(Producer producer, int msgSize) {
-        messagePublishBufferSize += msgSize;
+        MSG_PUBLISH_BUFFER_SIZE_UPDATER.getAndAdd(this, msgSize);
         boolean isPublishRateExceeded = producer.getTopic().isPublishRateExceeded();
         if (++pendingSendRequest == maxPendingSendRequests || isPublishRateExceeded) {
             // When the quota of pending send requests is reached, stop reading from socket to cause backpressure on
@@ -1775,7 +1778,7 @@ public void startSendOperation(Producer producer, int msgSize) {
     }
 
     public void completedSendOperation(boolean isNonPersistentTopic, int msgSize) {
-        messagePublishBufferSize -= msgSize;
+        MSG_PUBLISH_BUFFER_SIZE_UPDATER.getAndAdd(this, -msgSize);
         if (--pendingSendRequest == resumeReadsThreshold) {
             // Resume reading from socket
             ctx.channel().config().setAutoRead(true);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -88,6 +88,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
     private boolean shouldRewindBeforeReadingOrReplaying = false;
     protected final String name;
 
+    protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers> TOTAL_AVAILABLE_PERMITS_UPDATER =
+            AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class, "totalAvailablePermits");
     protected volatile int totalAvailablePermits = 0;
     private volatile int readBatchSize;
     private final Backoff readFailureBackoff = new Backoff(15, TimeUnit.SECONDS, 1, TimeUnit.MINUTES, 0, TimeUnit.MILLISECONDS);
@@ -500,10 +502,10 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
                 c.sendMessages(entriesForThisConsumer, batchSizes, sendMessageInfo.getTotalMessages(),
                         sendMessageInfo.getTotalBytes(), redeliveryTracker);
 
-                long msgSent = sendMessageInfo.getTotalMessages();
+                int msgSent = sendMessageInfo.getTotalMessages();
                 start += messagesForC;
                 entriesToDispatch -= messagesForC;
-                totalAvailablePermits -= msgSent;
+                TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -msgSent);
                 totalMessagesSent += sendMessageInfo.getTotalMessages();
                 totalBytesSent += sendMessageInfo.getTotalBytes();
             }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

@@ -116,7 +116,7 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
                 });
                 entriesWithSameKey.getValue().removeAll(subList);
 
-                totalAvailablePermits -= sendMessageInfo.getTotalMessages();
+                TOTAL_AVAILABLE_PERMITS_UPDATER.getAndAdd(this, -sendMessageInfo.getTotalMessages());
                 totalMessagesSent += sendMessageInfo.getTotalMessages();
                 totalBytesSent += sendMessageInfo.getTotalBytes();
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java

@@ -88,7 +88,7 @@ public boolean add(MessageImpl<?> msg, SendCallback callback) {
             messageMetadata.setSequenceId(lowestSequenceId);
         }
         highestSequenceId = msg.getSequenceId();
-        producer.lastSequenceIdPushed = Math.max(producer.lastSequenceIdPushed, msg.getSequenceId());
+        ProducerImpl.LAST_SEQ_ID_PUSHED_UPDATER.getAndUpdate(producer, prev -> Math.max(prev, msg.getSequenceId()));
 
         return isBatchFull();
     }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -111,7 +111,12 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
 
     private final CompressionCodec compressor;
 
+    static final AtomicLongFieldUpdater<ProducerImpl> LAST_SEQ_ID_PUBLISHED_UPDATER = AtomicLongFieldUpdater
+            .newUpdater(ProducerImpl.class, "lastSequenceIdPublished");
     private volatile long lastSequenceIdPublished;
+
+    static final AtomicLongFieldUpdater<ProducerImpl> LAST_SEQ_ID_PUSHED_UPDATER = AtomicLongFieldUpdater
+            .newUpdater(ProducerImpl.class, "lastSequenceIdPushed");
     protected volatile long lastSequenceIdPushed;
     private volatile boolean isLastSequenceIdPotentialDuplicated;
 
@@ -847,7 +852,9 @@ void ackReceived(ClientCnx cnx, long sequenceId, long highestSequenceId, long le
         if (callback) {
             op = pendingCallbacks.poll();
             if (op != null) {
-                lastSequenceIdPublished = Math.max(lastSequenceIdPublished, getHighestSequenceId(op));
+                OpSendMsg finalOp = op;
+                LAST_SEQ_ID_PUBLISHED_UPDATER.getAndUpdate(this,
+                        last -> Math.max(last, getHighestSequenceId(finalOp)));
                 op.setMessageId(ledgerId, entryId, partitionIndex);
                 try {
                     // Need to protect ourselves from any exception being thrown in the future handler from the
@@ -1481,7 +1488,8 @@ private void processOpSendMsg(OpSendMsg op) {
             }
             pendingMessages.put(op);
             if (op.msg != null) {
-                lastSequenceIdPushed = Math.max(lastSequenceIdPushed, getHighestSequenceId(op));
+                LAST_SEQ_ID_PUSHED_UPDATER.getAndUpdate(this,
+                        last -> Math.max(last, getHighestSequenceId(op)));
             }
             ClientCnx cnx = cnx();
             if (isConnected()) {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarServiceNameResolver.java

@@ -26,6 +26,7 @@
 import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.PulsarClientException.InvalidServiceURL;
 import org.apache.pulsar.common.net.ServiceURI;
@@ -38,6 +39,8 @@ public class PulsarServiceNameResolver implements ServiceNameResolver {
 
     private volatile ServiceURI serviceUri;
     private volatile String serviceUrl;
+    private static final AtomicIntegerFieldUpdater<PulsarServiceNameResolver> CURRENT_INDEX_UPDATER =
+            AtomicIntegerFieldUpdater.newUpdater(PulsarServiceNameResolver.class, "currentIndex");
     private volatile int currentIndex;
     private volatile List<InetSocketAddress> addressList;
 
@@ -51,7 +54,7 @@ public InetSocketAddress resolveHost() {
         if (list.size() == 1) {
             return list.get(0);
         } else {
-            currentIndex = (currentIndex + 1) % list.size();
+            CURRENT_INDEX_UPDATER.getAndUpdate(this, last -> (last + 1) % list.size());
             return list.get(currentIndex);
 
         }

pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java

@@ -24,6 +24,7 @@
 import com.google.common.collect.Lists;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.locks.StampedLock;
 import java.util.function.LongFunction;
 
@@ -195,6 +196,9 @@ private static final class Section<V> extends StampedLock {
         private volatile V[] values;
 
         private volatile int capacity;
+        private static final AtomicIntegerFieldUpdater<Section> SIZE_UPDATER =
+                AtomicIntegerFieldUpdater.newUpdater(Section.class, "size");
+
         private volatile int size;
         private int usedBuckets;
         private int resizeThreshold;
@@ -282,12 +286,12 @@ V put(long key, V value, int keyHash, boolean onlyIfAbsent, LongFunction<V> valu
                     if (storedKey == key) {
                         if (storedValue == EmptyValue) {
                             values[bucket] = value != null ? value : valueProvider.apply(key);
-                            ++size;
+                            SIZE_UPDATER.incrementAndGet(this);
                             ++usedBuckets;
                             return valueProvider != null ? values[bucket] : null;
                         } else if (storedValue == DeletedValue) {
                             values[bucket] = value != null ? value : valueProvider.apply(key);
-                            ++size;
+                            SIZE_UPDATER.incrementAndGet(this);
                             return valueProvider != null ? values[bucket] : null;
                         } else if (!onlyIfAbsent) {
                             // Over written an old value for same key
@@ -307,7 +311,7 @@ V put(long key, V value, int keyHash, boolean onlyIfAbsent, LongFunction<V> valu
 
                         keys[bucket] = key;
                         values[bucket] = value != null ? value : valueProvider.apply(key);
-                        ++size;
+                        SIZE_UPDATER.incrementAndGet(this);
                         return valueProvider != null ? values[bucket] : null;
                     } else if (storedValue == DeletedValue) {
                         // The bucket contained a different deleted key
@@ -348,7 +352,7 @@ private V remove(long key, Object value, int keyHash) {
                                 return null;
                             }
 
-                            --size;
+                            SIZE_UPDATER.decrementAndGet(this);
                             V nextValueInArray = values[signSafeMod(bucket + 1, capacity)];
                             if (nextValueInArray == EmptyValue) {
                                 values[bucket] = (V) EmptyValue;

pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java

@@ -25,6 +25,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.locks.StampedLock;
 
 /**
@@ -214,6 +215,8 @@ private static final class Section extends StampedLock {
         private volatile long[] table;
 
         private volatile int capacity;
+        private static final AtomicIntegerFieldUpdater<Section> SIZE_UPDATER = AtomicIntegerFieldUpdater
+                .newUpdater(Section.class, "size");
         private volatile int size;
         private int usedBuckets;
         private int resizeThreshold;
@@ -300,7 +303,7 @@ boolean add(long item1, long item2, long hash) {
 
                         table[bucket] = item1;
                         table[bucket + 1] = item2;
-                        ++size;
+                        SIZE_UPDATER.incrementAndGet(this);
                         return true;
                     } else if (storedItem1 == DeletedItem) {
                         // The bucket contained a different deleted key
@@ -333,7 +336,7 @@ private boolean remove(long item1, long item2, int hash) {
                     long storedItem1 = table[bucket];
                     long storedItem2 = table[bucket + 1];
                     if (item1 == storedItem1 && item2 == storedItem2) {
-                        --size;
+                        SIZE_UPDATER.decrementAndGet(this);
 
                         cleanBucket(bucket);
                         return true;

pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java

@@ -24,6 +24,7 @@
 import com.google.common.collect.Lists;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.locks.StampedLock;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
@@ -182,6 +183,8 @@ private static final class Section<K, V> extends StampedLock {
         private volatile Object[] table;
 
         private volatile int capacity;
+        private static final AtomicIntegerFieldUpdater<Section> SIZE_UPDATER =
+                AtomicIntegerFieldUpdater.newUpdater(Section.class, "size");
         private volatile int size;
         private int usedBuckets;
         private int resizeThreshold;
@@ -276,7 +279,7 @@ V put(K key, V value, int keyHash, boolean onlyIfAbsent, Function<K, V> valuePro
 
                         table[bucket] = key;
                         table[bucket + 1] = value;
-                        ++size;
+                        SIZE_UPDATER.incrementAndGet(this);
                         return valueProvider != null ? value : null;
                     } else if (storedKey == DeletedKey) {
                         // The bucket contained a different deleted key
@@ -310,7 +313,7 @@ private V remove(K key, Object value, int keyHash) {
                     V storedValue = (V) table[bucket + 1];
                     if (key.equals(storedKey)) {
                         if (value == null || value.equals(storedValue)) {
-                            --size;
+                            SIZE_UPDATER.decrementAndGet(this);
 
                             int nextInArray = (bucket + 2) & (table.length - 1);
                             if (table[nextInArray] == EmptyKey) {