You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "BewareMyPower (via GitHub)" <gi...@apache.org> on 2025/07/15 03:40:21 UTC

[PR] [fix][broker] Fix deduplication replay might never complete for exceptions [pulsar]

BewareMyPower opened a new pull request, #24511:
URL: https://github.com/apache/pulsar/pull/24511

   Main Issue: #24497
   
   ### Motivation
   
   If any exception is thrown by managed cursor's API or the callback, `MessageDeduplication#checkStatus` will never have a chance to complete, which will make the `PersistentTopic` never available unless unloading the namespace or topic.
   
   ### Modifications
   
   - Add `ManagedLedgerUtils`, which includes a set of future-based managed ledger APIs, which don't need explicit exception catching for callback-based APIs.
   - Add a `ManagedLedgerReplayTask` to perform replay on a managed ledger from a given position. It can handle any unexpected exception from the managed ledger APIs or callbacks.
   - Leverage `ManagedLedgerReplayTask` in `MessageDeduplication` for recovery and `ManagedLedgerUtils` for safe snapshot taking.
   
   ### Verifying this change
   
   - Add `ManagedLedgerReplayTaskTest` to verify `ManagedLedgerReplayTask` works in many cases.
   - Add `checkStatusFail` to verify `MessageDeduplication#checkStatus` will complete exceptionally for unexpected exception.
   
   ### Documentation
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   - [ ] `doc` <!-- Your PR contains doc changes. -->
   - [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
   - [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
   - [ ] `doc-complete` <!-- Docs have been already added -->
   
   ### Matching PR in forked repository
   
   PR in forked repository: https://github.com/BewareMyPower/pulsar/pull/49


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][broker] Fix deduplication replay might never complete for exceptions [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #24511:
URL: https://github.com/apache/pulsar/pull/24511#discussion_r2228927931


##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerUtils.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.util;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.common.classification.InterfaceStability;
+
+/**
+ * This util class contains some future-based methods to replace callback-based APIs. With a callback-based API, if any
+ * exception is thrown in the callback, the callback will never have a chance to be called. While with a future-based
+ * API, if any exception is thrown in future's callback (e.g. `thenApply`), the future will eventually be completed
+ * exceptionally. In addition, future-based API is easier for users to switch a different executor to execute the
+ * callback (e.g. `thenApplyAsync`).
+ */
+@InterfaceStability.Evolving
+public class ManagedLedgerUtils {
+
+    public static CompletableFuture<ManagedCursor> openCursor(ManagedLedger ml, String cursorName) {
+        final var future = new CompletableFuture<ManagedCursor>();
+        ml.asyncOpenCursor(cursorName, new AsyncCallbacks.OpenCursorCallback() {
+            @Override
+            public void openCursorComplete(ManagedCursor cursor, Object ctx) {
+                future.complete(cursor);
+            }
+
+            @Override
+            public void openCursorFailed(ManagedLedgerException exception, Object ctx) {
+                future.completeExceptionally(exception);
+            }
+        }, null);
+        return future;
+    }
+
+    public static CompletableFuture<List<Entry>> readEntries(ManagedCursor cursor, int numberOfEntriesToRead,
+                                                             Position maxPosition) {
+        final var future = new CompletableFuture<List<Entry>>();
+        cursor.asyncReadEntries(numberOfEntriesToRead, new AsyncCallbacks.ReadEntriesCallback() {
+            @Override
+            public void readEntriesComplete(List<Entry> entries, Object ctx) {
+                future.complete(entries);
+            }
+
+            @Override
+            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+                future.completeExceptionally(exception);
+            }
+        }, null, maxPosition);
+        return future;
+    }
+
+    public static CompletableFuture<Void> markDelete(ManagedCursor cursor, Position position,
+                                                     Map<String, Long> properties) {
+        final var future = new CompletableFuture<Void>();
+        cursor.asyncMarkDelete(position, properties, new AsyncCallbacks.MarkDeleteCallback() {
+            @Override
+            public void markDeleteComplete(Object ctx) {
+                future.complete(null);
+            }
+
+            @Override
+            public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
+                future.completeExceptionally(exception);
+            }
+        }, new Object());

Review Comment:
   maybe pass null is OK



##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java:
##########
@@ -84,5 +109,24 @@ public void markerMessageNotRecordPersistent() {
         }
     }
 
-
+    @Test
+    public void checkStatusFail() throws Exception {

Review Comment:
   LGTM



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][broker] Fix deduplication replay might never complete for exceptions [pulsar]

Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #24511:
URL: https://github.com/apache/pulsar/pull/24511


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][broker] Fix deduplication replay might never complete for exceptions [pulsar]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24511:
URL: https://github.com/apache/pulsar/pull/24511#issuecomment-3116230809

   ## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24511?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   Attention: Patch coverage is `92.78351%` with `7 lines` in your changes missing coverage. Please review.
   > Project coverage is 74.32%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`68b21b0`)](https://app.codecov.io/gh/apache/pulsar/commit/68b21b0775e3a553363b208da3dae13c7f6bfa66?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   > Report is 1223 commits behind head on master.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24511?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [...he/bookkeeper/mledger/util/ManagedLedgerUtils.java](https://app.codecov.io/gh/apache/pulsar/pull/24511?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Futil%2FManagedLedgerUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci91dGlsL01hbmFnZWRMZWRnZXJVdGlscy5qYXZh) | 77.27% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24511?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [...roker/service/persistent/MessageDeduplication.java](https://app.codecov.io/gh/apache/pulsar/pull/24511?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fpersistent%2FMessageDeduplication.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvTWVzc2FnZURlZHVwbGljYXRpb24uamF2YQ==) | 96.00% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24511?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   
   <details><summary>Additional details and impacted files</summary>
   
   
   [![Impacted file tree graph](https://app.codecov.io/gh/apache/pulsar/pull/24511/graphs/tree.svg?width=650&height=150&src=pr&token=acYqCpsK9J&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)](https://app.codecov.io/gh/apache/pulsar/pull/24511?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #24511      +/-   ##
   ============================================
   + Coverage     73.57%   74.32%   +0.75%     
   - Complexity    32624    32970     +346     
   ============================================
     Files          1877     1876       -1     
     Lines        139502   146270    +6768     
     Branches      15299    16773    +1474     
   ============================================
   + Hits         102638   108722    +6084     
   - Misses        28908    28921      +13     
   - Partials       7956     8627     +671     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24511/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24511/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.76% <2.06%> (+2.18%)` | :arrow_up: |
   | [systests](https://app.codecov.io/gh/apache/pulsar/pull/24511/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.30% <2.06%> (-1.02%)` | :arrow_down: |
   | [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24511/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.81% <92.78%> (+0.96%)` | :arrow_up: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24511?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [...he/bookkeeper/mledger/ManagedLedgerReplayTask.java](https://app.codecov.io/gh/apache/pulsar/pull/24511?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2FManagedLedgerReplayTask.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9NYW5hZ2VkTGVkZ2VyUmVwbGF5VGFzay5qYXZh) | `100.00% <100.00%> (ø)` | |
   | [...roker/service/persistent/MessageDeduplication.java](https://app.codecov.io/gh/apache/pulsar/pull/24511?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fpersistent%2FMessageDeduplication.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvTWVzc2FnZURlZHVwbGljYXRpb24uamF2YQ==) | `79.59% <96.00%> (-1.32%)` | :arrow_down: |
   | [...he/bookkeeper/mledger/util/ManagedLedgerUtils.java](https://app.codecov.io/gh/apache/pulsar/pull/24511?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Futil%2FManagedLedgerUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci91dGlsL01hbmFnZWRMZWRnZXJVdGlscy5qYXZh) | `77.27% <77.27%> (ø)` | |
   
   ... and [1109 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24511/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   
   </details>
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][broker] Fix deduplication replay might never complete for exceptions [pulsar]

Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24511:
URL: https://github.com/apache/pulsar/pull/24511#discussion_r2229941961


##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerUtils.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.util;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.common.classification.InterfaceStability;
+
+/**
+ * This util class contains some future-based methods to replace callback-based APIs. With a callback-based API, if any
+ * exception is thrown in the callback, the callback will never have a chance to be called. While with a future-based
+ * API, if any exception is thrown in future's callback (e.g. `thenApply`), the future will eventually be completed
+ * exceptionally. In addition, future-based API is easier for users to switch a different executor to execute the
+ * callback (e.g. `thenApplyAsync`).
+ */
+@InterfaceStability.Evolving
+public class ManagedLedgerUtils {
+
+    public static CompletableFuture<ManagedCursor> openCursor(ManagedLedger ml, String cursorName) {
+        final var future = new CompletableFuture<ManagedCursor>();
+        ml.asyncOpenCursor(cursorName, new AsyncCallbacks.OpenCursorCallback() {
+            @Override
+            public void openCursorComplete(ManagedCursor cursor, Object ctx) {
+                future.complete(cursor);
+            }
+
+            @Override
+            public void openCursorFailed(ManagedLedgerException exception, Object ctx) {
+                future.completeExceptionally(exception);
+            }
+        }, null);
+        return future;
+    }
+
+    public static CompletableFuture<List<Entry>> readEntries(ManagedCursor cursor, int numberOfEntriesToRead,
+                                                             Position maxPosition) {
+        final var future = new CompletableFuture<List<Entry>>();
+        cursor.asyncReadEntries(numberOfEntriesToRead, new AsyncCallbacks.ReadEntriesCallback() {
+            @Override
+            public void readEntriesComplete(List<Entry> entries, Object ctx) {
+                future.complete(entries);
+            }
+
+            @Override
+            public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+                future.completeExceptionally(exception);
+            }
+        }, null, maxPosition);
+        return future;
+    }
+
+    public static CompletableFuture<Void> markDelete(ManagedCursor cursor, Position position,
+                                                     Map<String, Long> properties) {
+        final var future = new CompletableFuture<Void>();
+        cursor.asyncMarkDelete(position, properties, new AsyncCallbacks.MarkDeleteCallback() {
+            @Override
+            public void markDeleteComplete(Object ctx) {
+                future.complete(null);
+            }
+
+            @Override
+            public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
+                future.completeExceptionally(exception);
+            }
+        }, new Object());

Review Comment:
   Yes, I intended to use `null`, `new Object()` should be written by the auto completion.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org