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 index 9498145de2823..05f6f5c762133 100644 --- 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 @@ -1116,7 +1116,14 @@ public synchronized void readEntryFailed(ManagedLedgerException mle, Object ctx) }; positions.stream().filter(position -> !alreadyAcknowledgedPositions.contains(position)) - .forEach(p -> ledger.asyncReadEntry((PositionImpl) p, cb, ctx)); + .forEach(p ->{ + if (((PositionImpl) p).compareTo(this.readPosition) == 0) { + this.setReadPosition(this.readPosition.getNext()); + log.warn("[{}][{}] replayPosition{} equals readPosition{}," + " need set next readPositio", + ledger.getName(), name, (PositionImpl) p, this.readPosition); + } + ledger.asyncReadEntry((PositionImpl) p, cb, ctx); + }); return alreadyAcknowledgedPositions; }