You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/07/25 11:03:29 UTC
(pulsar) 01/02: [fix][broker] Fix Broker OOM due to too many waiting cursors and reuse a recycled OpReadEntry incorrectly (#24551)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f68bf2be85fa98ed949892ac7ad1e55e5643f206
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri Jul 25 16:55:34 2025 +0800
[fix][broker] Fix Broker OOM due to too many waiting cursors and reuse a recycled OpReadEntry incorrectly (#24551)
(cherry picked from commit 48dc9ec6211e97443b251c07c33219d5e6270c31)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 145 +++++++++++++++++----
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 1 +
.../service/persistent/PersistentSubscription.java | 3 -
.../client/api/FailoverSubscriptionTest.java | 85 ++++++++++++
4 files changed, 208 insertions(+), 26 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 63087419b09..a48fe8a9f9f 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -49,6 +49,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -155,10 +156,17 @@ public class ManagedCursorImpl implements ManagedCursor {
MarkDeleteEntry.class, "lastMarkDeleteEntry");
protected volatile MarkDeleteEntry lastMarkDeleteEntry;
+ /** Protects the method "asyncReadEntriesWithSkipOrWait" and "cancelPendingReadRequest" runs concurrently. **/
+ private final Object pendingReadOpMutex = new Object();
+ /**
+ * 'ManagedLedger.notifyCursors' relies on this CAS to avoid using "pendingReadOpMutex" to guarantee thread-safety,
+ * which improved the performance of publishing messages.
+ */
protected static final AtomicReferenceFieldUpdater<ManagedCursorImpl, OpReadEntry> WAITING_READ_OP_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, OpReadEntry.class, "waitingReadOp");
@SuppressWarnings("unused")
private volatile OpReadEntry waitingReadOp = null;
+ private DelayCheckForNewEntriesTask delayCheckForNewEntriesTask;
public static final int FALSE = 0;
public static final int TRUE = 1;
@@ -1021,25 +1029,78 @@ public class ManagedCursorImpl implements ManagedCursor {
OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
ctx, maxPosition, skipCondition);
- if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
- op.recycle();
- callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
- return;
- }
+ synchronized (pendingReadOpMutex) {
+ if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {
+ op.recycle();
+ callback.readEntriesFailed(new ManagedLedgerException.ConcurrentWaitCallbackException(), ctx);
+ return;
+ }
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name, op.readPosition);
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] [{}] Deferring retry of read at position {}", ledger.getName(), name,
+ op.readPosition);
+ }
+
+ // Check again for new entries after the configured time, then if still no entries are available
+ // register to be notified.
+ if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
+ delayCheckForNewEntriesTask = new DelayCheckForNewEntriesTask(op, callback, ctx);
+ } else {
+ // If there's no delay, check directly from the same thread
+ checkForNewEntries(op, callback, ctx);
+ }
}
+ }
+ }
- // Check again for new entries after the configured time, then if still no entries are available register
- // to be notified
- if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
- ledger.getScheduledExecutor()
- .schedule(() -> checkForNewEntries(op, callback, ctx),
- getConfig().getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);
- } else {
- // If there's no delay, check directly from the same thread
+ private enum DelayCheckForNewEntriesTaskState {
+ INIT, RUNNING, CANCELLED, DONE
+ }
+
+ private class DelayCheckForNewEntriesTask implements Runnable {
+
+ private final OpReadEntry op;
+ private final ReadEntriesCallback callback;
+ private final Object ctx;
+ private final ScheduledFuture<?> scheduledFuture;
+ private DelayCheckForNewEntriesTaskState state = DelayCheckForNewEntriesTaskState.INIT;
+
+ public DelayCheckForNewEntriesTask(OpReadEntry op, ReadEntriesCallback callback, Object ctx) {
+ this.op = op;
+ this.callback = callback;
+ this.ctx = ctx;
+ scheduledFuture = ledger.getScheduledExecutor().schedule(this,
+ getConfig().getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public void run() {
+ synchronized (pendingReadOpMutex) {
+ if (state != DelayCheckForNewEntriesTaskState.INIT) {
+ return;
+ }
+ state = DelayCheckForNewEntriesTaskState.RUNNING;
checkForNewEntries(op, callback, ctx);
+ state = DelayCheckForNewEntriesTaskState.DONE;
+ }
+ }
+
+ public boolean isDone() {
+ return state == DelayCheckForNewEntriesTaskState.DONE;
+ }
+
+ public boolean cancel() {
+ synchronized (pendingReadOpMutex) {
+ // Not all implementations of Executor guarantee that the Runnable will be no long be executed after a
+ // successful cancel, such as Guava MoreExecutors, see also https://github.com/google/guava/blob
+ // /v32.1.2/guava/src/com/google/common/util/concurrent/MoreExecutors.java#L709.
+ // The current task guarantees.
+ if (state != DelayCheckForNewEntriesTaskState.INIT) {
+ return false;
+ }
+ state = DelayCheckForNewEntriesTaskState.CANCELLED;
+ scheduledFuture.cancel(false);
+ return true;
}
}
}
@@ -1077,16 +1138,16 @@ public class ManagedCursorImpl implements ManagedCursor {
}
// Try to cancel the notification request
if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
+ ledger.removeWaitingCursor(this);
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancelled notification and scheduled read at {}", ledger.getName(),
- name, op.readPosition);
+ name, op.readPosition);
}
PENDING_READ_OPS_UPDATER.incrementAndGet(this);
ledger.asyncReadEntries(op);
} else {
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] notification was already cancelled", ledger.getName(), name);
- }
+ log.info("[{}] [{}] notification that new entries added was already be called, skipped the current"
+ + " new entry checking", ledger.getName(), name);
}
} else if (ledger.isTerminated()) {
// At this point we registered for notification and still there were no more available
@@ -1109,11 +1170,49 @@ public class ManagedCursorImpl implements ManagedCursor {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Cancel pending read request", ledger.getName(), name);
}
- final OpReadEntry op = WAITING_READ_OP_UPDATER.getAndSet(this, null);
- if (op != null) {
- op.recycle();
+ synchronized (pendingReadOpMutex) {
+ final OpReadEntry op = WAITING_READ_OP_UPDATER.get(this);
+ // Case 1: the pending read has executed, or there is no pending read.
+ if (op == null) {
+ return false;
+ }
+ Function<Boolean, Boolean> clearWaitingReadOp = removeWaitingCursor -> {
+ if (WAITING_READ_OP_UPDATER.compareAndSet(this, op, null)) {
+ if (removeWaitingCursor) {
+ ledger.removeWaitingCursor(this);
+ }
+ op.recycle();
+ return true;
+ } else {
+ // Managed ledger has noticed that new entries was added.
+ if (WAITING_READ_OP_UPDATER.get(this) == null) {
+ return false;
+ }
+ // It will never occur, it means the lock "pendingReadOpMutex" does not work as expected, which
+ // allowed other thread to modify the "waitingReadOp" concurrently.
+ // The waitingReadOp has been modified to other instance, which will never occur.
+ log.warn("[{}] [{}] Cancel pending request encountered an unexpected error, the lock"
+ + " \"pendingReadOpMutex\" does not work as expected, which allowed other"
+ + " thread to modify the \"waitingReadOp\" concurrently..", ledger.getName(), name);
+ return cancelPendingReadRequest();
+ }
+ };
+ // There is a pending read,
+ // Case 2: delayCheckForNewEntriesTask can be cancelled, no need to remove cursor from "ml.waitingCursors",
+ // because it has not added successfully yet.
+ if (delayCheckForNewEntriesTask != null && delayCheckForNewEntriesTask.cancel()) {
+ return clearWaitingReadOp.apply(false);
+ }
+ // Case 3: managedLedgerNewEntriesCheckDelayInMillis is "0".
+ // Case 4: delayCheckForNewEntriesTask has done, which has added cursor into "ml.waitingCursors".
+ if (delayCheckForNewEntriesTask == null || delayCheckForNewEntriesTask.isDone()) {
+ return clearWaitingReadOp.apply(true);
+ }
+ // Case 5: delayCheckForNewEntriesTask is running, but not done, which only occurs at a corner case. It
+ // only happens when the task is starting. Calling "cancelPendingReadRequest" here will release the lock
+ // "pendingReadOpMutex" and let the task go ahead.
+ return cancelPendingReadRequest();
}
- return op != null;
}
public boolean hasPendingReadRequest() {
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 54fa3dbe8e4..43c224e7875 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
@@ -1056,6 +1056,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
return;
} else if (!cursor.isDurable()) {
cursor.setState(ManagedCursorImpl.State.Closed);
+ cursor.cancelPendingReadRequest();
cursors.removeCursor(consumerName);
deactivateCursorByName(consumerName);
callback.deleteCursorComplete(ctx);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 612a9c10533..2a27090a990 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -367,14 +367,11 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
if (!isResetCursor) {
try {
topic.getManagedLedger().deleteCursor(cursor.getName());
- topic.getManagedLedger().removeWaitingCursor(cursor);
} catch (InterruptedException | ManagedLedgerException e) {
log.warn("[{}] [{}] Failed to remove non durable cursor", topic.getName(), subName, e);
}
}
});
- } else {
- topic.getManagedLedger().removeWaitingCursor(cursor);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/FailoverSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/FailoverSubscriptionTest.java
new file mode 100644
index 00000000000..6263193b0a4
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/FailoverSubscriptionTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.pulsar.client.api;
+
+import static org.testng.Assert.assertEquals;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker-api")
+public class FailoverSubscriptionTest extends ProducerConsumerBase {
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test(timeOut = 30_000, invocationCount = 5)
+ public void testWaitingCursorsCountAfterSwitchingActiveConsumers() throws Exception {
+ final String tp = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/tp");
+ final String subscription = "s1";
+ admin.topics().createNonPartitionedTopic(tp);
+ admin.topics().createSubscription(tp, subscription, MessageId.earliest);
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(tp, false).join().get();
+ Map<String, ConsumerImpl<byte[]>> consumerMap = new HashMap<>();
+ ConsumerImpl<byte[]> firstConsumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(tp)
+ .subscriptionType(SubscriptionType.Failover).subscriptionName(subscription).subscribe();
+ consumerMap.put(firstConsumer.getConsumerName(), firstConsumer);
+ PersistentDispatcherSingleActiveConsumer dispatcher =
+ (PersistentDispatcherSingleActiveConsumer) topic.getSubscription(subscription).getDispatcher();
+ for (int i = 0; i < 100; i++) {
+ ConsumerImpl<byte[]> consumerLoop = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(tp)
+ .subscriptionType(SubscriptionType.Failover).subscriptionName(subscription).subscribe();
+ consumerMap.put(consumerLoop.getConsumerName(), consumerLoop);
+ if (dispatcher != null && dispatcher.getActiveConsumer() != null) {
+ String activeConsumer = dispatcher.getActiveConsumer().consumerName();
+ consumerMap.remove(activeConsumer).close();
+ }
+ }
+
+ ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger();
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(managedLedger.getWaitingCursorsCount(), 1);
+ });
+
+ // cleanup.
+ for (ConsumerImpl<byte[]> consumer : consumerMap.values()) {
+ consumer.close();
+ }
+ admin.topics().delete(tp, false);
+ }
+}