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 2026/07/26 19:52:13 UTC
(pulsar) branch branch-4.2 updated (58a059f2d75b -> 3ad321c3a89a)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 58a059f2d75b [fix][fn] Forward source message properties in Python runtime (#26191)
new 5d030f8e9d57 [fix][client] Fix unAckedMessageTracker cleanup on multi-topics batch ack (#26001)
new bc69930c401c [fix][broker] Fix TableViewLoadDataStoreImpl close deadlock that stalls broker shutdown (#26243)
new 338b922d8b18 [fix][broker] Prevent stale service unit callbacks from dropping active lookup and cleanup jobs (#26146)
new 3ad321c3a89a [fix][broker] Release entry on GetLastMessageId when parseMessageMetadata throws (#26089)
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../channel/ServiceUnitStateChannelImpl.java | 37 +++--
.../store/TableViewLoadDataStoreImpl.java | 21 ++-
.../apache/pulsar/broker/service/ServerCnx.java | 15 +-
.../channel/ServiceUnitStateChannelTest.java | 155 ++++++++++++++++++++-
.../extensions/store/LoadDataStoreTest.java | 53 +++++++
.../service/GetLastMessageIdEntryLeakTest.java | 93 +++++++++++++
.../client/impl/MultiTopicsConsumerImpl.java | 2 +-
.../client/impl/MultiTopicsConsumerImplTest.java | 35 +++++
8 files changed, 390 insertions(+), 21 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/GetLastMessageIdEntryLeakTest.java
(pulsar) 01/04: [fix][client] Fix unAckedMessageTracker cleanup on multi-topics batch ack (#26001)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5d030f8e9d576396f5a308c986f09368ac915d42
Author: Dream95 <86...@qq.com>
AuthorDate: Sun Jul 26 02:06:33 2026 +0800
[fix][client] Fix unAckedMessageTracker cleanup on multi-topics batch ack (#26001)
Signed-off-by: Dream95 <zh...@163.com>
(cherry picked from commit 3ebf796497bfda55c84ff985f54f0f25ac829734)
---
.../client/impl/MultiTopicsConsumerImpl.java | 2 +-
.../client/impl/MultiTopicsConsumerImplTest.java | 35 ++++++++++++++++++++++
2 files changed, 36 insertions(+), 1 deletion(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index f611abb179f6..faad4b2daf13 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -531,7 +531,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
}
consumerToMessageIds.forEach((consumer, messageIds) -> {
resultFutures.add(consumer.doAcknowledgeWithTxn(messageIds, ackType, properties, txn)
- .thenAccept((res) -> messageIdList.forEach(unAckedMessageTracker::remove)));
+ .thenAccept((res) -> messageIds.forEach(unAckedMessageTracker::remove)));
});
}
return CompletableFuture.allOf(resultFutures.toArray(new CompletableFuture[0]));
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java
index b4d007d855f8..f16702577a30 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java
@@ -23,6 +23,7 @@ import static org.apache.pulsar.client.impl.ClientTestFixtures.createExceptionFu
import static org.apache.pulsar.client.impl.ClientTestFixtures.createPulsarClientMockWithMockedClientCnx;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -317,4 +318,38 @@ public class MultiTopicsConsumerImplTest {
verify(partitionConsumer1).closeAsync();
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testBatchAcknowledgeRemovesOnlyAckedMessageIdsFromTracker() {
+ String topic0 = "persistent://public/default/topic-a-partition-0";
+ String topic1 = "persistent://public/default/topic-b-partition-0";
+
+ ConsumerConfigurationData<byte[]> conf = new ConsumerConfigurationData<>();
+ conf.setSubscriptionName("subscriptionName");
+ conf.setAckTimeoutMillis(1000);
+ MultiTopicsConsumerImpl<byte[]> impl = createMultiTopicsConsumer(conf);
+ impl.setState(HandlerState.State.Ready);
+
+ ConsumerImpl<byte[]> consumer0 = mock(ConsumerImpl.class);
+ ConsumerImpl<byte[]> consumer1 = mock(ConsumerImpl.class);
+ CompletableFuture<Void> pendingAck = new CompletableFuture<>();
+ when(consumer0.getTopic()).thenReturn(topic0);
+ when(consumer1.getTopic()).thenReturn(topic1);
+ when(consumer0.doAcknowledgeWithTxn(anyList(), any(), any(), any())).thenReturn(pendingAck);
+ when(consumer1.doAcknowledgeWithTxn(anyList(), any(), any(), any())).thenReturn(new CompletableFuture<>());
+
+ impl.consumers.put(topic0, consumer0);
+ impl.consumers.put(topic1, consumer1);
+
+ TopicMessageIdImpl messageId0 = new TopicMessageIdImpl(topic0, new MessageIdImpl(1, 1, 0));
+ TopicMessageIdImpl messageId1 = new TopicMessageIdImpl(topic1, new MessageIdImpl(2, 2, 0));
+ impl.getUnAckedMessageTracker().add(messageId0);
+ impl.getUnAckedMessageTracker().add(messageId1);
+
+ impl.acknowledgeAsync(Arrays.asList(messageId0, messageId1));
+ pendingAck.complete(null);
+
+ assertEquals(impl.getUnAckedMessageTracker().size(), 1);
+ }
+
}
(pulsar) 04/04: [fix][broker] Release entry on GetLastMessageId when parseMessageMetadata throws (#26089)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3ad321c3a89a8161fc488f003a4a76ecbe1a304a
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Sun Jul 26 18:41:34 2026 +0800
[fix][broker] Release entry on GetLastMessageId when parseMessageMetadata throws (#26089)
Co-authored-by: maxlisongsong <ma...@didiglobal.com>
(cherry picked from commit 4790c386c6d57f2016765f459cad1d77e4f17796)
---
.../apache/pulsar/broker/service/ServerCnx.java | 15 ++--
.../service/GetLastMessageIdEntryLeakTest.java | 93 ++++++++++++++++++++++
2 files changed, 102 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index b8b1fbb99702..91cf28e2e7ea 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -2522,13 +2522,16 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}, null);
CompletableFuture<Integer> batchSizeFuture = entryFuture.thenApply(entry -> {
- MessageMetadata metadata = entry.getMessageMetadata();
- if (metadata == null) {
- metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
+ try {
+ MessageMetadata metadata = entry.getMessageMetadata();
+ if (metadata == null) {
+ metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
+ }
+ int batchSize = metadata.getNumMessagesInBatch();
+ return metadata.hasNumMessagesInBatch() ? batchSize : -1;
+ } finally {
+ entry.release();
}
- int batchSize = metadata.getNumMessagesInBatch();
- entry.release();
- return metadata.hasNumMessagesInBatch() ? batchSize : -1;
});
batchSizeFuture.whenComplete((batchSize, e) -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/GetLastMessageIdEntryLeakTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/GetLastMessageIdEntryLeakTest.java
new file mode 100644
index 000000000000..1c3bc3404049
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/GetLastMessageIdEntryLeakTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.broker.service;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import lombok.Cleanup;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class GetLastMessageIdEntryLeakTest extends SharedPulsarBaseTest {
+
+ /**
+ * Reproduces the ByteBuf leak in ServerCnx#getLargestBatchIndexWhenPossible: when
+ * Commands.parseMessageMetadata throws on the entry read for the last position, the entry (and
+ * its backing ByteBuf) must still be released.
+ */
+ @Test
+ public void testEntryReleasedWhenParseMetadataThrows() throws Exception {
+ final String topic = newTopicName();
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("payload".getBytes());
+
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
+ .topic(topic).subscriptionName("sub").subscribe();
+
+ PersistentTopic persistentTopic = (PersistentTopic) getTopicReference(topic).get();
+
+ // A 2-byte, non-magic buffer: parseMessageMetadata reads it but fails at readUnsignedInt
+ // (needs 4 bytes), which is exactly the corrupt-entry case that triggers the leak.
+ ByteBuf corruptBuf = Unpooled.buffer(2);
+ corruptBuf.writeShort(0x0000);
+
+ // Build the entry eagerly so its retain happens now; then drop our own ref so the entry
+ // "owns" the only ref and we can observe whether the broker releases it.
+ Position lastPosition = ((ManagedLedgerImpl) persistentTopic.getManagedLedger()).getLastConfirmedEntry();
+ EntryImpl corruptEntry =
+ EntryImpl.create(lastPosition.getLedgerId(), lastPosition.getEntryId(), corruptBuf);
+ corruptBuf.release();
+ assertEquals(corruptBuf.refCnt(), 1);
+
+ // Spy the real ManagedLedgerImpl so asyncReadEntry hands back our corrupt entry, while
+ // getLastPosition() etc. still delegate to the real ledger.
+ ManagedLedgerImpl spyLedger = spy((ManagedLedgerImpl) persistentTopic.getManagedLedger());
+ doAnswer(inv -> {
+ ReadEntryCallback callback = inv.getArgument(1);
+ callback.readEntryComplete(corruptEntry, inv.getArgument(2));
+ return null;
+ }).when(spyLedger).asyncReadEntry(any(Position.class), any(ReadEntryCallback.class), any());
+
+ FieldUtils.writeField(persistentTopic, "ledger", spyLedger, true);
+
+ // The broker fails the request (MetadataError) - that's expected; the point is the buffer.
+ assertThrows(Exception.class, consumer::getLastMessageId);
+
+ // Before the fix: parseMessageMetadata throws, entry.release() is skipped -> refCnt stays 1.
+ // After the fix: release runs in finally -> refCnt drops to 0.
+ assertEquals(corruptBuf.refCnt(), 0, "entry's ByteBuf leaked when parseMessageMetadata threw");
+ }
+}
(pulsar) 02/04: [fix][broker] Fix TableViewLoadDataStoreImpl close deadlock that stalls broker shutdown (#26243)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bc69930c401c784c90e14a84e199ea0b8ffccc55
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Jul 25 22:54:39 2026 +0300
[fix][broker] Fix TableViewLoadDataStoreImpl close deadlock that stalls broker shutdown (#26243)
(cherry picked from commit aa3ec21b10fd34f681b7c259e0070828afdc665c)
---
.../store/TableViewLoadDataStoreImpl.java | 21 ++++++++-
.../extensions/store/LoadDataStoreTest.java | 53 ++++++++++++++++++++++
2 files changed, 72 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/store/TableViewLoadDataStoreImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/store/TableViewLoadDataStoreImpl.java
index 3ce44a1e65a7..21b687337c9a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/store/TableViewLoadDataStoreImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/store/TableViewLoadDataStoreImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.loadbalance.extensions.store;
+import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.util.Map;
import java.util.Optional;
@@ -145,7 +146,14 @@ public class TableViewLoadDataStoreImpl<T> implements LoadDataStore<T> {
public synchronized void closeTableView() throws IOException {
validateState();
if (tableView != null) {
- tableView.close();
+ // Close asynchronously without waiting: the close future may only be able to complete on the
+ // client's internal executor thread, and that thread can be blocked on this store's monitor
+ // (e.g. a ServiceUnitStateChannel StateChangeListener calling pushAsync/removeAsync), so a
+ // blocking close while holding the monitor could deadlock.
+ tableView.closeAsync().exceptionally(e -> {
+ log.warn("Failed to close table view on {}", topic, e);
+ return null;
+ });
tableView = null;
}
}
@@ -160,7 +168,11 @@ public class TableViewLoadDataStoreImpl<T> implements LoadDataStore<T> {
private synchronized void closeProducer() throws IOException {
validateState();
if (producer != null) {
- producer.close();
+ // Close asynchronously without waiting; see closeTableView() for the deadlock rationale.
+ producer.closeAsync().exceptionally(e -> {
+ log.warn("Failed to close producer on {}", topic, e);
+ return null;
+ });
producer = null;
}
}
@@ -210,6 +222,11 @@ public class TableViewLoadDataStoreImpl<T> implements LoadDataStore<T> {
isShutdown = true;
}
+ @VisibleForTesting
+ synchronized void setTableView(TableView<T> tableView) {
+ this.tableView = tableView;
+ }
+
private String validateProducer() {
if (isShutdown) {
return SHUTDOWN_ERR_MSG;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/store/LoadDataStoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/store/LoadDataStoreTest.java
index fe03446e58b7..d456369b1e76 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/store/LoadDataStoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/store/LoadDataStoreTest.java
@@ -18,9 +18,12 @@
*/
package org.apache.pulsar.broker.loadbalance.extensions.store;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertThrows;
@@ -29,12 +32,15 @@ import com.google.common.collect.Sets;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import lombok.AllArgsConstructor;
import lombok.Cleanup;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.client.api.TableView;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
@@ -187,6 +193,53 @@ public class LoadDataStoreTest extends MockedPulsarServiceBaseTest {
loadDataStore.removeAsync("2").get();
}
+ @Test(timeOut = 30_000)
+ @SuppressWarnings("unchecked")
+ public void testShutdownDoesNotDeadlockWithConcurrentStoreAccess() throws Exception {
+ String topic = TopicDomain.persistent + "://" + NamespaceName.SYSTEM_NAMESPACE + "/" + UUID.randomUUID();
+ var loadDataStore =
+ (TableViewLoadDataStoreImpl<Integer>) LoadDataStoreFactory.create(pulsar, topic, Integer.class);
+ loadDataStore.start();
+ loadDataStore.closeTableView();
+
+ // Replace the table view with a stub whose close future completes only after another thread has
+ // called a synchronized method of the store. This mirrors the production interleaving where the
+ // reader close future completes on the client's internal executor thread while a channel
+ // StateChangeListener on that same thread is blocked in removeAsync() waiting for the store
+ // monitor held by the closing thread.
+ CompletableFuture<Void> closeSignal = new CompletableFuture<>();
+ CountDownLatch closeStarted = new CountDownLatch(1);
+ TableView<Integer> stubTableView = mock(TableView.class);
+ when(stubTableView.closeAsync()).thenAnswer(invocation -> {
+ closeStarted.countDown();
+ return closeSignal;
+ });
+ doAnswer(invocation -> {
+ closeStarted.countDown();
+ closeSignal.get();
+ return null;
+ }).when(stubTableView).close();
+ loadDataStore.setTableView(stubTableView);
+
+ Thread concurrentAccess = new Thread(() -> {
+ try {
+ closeStarted.await();
+ loadDataStore.removeAsync("key");
+ closeSignal.complete(null);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ concurrentAccess.start();
+ try {
+ loadDataStore.shutdown();
+ } finally {
+ concurrentAccess.join(5000);
+ }
+ assertFalse(concurrentAccess.isAlive());
+ assertTrue(closeSignal.isDone());
+ }
+
@Test
public void testShutdown() throws Exception {
String topic = TopicDomain.persistent + "://" + NamespaceName.SYSTEM_NAMESPACE + "/" + UUID.randomUUID();
(pulsar) 03/04: [fix][broker] Prevent stale service unit callbacks from dropping active lookup and cleanup jobs (#26146)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 338b922d8b18f33d416ac963220c823e3dbe19d6
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Sun Jul 26 04:20:27 2026 +0800
[fix][broker] Prevent stale service unit callbacks from dropping active lookup and cleanup jobs (#26146)
(cherry picked from commit 32146ae350ca0526b04c3d089e2e4d2cde648688)
---
.../channel/ServiceUnitStateChannelImpl.java | 37 +++--
.../channel/ServiceUnitStateChannelTest.java | 155 ++++++++++++++++++++-
2 files changed, 180 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index c462edc63460..976525bc7231 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -231,6 +231,16 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
this.channelState = Constructed;
}
+ @VisibleForTesting
+ Map<String, CompletableFuture<String>> getOwnerRequests() {
+ return getOwnerRequests;
+ }
+
+ @VisibleForTesting
+ Map<String, CompletableFuture<Void>> getCleanupJobs() {
+ return cleanupJobs;
+ }
+
@Override
public void scheduleOwnershipMonitor() {
if (monitorTask == null) {
@@ -845,13 +855,15 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
}
- private void handleSkippedEvent(String serviceUnit) {
+ @VisibleForTesting
+ void handleSkippedEvent(String serviceUnit) {
var getOwnerRequest = getOwnerRequests.get(serviceUnit);
if (getOwnerRequest != null) {
var data = tableview.get(serviceUnit);
if (data != null && data.state() == Owned) {
getOwnerRequest.complete(data.dstBroker());
- getOwnerRequests.remove(serviceUnit);
+ // Completing the request can run callbacks that install a newer request for the same service unit.
+ getOwnerRequests.remove(serviceUnit, getOwnerRequest);
stateChangeListeners.notify(serviceUnit, data, null);
}
}
@@ -1014,7 +1026,8 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
return future;
}
- private CompletableFuture<String> dedupeGetOwnerRequest(String serviceUnit) {
+ @VisibleForTesting
+ CompletableFuture<String> dedupeGetOwnerRequest(String serviceUnit) {
var requested = new MutableObject<CompletableFuture<String>>();
try {
@@ -1047,7 +1060,8 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
var future = requested.getValue();
if (future != null) {
future.whenComplete((__, e) -> {
- getOwnerRequests.remove(serviceUnit);
+ // The request may have been replaced by a later lookup before this callback runs.
+ getOwnerRequests.remove(serviceUnit, future);
if (e != null) {
log.warn("{} failed to getOwner for serviceUnit:{}", brokerId, serviceUnit, e);
}
@@ -1324,14 +1338,15 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
private void handleBrokerCreationEvent(String broker) {
- if (!cleanupJobs.isEmpty() && cleanupJobs.containsKey(broker)) {
+ CompletableFuture<Void> cleanupJob = cleanupJobs.get(broker);
+ if (cleanupJob != null) {
healthCheckBrokerAsync(broker)
.orTimeout(MAX_BROKER_HEALTH_CHECK_DELAY_IN_MILLIS * (MAX_BROKER_HEALTH_CHECK_RETRY + 1)
, MILLISECONDS)
.thenAccept(__ -> {
- CompletableFuture<Void> future = cleanupJobs.remove(broker);
- if (future != null) {
- future.cancel(false);
+ // The health check is async; only cancel the cleanup job observed before the check.
+ if (cleanupJobs.remove(broker, cleanupJob)) {
+ cleanupJob.cancel(false);
totalInactiveBrokerCleanupCancelledCnt++;
log.info("Successfully cancelled the ownership cleanup for broker:{}."
+ " Active cleanup job count:{}",
@@ -1392,7 +1407,8 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
return false;
}
- private void scheduleCleanup(String broker, long delayInSecs) {
+ @VisibleForTesting
+ void scheduleCleanup(String broker, long delayInSecs) {
var scheduled = new MutableObject<CompletableFuture<Void>>();
try {
if (channelDisabled()) {
@@ -1420,7 +1436,8 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
var future = scheduled.getValue();
if (future != null) {
future.whenComplete((v, ex) -> {
- cleanupJobs.remove(broker);
+ // The job may have been replaced by a later cleanup schedule before this callback runs.
+ cleanupJobs.remove(broker, future);
});
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
index 108c4a391065..c171557f4fc0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
@@ -280,6 +280,104 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
}
}
+ @Test(priority = 1)
+ public void testCompletedGetOwnerRequestDoesNotRemoveNewRequest() {
+ ServiceUnitStateChannelImpl channel = (ServiceUnitStateChannelImpl) channel1;
+ String serviceUnit = namespaceName + "/0x10000000_0x10000001";
+ var getOwnerRequests = channel.getOwnerRequests();
+ getOwnerRequests.remove(serviceUnit);
+ CompletableFuture<String> oldRequest = channel.dedupeGetOwnerRequest(serviceUnit);
+ assertEquals(getOwnerRequests.get(serviceUnit), oldRequest);
+
+ CompletableFuture<String> newRequest = null;
+ try {
+ // State-event handlers remove the current request before completing it, allowing a later lookup
+ // to install a new request generation before the old request's completion cleanup runs.
+ assertTrue(getOwnerRequests.remove(serviceUnit, oldRequest));
+ newRequest = channel.dedupeGetOwnerRequest(serviceUnit);
+ assertTrue(newRequest != oldRequest);
+ assertTrue(getOwnerRequests.get(serviceUnit) == newRequest);
+
+ // The previous unconditional removal would remove newRequest here.
+ assertTrue(oldRequest.complete(brokerId1));
+
+ assertTrue(getOwnerRequests.get(serviceUnit) == newRequest,
+ "A stale get-owner cleanup must not remove a newer request future");
+
+ assertTrue(newRequest.complete(brokerId2));
+ assertFalse(getOwnerRequests.containsKey(serviceUnit),
+ "The newer request must remove itself after completion");
+ } finally {
+ getOwnerRequests.remove(serviceUnit);
+ oldRequest.cancel(false);
+ if (newRequest != null) {
+ newRequest.cancel(false);
+ }
+ }
+ }
+
+ @Test(priority = 1)
+ public void testSkippedEventDoesNotRemoveNewGetOwnerRequest() throws Exception {
+ ServiceUnitStateChannelImpl channel = (ServiceUnitStateChannelImpl) channel1;
+ String serviceUnit = namespaceName + "/0x10000002_0x10000003";
+ var getOwnerRequests = channel.getOwnerRequests();
+ CompletableFuture<String> oldRequest = new CompletableFuture<>();
+ CompletableFuture<String> newRequest = new CompletableFuture<>();
+ try {
+ overrideTableView(channel, serviceUnit, new ServiceUnitStateData(Owned, brokerId1, 1));
+ getOwnerRequests.put(serviceUnit, oldRequest);
+ oldRequest.whenComplete((__, ___) -> getOwnerRequests.put(serviceUnit, newRequest));
+
+ channel.handleSkippedEvent(serviceUnit);
+
+ assertEquals(oldRequest.getNow(null), brokerId1);
+ assertTrue(getOwnerRequests.get(serviceUnit) == newRequest,
+ "A stale skipped-event cleanup must not remove a newer request future");
+ } finally {
+ getOwnerRequests.remove(serviceUnit);
+ oldRequest.cancel(false);
+ overrideTableView(channel, serviceUnit, null);
+ }
+ }
+
+ @Test(priority = 1)
+ public void testCompletedCleanupJobDoesNotRemoveNewCleanupJob() {
+ ServiceUnitStateChannelImpl channel = (ServiceUnitStateChannelImpl) channel1;
+ String broker = brokerId3;
+ var cleanupJobs = channel.getCleanupJobs();
+ cleanupJobs.remove(broker);
+ channel.scheduleCleanup(broker, 60L);
+ CompletableFuture<Void> oldJob = cleanupJobs.get(broker);
+ assertNotNull(oldJob);
+
+ CompletableFuture<Void> newJob = null;
+ try {
+ // Broker-creation handling removes a cleanup job before cancelling it. A later broker-deletion
+ // event can therefore schedule a new job before the old job's completion cleanup runs.
+ assertTrue(cleanupJobs.remove(broker, oldJob));
+ channel.scheduleCleanup(broker, 60L);
+ newJob = cleanupJobs.get(broker);
+ assertNotNull(newJob);
+ assertTrue(newJob != oldJob);
+
+ // The previous unconditional removal would remove newJob here.
+ assertTrue(oldJob.cancel(false));
+
+ assertTrue(cleanupJobs.get(broker) == newJob,
+ "A stale cleanup job completion must not remove a newer cleanup job future");
+
+ assertTrue(newJob.cancel(false));
+ assertFalse(cleanupJobs.containsKey(broker),
+ "The newer cleanup job must remove itself after completion");
+ } finally {
+ cleanupJobs.remove(broker);
+ oldJob.cancel(false);
+ if (newJob != null) {
+ newJob.cancel(false);
+ }
+ }
+ }
+
@Test(priority = 100)
public void channelValidationTest()
throws ExecutionException, InterruptedException, IllegalAccessException, PulsarServerException,
@@ -822,6 +920,59 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
}
+ @Test(priority = 8)
+ public void handleBrokerCreationEventDoesNotCancelNewCleanupJobTest() {
+ ServiceUnitStateChannelImpl channel = (ServiceUnitStateChannelImpl) channel1;
+ var cleanupJobs = channel.getCleanupJobs();
+ String broker = brokerId2;
+ CompletableFuture<Void> healthCheck = new CompletableFuture<>();
+ cleanupJobs.remove(broker);
+ channel.scheduleCleanup(broker, 60L);
+ CompletableFuture<Void> oldJob = cleanupJobs.get(broker);
+ assertNotNull(oldJob);
+
+ reset(brokers);
+ doReturn(healthCheck).when(brokers).healthcheckAsync(any());
+ doReturn(brokers).when(pulsarAdmin).brokers();
+ CompletableFuture<Void> newJob = null;
+ try {
+ channel.handleBrokerRegistrationEvent(broker, NotificationType.Created);
+ verify(brokers, times(1)).healthcheckAsync(any());
+
+ // The old cleanup can finish while the asynchronous health check is still pending. A later
+ // broker-deletion event can then schedule a new cleanup job for the same broker.
+ assertTrue(oldJob.complete(null));
+ assertFalse(cleanupJobs.containsKey(broker));
+ channel.scheduleCleanup(broker, 60L);
+ newJob = cleanupJobs.get(broker);
+ assertNotNull(newJob);
+ assertTrue(newJob != oldJob);
+
+ healthCheck.complete(null);
+
+ CompletableFuture<Void> expectedNewJob = newJob;
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertTrue(cleanupJobs.get(broker) == expectedNewJob,
+ "A stale broker-creation callback must not remove a newer cleanup job");
+ assertFalse(expectedNewJob.isCancelled());
+ });
+
+ assertTrue(newJob.cancel(false));
+ assertFalse(cleanupJobs.containsKey(broker),
+ "The newer cleanup job must remove itself after cancellation");
+ } finally {
+ cleanupJobs.remove(broker);
+ oldJob.cancel(false);
+ if (newJob != null) {
+ newJob.cancel(false);
+ }
+ reset(brokers);
+ doReturn(CompletableFuture.failedFuture(new RuntimeException("failed"))).when(brokers)
+ .healthcheckAsync(any());
+ reset(pulsarAdmin);
+ }
+ }
+
@Test(priority = 9)
public void handleBrokerDeletionEventTest() throws Exception {
@@ -2075,9 +2226,9 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
}
}
- private static ConcurrentHashMap<String, CompletableFuture<Optional<String>>> getOwnerRequests(
+ private static ConcurrentHashMap<String, CompletableFuture<String>> getOwnerRequests(
ServiceUnitStateChannel channel) throws IllegalAccessException {
- return (ConcurrentHashMap<String, CompletableFuture<Optional<String>>>)
+ return (ConcurrentHashMap<String, CompletableFuture<String>>)
FieldUtils.readDeclaredField(channel,
"getOwnerRequests", true);
}