This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new 07a3767df36 [fix] [ml] Add entry fail due to race condition about add
entry failed/timeout and switch ledger (#22221)
07a3767df36 is described below
commit 07a3767df36dd86717887075c3bf76ba0fbd2081
Author: fengyubiao <yubiao.f...@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 +++++++++-------
.../mledger/impl/ShadowManagedLedgerImpl.java | 3 ++
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 53 +++++++++++++++++++++-
.../bookkeeper/client/PulsarMockBookKeeper.java | 7 +++
.../bookkeeper/client/PulsarMockLedgerHandle.java | 7 +++
6 files changed, 113 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 07d0d40569a..167c0a1bad3 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
@@ -59,6 +59,7 @@ import java.util.concurrent.Executor;
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 {
protected 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;
private static final String MIGRATION_STATE_PROPERTY = "migrated";
@@ -532,6 +536,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.
@@ -774,7 +779,8 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
// Jump to specific thread to avoid contention from writers writing
from different threads
executor.execute(() -> {
- OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this,
buffer, callback, ctx);
+ OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this,
buffer, callback, ctx,
+ currentLedgerTimeoutTriggered);
internalAsyncAddEntry(addOperation);
});
}
@@ -790,7 +796,8 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
// Jump to specific thread to avoid contention from writers writing
from different threads
executor.execute(() -> {
- OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this,
buffer, numberOfMessages, callback, ctx);
+ OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this,
buffer, numberOfMessages, callback, ctx,
+ currentLedgerTimeoutTriggered);
internalAsyncAddEntry(addOperation);
});
}
@@ -842,6 +849,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
// Write into lastLedger
addOperation.setLedger(currentLedger);
+ addOperation.setTimeoutTriggered(currentLedgerTimeoutTriggered);
++currentLedgerEntries;
currentLedgerSize += addOperation.data.readableBytes();
@@ -1585,6 +1593,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);
@@ -1668,9 +1677,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);
@@ -4164,13 +4175,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 ae2beafb643..acbb0da5a4e 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;
@@ -45,7 +47,7 @@ import
org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
public class OpAddEntry implements AddCallback, CloseCallback, Runnable {
protected ManagedLedgerImpl ml;
LedgerHandle ledger;
- private long entryId;
+ long entryId;
private int numberOfMessages;
@SuppressWarnings("unused")
@@ -68,6 +70,9 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
AtomicReferenceFieldUpdater.newUpdater(OpAddEntry.class,
OpAddEntry.State.class, "state");
volatile State state;
+ @Setter
+ private AtomicBoolean timeoutTriggered;
+
enum State {
OPEN,
INITIATED,
@@ -76,8 +81,8 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
}
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);
}
@@ -85,8 +90,9 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
}
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);
@@ -95,7 +101,8 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
}
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;
@@ -109,6 +116,7 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
op.startTime = System.nanoTime();
op.state = State.OPEN;
op.payloadProcessorHandle = null;
+ op.timeoutTriggered = timeoutTriggered;
ml.mbean.addAddEntrySample(op.dataLength);
return op;
}
@@ -176,7 +184,9 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
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;
}
@@ -200,7 +210,7 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
lh == null ? -1 : 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
@@ -307,13 +317,6 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
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.
*
@@ -333,8 +336,11 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
});
}
- 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() {
@@ -389,6 +395,7 @@ public class OpAddEntry implements AddCallback,
CloseCallback, Runnable {
startTime = -1;
lastInitTime = -1;
payloadProcessorHandle = null;
+ timeoutTriggered = null;
recyclerHandle.recycle(this);
}
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
index 8b2742d9587..ec5b006c474 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
@@ -25,6 +25,7 @@ import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.AsyncCallback;
@@ -54,6 +55,8 @@ public class ShadowManagedLedgerImpl extends
ManagedLedgerImpl {
String name, final
Supplier<CompletableFuture<Boolean>> mlOwnershipChecker) {
super(factory, bookKeeper, store, config, scheduledExecutor, name,
mlOwnershipChecker);
this.sourceMLName = config.getShadowSourceName();
+ // ShadowManagedLedgerImpl does not implement add entry timeout yet,
so this variable will always be false.
+ this.currentLedgerTimeoutTriggered = new AtomicBoolean(false);
}
/**
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 122bada487a..4f521f1e99e 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
@@ -146,6 +146,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.testng.Assert;
import org.testng.annotations.DataProvider;
@@ -3185,6 +3186,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 enough
* to create new ledger and add entry successfully.
@@ -3260,7 +3310,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 f0d279ef250..4516cfea01f 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 dea33a0e676..aa61e541d0d 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);
