Merge branch '3.1_ds' into 3.1_fx_merged

datastax · May 2, 2024 · 10ffade · 10ffade
2 parents 9a11c26 + 5bf373d
commit 10ffade
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 1 deletion.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -2855,6 +2855,15 @@ public void operationFailed(MetaStoreException e) {
         }
     }
 
+    @Override
+    public void rolloverCursorsInBackground() {
+        if (cursors.hasDurableCursors()) {
+            executor.execute(() -> {
+                cursors.forEach(ManagedCursor::periodicRollover);
+            });
+        }
+    }
+
     /**
      * @param ledgerId the ledger handle which maybe will be released.
      * @return if the ledger handle was released.

diff --git a/.../src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/.../src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -522,7 +522,8 @@ private void readMorePoliciesAsync(SystemTopicClient.Reader<PulsarEvent> reader)
                     } else {
                         Throwable cause = FutureUtil.unwrapCompletionException(ex);
                         if (cause instanceof PulsarClientException.AlreadyClosedException) {
-                            log.info("Closing the topic policies reader for {}", reader.getSystemTopic().getTopicName());
+                            log.info("Closing the topic policies reader for {}",
+                                    reader.getSystemTopic().getTopicName());
                             cleanCacheAndCloseReader(
                                     reader.getSystemTopic().getTopicName().getNamespaceObject(), false);
                         } else {