You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2024/12/25 08:59:18 UTC
(pulsar) branch branch-4.0 updated: [fix][ml] Topic load timeout due to ml data ledger future never finishes (#23772)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 6697aa165ed [fix][ml] Topic load timeout due to ml data ledger future never finishes (#23772)
6697aa165ed is described below
commit 6697aa165ed440c68eaf42799f86ceb85c97b244
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Wed Dec 25 14:14:01 2024 +0800
[fix][ml] Topic load timeout due to ml data ledger future never finishes (#23772)
### Motivation
**Background**
There is a mechanism that repeatedly prevents the callback of ML data ledger creation:
- Start a scheduled task to check whether the creation will be timeout.
- Received a callback
- Check whether the future(`@param ctx` of `BK.createAsync`) has been done or not.
- If done: it means the creation has timeout before the creation is completed
- Otherwise: it is a real callback from BK.
**Issue:**
But the timeout event will call the same callback as above, then the steps are as follows, which you ca reproduce by the test `testCreateDataLedgerTimeout`:
- Start creating a data ledger
- Call `BK.createAsync`
- Timeout
- Mark the future(`@param ctx` of `BK.createAsync`) as completed exceptionally.
- Trigger the callback related to ledger creation.
- Check whether the future(`@param ctx` of `BK.createAsync`) has been done or not.
- If done: do nothing.
- Creation is compelled.
- Trigger the callback related to ledger creation.
- Check whether the future(`@param ctx` of `BK.createAsync`) has been done or not.
- If done: do nothing.
- Issue: The callback for ledger creation will never be called.
### Modifications
Fix the issue
### 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: x
(cherry picked from commit 9699dc25deb5eb701de88dc209e6c42efe58556c)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 3 ++-
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 25 ++++++++++++++++++++++
.../bookkeeper/client/PulsarMockBookKeeper.java | 15 ++++++++++++-
3 files changed, 41 insertions(+), 2 deletions(-)
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
index 926e7c7f810..966aa068f2f 100644
--- 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
@@ -574,6 +574,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
executor.execute(() -> {
mbean.endDataLedgerCreateOp();
if (rc != BKException.Code.OK) {
+ log.error("[{}] Error creating ledger rc={} {}", name, rc, BKException.getMessage(rc));
callback.initializeFailed(createManagedLedgerException(rc));
return;
}
@@ -4141,7 +4142,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
protected boolean checkAndCompleteLedgerOpTask(int rc, LedgerHandle lh, Object ctx) {
if (ctx instanceof CompletableFuture) {
// ledger-creation is already timed out and callback is already completed so, delete this ledger and return.
- if (((CompletableFuture) ctx).complete(lh)) {
+ if (((CompletableFuture) ctx).complete(lh) || rc == BKException.Code.TimeoutException) {
return false;
} else {
if (rc == BKException.Code.OK) {
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index 67c928c8a56..04f8eecbe9a 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -4236,6 +4236,31 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
assertNotNull(ml.newNonDurableCursor(Position));
}
+ @Test(timeOut = 60 * 1000)
+ public void testCreateDataLedgerTimeout() throws Exception {
+ String mlName = UUID.randomUUID().toString();
+ ManagedLedgerFactoryImpl factory = null;
+ ManagedLedger ml = null;
+ try {
+ factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMetadataOperationsTimeoutSeconds(5);
+ bkc.delay(10 * 1000);
+ ml = factory.open(mlName, config);
+ fail("Should get a timeout ex");
+ } catch (ManagedLedgerException ex) {
+ assertTrue(ex.getMessage().contains("timeout"));
+ } finally {
+ // cleanup.
+ if (ml != null) {
+ ml.delete();
+ }
+ if (factory != null) {
+ factory.shutdown();
+ }
+ }
+ }
+
/***
* When a ML tries to create a ledger, it will create a delay task to check if the ledger create request is timeout.
* But we should guarantee that the delay task should be canceled after the ledger create request responded.
diff --git a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java
index 11ec2dec938..1e979206e16 100644
--- a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java
+++ b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.client;
import static com.google.common.base.Preconditions.checkArgument;
import com.google.common.collect.Lists;
+import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -35,6 +36,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -70,6 +72,7 @@ public class PulsarMockBookKeeper extends BookKeeper {
final OrderedExecutor orderedExecutor;
final ExecutorService executor;
+ final ScheduledExecutorService scheduler;
@Override
public ClientConfiguration getConf() {
@@ -97,6 +100,7 @@ public class PulsarMockBookKeeper extends BookKeeper {
public PulsarMockBookKeeper(OrderedExecutor orderedExecutor) throws Exception {
this.orderedExecutor = orderedExecutor;
this.executor = orderedExecutor.chooseThread();
+ scheduler = Executors.newScheduledThreadPool(1, new DefaultThreadFactory("mock-bk-scheduler"));
}
@Override
@@ -290,7 +294,7 @@ public class PulsarMockBookKeeper extends BookKeeper {
for (PulsarMockLedgerHandle ledger : ledgers.values()) {
ledger.entries.clear();
}
-
+ scheduler.shutdown();
ledgers.clear();
}
@@ -331,6 +335,15 @@ public class PulsarMockBookKeeper extends BookKeeper {
return failures.isEmpty() ? defaultResponse : failures.remove(0);
}
+ public void delay(long millis) {
+ CompletableFuture<Void> delayFuture = new CompletableFuture<>();
+ scheduler.schedule(() -> {
+ delayFuture.complete(null);
+ }, millis, TimeUnit.MILLISECONDS);
+ failures.add(delayFuture);
+ }
+
+
public void failNow(int rc) {
failAfter(0, rc);
}