access individualDeletedMessages with thread-safe while filtering out…

… replay-message (#45)
apache · Sep 29, 2016 · 26671f6 · 26671f6
1 parent 0577fda
commit 26671f6
Showing 1 changed file with 9 additions and 4 deletions.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -812,10 +812,15 @@ public Set<? extends Position> asyncReplayEntries(final Set<? extends Position>
 
         // filters out messages which are already acknowledged
         Set<Position> alreadyAcknowledgedPositions = Sets.newHashSet();
-        positions.stream().filter(position -> {
-            return individualDeletedMessages.contains((PositionImpl) position)
-                    || ((PositionImpl) position).compareTo(markDeletePosition) < 0;
-        }).forEach(pos -> alreadyAcknowledgedPositions.add(pos));
+        lock.readLock().lock();
+        try {
+            positions.stream().filter(position -> {
+                return individualDeletedMessages.contains((PositionImpl) position)
+                        || ((PositionImpl) position).compareTo(markDeletePosition) < 0;
+            }).forEach(pos -> alreadyAcknowledgedPositions.add(pos));
+        } finally {
+            lock.readLock().unlock();
+        }
 
         final int totalValidPositions = positions.size() - alreadyAcknowledgedPositions.size();
         final AtomicReference<ManagedLedgerException> exception = new AtomicReference<>();