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/05/28 03:36:27 UTC
(pulsar) branch branch-2.11 updated: [fix] [ml] Add entry fail due to race condition about add entry failed/timeout and switch ledger (#22221)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new 72deaf51b76 [fix] [ml] Add entry fail due to race condition about add entry failed/timeout and switch ledger (#22221)
72deaf51b76 is described below
commit 72deaf51b76189f24d229bd1c7e5dcaaeadf2b4a
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Tue May 21 15:50:53 2024 +0800
[fix] [ml] Add entry fail due to race condition about add entry failed/timeout and switch ledger (#22221)
(cherry picked from commit ae9616b7ee41e77dde590045b8e1d5bd04192ffc)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 30 ++++++++----
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 39 +++++++++-------
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 53 +++++++++++++++++++++-
.../bookkeeper/client/PulsarMockBookKeeper.java | 7 +++
.../bookkeeper/client/PulsarMockLedgerHandle.java | 7 +++
5 files changed, 110 insertions(+), 26 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 3f502db7f0a..db8eb8d3059 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
@@ -58,6 +58,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
@@ -240,6 +241,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
private volatile long lastAddEntryTimeMs = 0;
private long inactiveLedgerRollOverTimeMs = 0;
+ /** A signal that may trigger all the subsequent OpAddEntry of current ledger to be failed due to timeout. **/
+ protected volatile AtomicBoolean currentLedgerTimeoutTriggered;
+
protected static final int DEFAULT_LEDGER_DELETE_RETRIES = 3;
protected static final int DEFAULT_LEDGER_DELETE_BACKOFF_TIME_SEC = 60;
@@ -513,6 +517,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
STATE_UPDATER.set(this, State.LedgerOpened);
updateLastLedgerCreatedTimeAndScheduleRolloverTask();
currentLedger = lh;
+ currentLedgerTimeoutTriggered = new AtomicBoolean();
lastConfirmedEntry = new PositionImpl(lh.getId(), -1);
// bypass empty ledgers, find last ledger with Message if possible.
@@ -755,7 +760,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// Jump to specific thread to avoid contention from writers writing from different threads
executor.executeOrdered(name, safeRun(() -> {
- OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this, buffer, callback, ctx);
+ OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this, buffer, callback, ctx,
+ currentLedgerTimeoutTriggered);
internalAsyncAddEntry(addOperation);
}));
}
@@ -771,7 +777,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// Jump to specific thread to avoid contention from writers writing from different threads
executor.executeOrdered(name, safeRun(() -> {
- OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this, buffer, numberOfMessages, callback, ctx);
+ OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this, buffer, numberOfMessages, callback, ctx,
+ currentLedgerTimeoutTriggered);
internalAsyncAddEntry(addOperation);
}));
}
@@ -823,6 +830,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// Write into lastLedger
addOperation.setLedger(currentLedger);
+ addOperation.setTimeoutTriggered(currentLedgerTimeoutTriggered);
++currentLedgerEntries;
currentLedgerSize += addOperation.data.readableBytes();
@@ -1539,6 +1547,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
LedgerHandle originalCurrentLedger = currentLedger;
ledgers.put(lh.getId(), newLedger);
currentLedger = lh;
+ currentLedgerTimeoutTriggered = new AtomicBoolean();
currentLedgerEntries = 0;
currentLedgerSize = 0;
updateLedgersIdsComplete(originalCurrentLedger);
@@ -1635,9 +1644,11 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
if (existsOp != null) {
// If op is used by another ledger handle, we need to close it and create a new one
if (existsOp.ledger != null) {
- existsOp.close();
- existsOp = OpAddEntry.createNoRetainBuffer(existsOp.ml, existsOp.data,
- existsOp.getNumberOfMessages(), existsOp.callback, existsOp.ctx);
+ existsOp = existsOp.duplicateAndClose(currentLedgerTimeoutTriggered);
+ } else {
+ // This scenario should not happen.
+ log.warn("[{}] An OpAddEntry's ledger is empty.", name);
+ existsOp.setTimeoutTriggered(currentLedgerTimeoutTriggered);
}
existsOp.setLedger(currentLedger);
pendingAddEntries.add(existsOp);
@@ -4002,13 +4013,14 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
OpAddEntry opAddEntry = pendingAddEntries.peek();
if (opAddEntry != null) {
- final long finalAddOpCount = opAddEntry.addOpCount;
boolean isTimedOut = opAddEntry.lastInitTime != -1
&& TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - opAddEntry.lastInitTime) >= timeoutSec;
if (isTimedOut) {
- log.error("Failed to add entry for ledger {} in time-out {} sec",
- (opAddEntry.ledger != null ? opAddEntry.ledger.getId() : -1), timeoutSec);
- opAddEntry.handleAddTimeoutFailure(opAddEntry.ledger, finalAddOpCount);
+ log.warn("[{}] Failed to add entry {}:{} in time-out {} sec", this.name,
+ opAddEntry.ledger != null ? opAddEntry.ledger.getId() : -1,
+ opAddEntry.entryId, timeoutSec);
+ currentLedgerTimeoutTriggered.set(true);
+ opAddEntry.handleAddFailure(opAddEntry.ledger);
}
}
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
index 1b6c6e96175..ba176561a65 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
@@ -24,8 +24,10 @@ import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
import io.netty.util.ReferenceCountUtil;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
@@ -46,7 +48,7 @@ import org.apache.bookkeeper.util.SafeRunnable;
public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback {
protected ManagedLedgerImpl ml;
LedgerHandle ledger;
- private long entryId;
+ long entryId;
private int numberOfMessages;
@SuppressWarnings("unused")
@@ -69,6 +71,9 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
AtomicReferenceFieldUpdater.newUpdater(OpAddEntry.class, OpAddEntry.State.class, "state");
volatile State state;
+ @Setter
+ private AtomicBoolean timeoutTriggered;
+
enum State {
OPEN,
INITIATED,
@@ -77,8 +82,8 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
}
public static OpAddEntry createNoRetainBuffer(ManagedLedgerImpl ml, ByteBuf data, AddEntryCallback callback,
- Object ctx) {
- OpAddEntry op = createOpAddEntryNoRetainBuffer(ml, data, callback, ctx);
+ Object ctx, AtomicBoolean timeoutTriggered) {
+ OpAddEntry op = createOpAddEntryNoRetainBuffer(ml, data, callback, ctx, timeoutTriggered);
if (log.isDebugEnabled()) {
log.debug("Created new OpAddEntry {}", op);
}
@@ -86,8 +91,9 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
}
public static OpAddEntry createNoRetainBuffer(ManagedLedgerImpl ml, ByteBuf data, int numberOfMessages,
- AddEntryCallback callback, Object ctx) {
- OpAddEntry op = createOpAddEntryNoRetainBuffer(ml, data, callback, ctx);
+ AddEntryCallback callback, Object ctx,
+ AtomicBoolean timeoutTriggered) {
+ OpAddEntry op = createOpAddEntryNoRetainBuffer(ml, data, callback, ctx, timeoutTriggered);
op.numberOfMessages = numberOfMessages;
if (log.isDebugEnabled()) {
log.debug("Created new OpAddEntry {}", op);
@@ -96,7 +102,8 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
}
private static OpAddEntry createOpAddEntryNoRetainBuffer(ManagedLedgerImpl ml, ByteBuf data,
- AddEntryCallback callback, Object ctx) {
+ AddEntryCallback callback, Object ctx,
+ AtomicBoolean timeoutTriggered) {
OpAddEntry op = RECYCLER.get();
op.ml = ml;
op.ledger = null;
@@ -110,6 +117,7 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
op.startTime = System.nanoTime();
op.state = State.OPEN;
op.payloadProcessorHandle = null;
+ op.timeoutTriggered = timeoutTriggered;
ml.mbean.addAddEntrySample(op.dataLength);
return op;
}
@@ -167,7 +175,9 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
if (!STATE_UPDATER.compareAndSet(OpAddEntry.this, State.INITIATED, State.COMPLETED)) {
log.warn("[{}] The add op is terminal legacy callback for entry {}-{} adding.", ml.getName(), lh.getId(),
entryId);
- OpAddEntry.this.recycle();
+ // Since there is a thread is coping this object, do not recycle this object to avoid other problems.
+ // For example: we recycled this object, other thread get a null "opAddEntry.{variable_name}".
+ // Recycling is not mandatory, JVM GC will collect it.
return;
}
@@ -188,7 +198,7 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
lh.getId(), entryId, dataLength, rc);
}
- if (rc != BKException.Code.OK) {
+ if (rc != BKException.Code.OK || timeoutTriggered.get()) {
handleAddFailure(lh);
} else {
// Trigger addComplete callback in a thread hashed on the managed ledger name
@@ -291,13 +301,6 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
return false;
}
- void handleAddTimeoutFailure(final LedgerHandle ledger, Object ctx) {
- if (checkAndCompleteOp(ctx)) {
- this.close();
- this.handleAddFailure(ledger);
- }
- }
-
/**
* It handles add failure on the given ledger. it can be triggered when add-entry fails or times out.
*
@@ -317,8 +320,11 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
}));
}
- void close() {
+ OpAddEntry duplicateAndClose(AtomicBoolean timeoutTriggered) {
STATE_UPDATER.set(OpAddEntry.this, State.CLOSED);
+ OpAddEntry duplicate =
+ OpAddEntry.createNoRetainBuffer(ml, data, getNumberOfMessages(), callback, ctx, timeoutTriggered);
+ return duplicate;
}
public State getState() {
@@ -373,6 +379,7 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
startTime = -1;
lastInitTime = -1;
payloadProcessorHandle = null;
+ timeoutTriggered = null;
recyclerHandle.recycle(this);
}
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 9cf9c49427f..0a37abe8d71 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
@@ -140,6 +140,7 @@ import org.apache.pulsar.metadata.api.extended.SessionEvent;
import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore;
import org.awaitility.Awaitility;
import org.awaitility.reflect.WhiteboxImpl;
+import org.eclipse.jetty.util.BlockingArrayQueue;
import org.mockito.Mockito;
import org.powermock.reflect.Whitebox;
import org.testng.Assert;
@@ -3163,6 +3164,55 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
ledger.close();
}
+ @Test
+ public void testAddEntryResponseTimeout() throws Exception {
+ // Create ML with feature Add Entry Timeout Check.
+ final ManagedLedgerConfig config = new ManagedLedgerConfig().setAddEntryTimeoutSeconds(2);
+ final ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("ml1", config);
+ final ManagedCursor cursor = ledger.openCursor("c1");
+ final CollectCtxAddEntryCallback collectCtxAddEntryCallback = new CollectCtxAddEntryCallback();
+
+ // Insert a response delay.
+ bkc.addEntryResponseDelay(8, TimeUnit.SECONDS);
+
+ // Add two entries.
+ final byte[] msg1 = new byte[]{1};
+ final byte[] msg2 = new byte[]{2};
+ int ctx1 = 1;
+ int ctx2 = 2;
+ ledger.asyncAddEntry(msg1, collectCtxAddEntryCallback, ctx1);
+ ledger.asyncAddEntry(msg2, collectCtxAddEntryCallback, ctx2);
+ // Verify all write requests are completed.
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(collectCtxAddEntryCallback.addCompleteCtxList, Arrays.asList(1, 2));
+ });
+ Entry entry1 = cursor.readEntries(1).get(0);
+ assertEquals(entry1.getData(), msg1);
+ entry1.release();
+ Entry entry2 = cursor.readEntries(1).get(0);
+ assertEquals(entry2.getData(), msg2);
+ entry2.release();
+
+ // cleanup.
+ factory.delete(ledger.name);
+ }
+
+ private static class CollectCtxAddEntryCallback implements AddEntryCallback {
+
+ public List<Object> addCompleteCtxList = new BlockingArrayQueue<>();
+ public List<Object> addFailedCtxList = new BlockingArrayQueue<>();
+
+ @Override
+ public void addComplete(Position position, ByteBuf entryData, Object ctx) {
+ addCompleteCtxList.add(ctx);
+ }
+
+ @Override
+ public void addFailed(ManagedLedgerException exception, Object ctx) {
+ addFailedCtxList.add(ctx);
+ }
+ }
+
/**
* It verifies that if bk-client doesn't complete the add-entry in given time out then broker is resilient enought
* to create new ledger and add entry successfully.
@@ -3238,7 +3288,8 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
List<OpAddEntry> oldOps = new ArrayList<>();
for (int i = 0; i < 10; i++) {
- OpAddEntry op = OpAddEntry.createNoRetainBuffer(ledger, ByteBufAllocator.DEFAULT.buffer(128).retain(), null, null);
+ OpAddEntry op = OpAddEntry.createNoRetainBuffer(ledger,
+ ByteBufAllocator.DEFAULT.buffer(128).retain(), null, null, new AtomicBoolean());
if (i > 4) {
op.setLedger(mock(LedgerHandle.class));
}
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 eebe46c738b..9ab0f2bdbcc 100644
--- a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java
+++ b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java
@@ -18,6 +18,7 @@
*/
package org.apache.bookkeeper.client;
+import static com.google.common.base.Preconditions.checkArgument;
import com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.Arrays;
@@ -89,6 +90,7 @@ public class PulsarMockBookKeeper extends BookKeeper {
}
final Queue<Long> addEntryDelaysMillis = new ConcurrentLinkedQueue<>();
+ final Queue<Long> addEntryResponseDelaysMillis = new ConcurrentLinkedQueue<>();
final List<CompletableFuture<Void>> failures = new ArrayList<>();
final List<CompletableFuture<Void>> addEntryFailures = new ArrayList<>();
@@ -367,6 +369,11 @@ public class PulsarMockBookKeeper extends BookKeeper {
addEntryDelaysMillis.add(unit.toMillis(delay));
}
+ public synchronized void addEntryResponseDelay(long delay, TimeUnit unit) {
+ checkArgument(delay >= 0, "The delay time must not be negative.");
+ addEntryResponseDelaysMillis.add(unit.toMillis(delay));
+ }
+
static int getExceptionCode(Throwable t) {
if (t instanceof BKException) {
return ((BKException) t).getCode();
diff --git a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
index ab85c9fc888..9e69e40bb6f 100644
--- a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
+++ b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
@@ -197,6 +197,13 @@ public class PulsarMockLedgerHandle extends LedgerHandle {
cb.addComplete(PulsarMockBookKeeper.getExceptionCode(exception),
PulsarMockLedgerHandle.this, LedgerHandle.INVALID_ENTRY_ID, ctx);
} else {
+ Long responseDelayMillis = bk.addEntryResponseDelaysMillis.poll();
+ if (responseDelayMillis != null) {
+ try {
+ Thread.sleep(responseDelayMillis);
+ } catch (InterruptedException e) {
+ }
+ }
cb.addComplete(BKException.Code.OK, PulsarMockLedgerHandle.this, entryId, ctx);
}
}, bk.executor);