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/25 19:54:45 UTC
(pulsar) branch master updated: [fix][broker] Fix TableViewLoadDataStoreImpl close deadlock that stalls broker shutdown (#26243)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new aa3ec21b10fd [fix][broker] Fix TableViewLoadDataStoreImpl close deadlock that stalls broker shutdown (#26243)
aa3ec21b10fd is described below
commit aa3ec21b10fd34f681b7c259e0070828afdc665c
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)
---
.../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 1ebfb0bfae5d..bbfd32d3fa21 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().attr("topic", topic).exception(e).log("Failed to close table view");
+ 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().attr("topic", topic).exception(e).log("Failed to close producer");
+ return null;
+ });
producer = null;
}
}
@@ -211,6 +223,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 dd6e635ce33f..ec34e4f9d57d 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;
@@ -188,6 +194,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();