[pulsar-broker] minor fix for removing atomic-updater at managed-ledg…

…er (apache#4146)
JunpingDu · Apr 26, 2019 · 07de52d · 07de52d
1 parent fbdf038
commit 07de52d
Showing 1 changed file with 8 additions and 10 deletions.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -236,8 +236,6 @@ enum PositionBound {
             .newUpdater(ManagedLedgerImpl.class, "readOpCount");
     private volatile long readOpCount = 0;
     // last read-operation's callback to check read-timeout on it. 
-    private static final AtomicReferenceFieldUpdater<ManagedLedgerImpl, ReadEntryCallbackWrapper> LAST_READ_CALLBACK = AtomicReferenceFieldUpdater
-            .newUpdater(ManagedLedgerImpl.class, ReadEntryCallbackWrapper.class, "lastReadCallback");
     private volatile ReadEntryCallbackWrapper lastReadCallback = null;
 
     /**
@@ -1592,7 +1590,7 @@ protected void asyncReadEntry(ReadHandle ledger, PositionImpl position, ReadEntr
             long createdTime = System.nanoTime();
             ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, position.getLedgerId(),
                     position.getEntryId(), callback, readOpCount, createdTime, ctx);
-            LAST_READ_CALLBACK.set(this, readCallback);
+            lastReadCallback = readCallback;
             entryCache.asyncReadEntry(ledger, position, readCallback, readOpCount);
         } else {
             entryCache.asyncReadEntry(ledger, position, callback, ctx);
@@ -1607,7 +1605,7 @@ protected void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry
             long createdTime = System.nanoTime();
             ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, ledger.getId(), firstEntry,
                     opReadEntry, readOpCount, createdTime, ctx);
-            LAST_READ_CALLBACK.set(this, readCallback);
+            lastReadCallback = readCallback;
             entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, isSlowestReader, readCallback, readOpCount);
         } else {
             entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, isSlowestReader, opReadEntry, ctx);
@@ -3128,12 +3126,12 @@ private void checkReadTimeout() {
         if (timeoutSec < 1) {
             return;
         }
-        ReadEntryCallbackWrapper callback = LAST_READ_CALLBACK.get(this);
-        if (callback != null && callback.isTimedOut(timeoutSec)) {
-            log.warn("[{}]-{} read entry timeout for {} after {} sec", this.name, callback.ledgerId, callback.entryId,
-                    timeoutSec);
-            callback.readFailed(createManagedLedgerException(BKException.Code.TimeoutException), callback.readOpCount);
-            LAST_READ_CALLBACK.set(this, null);
+        if (this.lastReadCallback != null && this.lastReadCallback.isTimedOut(timeoutSec)) {
+            log.warn("[{}]-{} read entry timeout for {} after {} sec", this.name, this.lastReadCallback.ledgerId,
+                    this.lastReadCallback.entryId, timeoutSec);
+            this.lastReadCallback.readFailed(createManagedLedgerException(BKException.Code.TimeoutException),
+                    this.lastReadCallback.readOpCount);
+            lastReadCallback = null;
         }
     }