From 5fde6f32b038327163ebf7babc556b72f337a1a1 Mon Sep 17 00:00:00 2001
From: Rajan Dhabalia <rdhabalia@apache.org>
Date: Mon, 19 Nov 2018 23:04:48 -0800
Subject: [PATCH] Prevent race condition while calculating unack-msg offset
 stats (#3015)

* Prevent race condition while calculating unack-msg offset stats

* local var
---
 .../apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

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 6c7e15c0c6c07..4085f80354c98 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
@@ -665,7 +665,13 @@ public long getNumberOfEntries() {
 
     @Override
     public long getNumberOfEntriesSinceFirstNotAckedMessage() {
-        return ledger.getNumberOfEntries(Range.openClosed(markDeletePosition, readPosition));
+        // sometimes for already caught up consumer: due to race condition markDeletePosition > readPosition. so,
+        // validate it before preparing range
+        PositionImpl markDeletePosition = this.markDeletePosition;
+        PositionImpl readPosition = this.readPosition;
+        return (markDeletePosition.compareTo(readPosition) < 0)
+                ? ledger.getNumberOfEntries(Range.openClosed(markDeletePosition, readPosition))
+                : 0;
     }
 
     @Override