Perform periodic flush of ManagedCursor mark-delete posistions

conf/broker.conf

@@ -788,6 +788,10 @@ managedLedgerDefaultWriteQuorum=2
 # Number of guaranteed copies (acks to wait before write is complete)
 managedLedgerDefaultAckQuorum=2
 
+# How frequently to flush the cursor positions that were accumulated due to rate limiting. (seconds).
+# Default is 60 seconds
+managedLedgerCursorPositionFlushSeconds = 60
+
 # Default type of checksum to use when writing to BookKeeper. Default is "CRC32C"
 # Other possible options are "CRC32", "MAC" or "DUMMY" (no checksum).
 managedLedgerDigestType=CRC32C

conf/standalone.conf

@@ -536,6 +536,10 @@ managedLedgerDefaultWriteQuorum=1
 # Number of guaranteed copies (acks to wait before write is complete)
 managedLedgerDefaultAckQuorum=1
 
+# How frequently to flush the cursor positions that were accumulated due to rate limiting. (seconds).
+# Default is 60 seconds
+managedLedgerCursorPositionFlushSeconds = 60
+
 # Default type of checksum to use when writing to BookKeeper. Default is "CRC32C"
 # Other possible options are "CRC32", "MAC" or "DUMMY" (no checksum).
 managedLedgerDigestType=CRC32C

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java

