[pulsar] branch master updated: Fix NPE caused by the race condition (#2973)

[nk...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

nkurihar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 9ecfb7d  Fix NPE caused by the race condition (#2973)
9ecfb7d is described below

commit 9ecfb7da4a2b5bbf9e3d1dbd10753149f3c0826f
Author: massakam <ma...@yahoo-corp.jp>
AuthorDate: Wed Nov 14 00:36:36 2018 +0900

    Fix NPE caused by the race condition (#2973)
---
 .../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 22 ++++++++++++----------
 1 file changed, 12 insertions(+), 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
index d33f151..6083c64 100644
--- 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
@@ -2609,16 +2609,18 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
     }
 
     public void deactivateCursor(ManagedCursor cursor) {
-        if (activeCursors.get(cursor.getName()) != null) {
-            activeCursors.removeCursor(cursor.getName());
-            if (activeCursors.isEmpty()) {
-                // cleanup cache if there is no active subscription
-                entryCache.clear();
-            } else {
-                // if removed subscription was the slowest subscription : update cursor and let it clear cache: till
-                // new slowest-cursor's read-position
-                discardEntriesFromCache((ManagedCursorImpl) activeCursors.getSlowestReader(),
-                        getPreviousPosition((PositionImpl) activeCursors.getSlowestReader().getReadPosition()));
+        synchronized (activeCursors) {
+            if (activeCursors.get(cursor.getName()) != null) {
+                activeCursors.removeCursor(cursor.getName());
+                if (activeCursors.isEmpty()) {
+                    // cleanup cache if there is no active subscription
+                    entryCache.clear();
+                } else {
+                    // if removed subscription was the slowest subscription : update cursor and let it clear cache:
+                    // till new slowest-cursor's read-position
+                    discardEntriesFromCache((ManagedCursorImpl) activeCursors.getSlowestReader(),
+                            getPreviousPosition((PositionImpl) activeCursors.getSlowestReader().getReadPosition()));
+                }
             }
         }
     }