[cleanup][ml] ManagedCursor clean up. (apache#22246)

streamnative · Mar 12, 2024 · 532b0d9 · 532b0d9
1 parent 9f63e24
commit 532b0d9
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 9 deletions.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
@@ -42,6 +42,7 @@ protected EntryImpl newObject(Handle<EntryImpl> handle) {
     private long timestamp;
     private long ledgerId;
     private long entryId;
+    private PositionImpl position;
     ByteBuf data;
 
     private Runnable onDeallocate;
@@ -151,7 +152,10 @@ public int getLength() {
 
     @Override
     public PositionImpl getPosition() {
-        return new PositionImpl(ledgerId, entryId);
+        if (position == null) {
+            position = PositionImpl.get(ledgerId, entryId);
+        }
+        return position;
     }
 
     @Override
@@ -197,6 +201,7 @@ protected void deallocate() {
         timestamp = -1;
         ledgerId = -1;
         entryId = -1;
+        position = null;
         recyclerHandle.recycle(this);
     }
 

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
@@ -1502,10 +1502,7 @@ public Set<? extends Position> asyncReplayEntries(Set<? extends Position> positi
         Set<Position> alreadyAcknowledgedPositions = new HashSet<>();
         lock.readLock().lock();
         try {
-            positions.stream()
-                    .filter(position -> ((PositionImpl) position).compareTo(markDeletePosition) <= 0
-                            || individualDeletedMessages.contains(position.getLedgerId(), position.getEntryId()))
-                    .forEach(alreadyAcknowledgedPositions::add);
+            positions.stream().filter(this::isMessageDeleted).forEach(alreadyAcknowledgedPositions::add);
         } finally {
             lock.readLock().unlock();
         }
@@ -2278,8 +2275,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
                     return;
                 }
 
-                if (position.compareTo(markDeletePosition) <= 0
-                        || individualDeletedMessages.contains(position.getLedgerId(), position.getEntryId())) {
+                if (isMessageDeleted(position)) {
                     if (config.isDeletionAtBatchIndexLevelEnabled()) {
                         BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
                         if (bitSetRecyclable != null) {
@@ -3504,8 +3500,7 @@ public Range<PositionImpl> getLastIndividualDeletedRange() {
     @Override
     public void trimDeletedEntries(List<Entry> entries) {
         entries.removeIf(entry -> {
-            boolean isDeleted = markDeletePosition.compareTo(entry.getLedgerId(), entry.getEntryId()) >= 0
-                    || individualDeletedMessages.contains(entry.getLedgerId(), entry.getEntryId());
+            boolean isDeleted = isMessageDeleted(entry.getPosition());
             if (isDeleted) {
                 entry.release();
             }