@@ -66,6 +66,11 @@ public class ManagedLedgerFactoryConfig {
      */
     private int prometheusStatsLatencyRolloverSeconds = 60;
 
+    /**
+     * How frequently to flush the cursor positions that were accumulated due to rate limiting.
+     */
+    private int cursorPositionFlushSeconds = 60;
+
     /**
      * cluster name for prometheus stats
      */

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -169,6 +169,9 @@ public class ManagedCursorImpl implements ManagedCursor {
     private final ReadWriteLock lock = new ReentrantReadWriteLock();
 
     private RateLimiter markDeleteLimiter;
+    // The cursor is considered "dirty" when there are mark-delete updates that are only applied in memory,
+    // because of the rate limiting.
+    private volatile boolean isDirty = false;
 
     private boolean alwaysInactive = false;
 
@@ -1626,6 +1629,7 @@ public void asyncMarkDelete(final Position position, Map<String, Long> propertie
 
         // Apply rate limiting to mark-delete operations
         if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) {
+            isDirty = true;
             lastMarkDeleteEntry = new MarkDeleteEntry(newPosition, properties, null, null);
             callback.markDeleteComplete(ctx);
             return;
@@ -2883,6 +2887,24 @@ void updateReadStats(int readEntriesCount, long readEntriesSize) {
         this.entriesReadSize += readEntriesSize;
     }
 
+    void flush() {
+        if (!isDirty) {
+            return;
+        }
+
+        isDirty = false;
+        asyncMarkDelete(lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties, new MarkDeleteCallback() {
+            @Override
+            public void markDeleteComplete(Object ctx) {
+            }
+
+            @Override
+            public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
+                log.warn("[{}][{}] Failed to flush mark-delete position", ledger.getName(), name, exception);
+            }
+        }, null);
+    }
+
     private int applyMaxSizeCap(int maxEntries, long maxSizeBytes) {
         if (maxSizeBytes == NO_MAX_SIZE_LIMIT) {
             return maxEntries;

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java

@@ -108,6 +108,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
 
     private long lastStatTimestamp = System.nanoTime();
     private final ScheduledFuture<?> statsTask;
+    private final ScheduledFuture<?> flushCursorsTask;
 
     private final long cacheEvictionTimeThresholdNanos;
     private final MetadataStore metadataStore;
@@ -202,6 +203,8 @@ private ManagedLedgerFactoryImpl(BookkeeperFactoryForCustomEnsemblePlacementPoli
         this.mbean = new ManagedLedgerFactoryMBeanImpl(this);
         this.entryCacheManager = new EntryCacheManager(this);
         this.statsTask = scheduledExecutor.scheduleAtFixedRate(this::refreshStats, 0, StatsPeriodSeconds, TimeUnit.SECONDS);
+        this.flushCursorsTask = scheduledExecutor.scheduleAtFixedRate(this::flushCursors,
+                config.getCursorPositionFlushSeconds(), config.getCursorPositionFlushSeconds(), TimeUnit.SECONDS);
 
 
         this.cacheEvictionTimeThresholdNanos = TimeUnit.MILLISECONDS
@@ -230,6 +233,17 @@ public BookKeeper get(EnsemblePlacementPolicyConfig policy) {
         }
     }
 
+    private synchronized void flushCursors() {
+        ledgers.values().forEach(mlfuture -> {
+            if (mlfuture.isDone() && !mlfuture.isCompletedExceptionally()) {
+                ManagedLedgerImpl ml = mlfuture.getNow(null);
+                if (ml != null) {
+                    ml.getCursors().forEach(c -> ((ManagedCursorImpl) c).flush());
+                }
+            }
+        });
+    }
+
     private synchronized void refreshStats() {
         long now = System.nanoTime();
         long period = now - lastStatTimestamp;
@@ -483,6 +497,7 @@ void close(ManagedLedger ledger) {
     @Override
     public void shutdown() throws InterruptedException, ManagedLedgerException {
         statsTask.cancel(true);
+        flushCursorsTask.cancel(true);
 
         int numLedgers = ledgers.size();
         final CountDownLatch latch = new CountDownLatch(numLedgers);

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java

@@ -3190,8 +3190,9 @@ private List<IntRange> getAckedIndexRange(long[] bitSetLongArray, int batchSize)
         }
         return result;
     }
-
-    void testReadEntriesOrWaitWithMaxSize() throws Exception {
+
+    @Test
+    public void testReadEntriesOrWaitWithMaxSize() throws Exception {
         ManagedLedger ledger = factory.open("testReadEntriesOrWaitWithMaxSize");
         ManagedCursor c = ledger.openCursor("c");
 
@@ -3215,5 +3216,53 @@ void testReadEntriesOrWaitWithMaxSize() throws Exception {
         entries.forEach(e -> e.release());
     }
 
+    @Test
+    public void testFlushCursorAfterInactivity() throws Exception {
+        ManagedLedgerConfig config = new ManagedLedgerConfig();
+        config.setThrottleMarkDelete(1.0);
+
+        ManagedLedgerFactoryConfig factoryConfig = new ManagedLedgerFactoryConfig();
+        factoryConfig.setCursorPositionFlushSeconds(1);
+        ManagedLedgerFactory factory1 = new ManagedLedgerFactoryImpl(bkc, bkc.getZkHandle(), factoryConfig);
+        ManagedLedger ledger1 = factory1.open("testFlushCursorAfterInactivity", config);
+        ManagedCursor c1 = ledger1.openCursor("c");
+        List<Position> positions = new ArrayList<Position>();
+
+        for (int i = 0; i < 20; i++) {
+            positions.add(ledger1.addEntry(new byte[1024]));
+        }
+
+        CountDownLatch latch = new CountDownLatch(positions.size());
+
+        positions.forEach(p -> c1.asyncMarkDelete(p, new MarkDeleteCallback() {
+            @Override
+            public void markDeleteComplete(Object ctx) {
+                latch.countDown();
+            }
+
+            @Override
+            public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
+                throw new RuntimeException(exception);
+            }
+        }, null));
+
+        latch.await();
+
+        assertEquals(c1.getMarkDeletedPosition(), positions.get(positions.size() - 1));
+
+        // Give chance to the flush to be automatically triggered.
+        Thread.sleep(3000);
+
+        // Abruptly re-open the managed ledger without graceful close
+        ManagedLedgerFactory factory2 = new ManagedLedgerFactoryImpl(bkc, bkc.getZkHandle());
+        ManagedLedger ledger2 = factory2.open("testFlushCursorAfterInactivity", config);
+        ManagedCursor c2 = ledger2.openCursor("c");
+
+        assertEquals(c2.getMarkDeletedPosition(), positions.get(positions.size() - 1));
+
+        factory1.shutdown();
+        factory2.shutdown();
+    }
+
     private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class);
 }

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -1215,6 +1215,11 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private int managedLedgerDefaultAckQuorum = 2;
 
+    @FieldContext(minValue = 1,
+            category = CATEGORY_STORAGE_ML,
+            doc = "How frequently to flush the cursor positions that were accumulated due to rate limiting. (seconds). Default is 60 seconds")
+    private int managedLedgerCursorPositionFlushSeconds = 60;
+
     //
     //
     @FieldContext(

pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java

@@ -64,6 +64,7 @@ public ManagedLedgerClientFactory(ServiceConfiguration conf, ZooKeeper zkClient,
         managedLedgerFactoryConfig.setCopyEntriesInCache(conf.isManagedLedgerCacheCopyEntries());
         managedLedgerFactoryConfig.setPrometheusStatsLatencyRolloverSeconds(conf.getManagedLedgerPrometheusStatsLatencyRolloverSeconds());
         managedLedgerFactoryConfig.setTraceTaskExecution(conf.isManagedLedgerTraceTaskExecution());
+        managedLedgerFactoryConfig.setCursorPositionFlushSeconds(conf.getManagedLedgerCursorPositionFlushSeconds());
 
         Configuration configuration = new ClientConfiguration();
         if (conf.isBookkeeperClientExposeStatsToPrometheus()) {