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 2024/11/23 19:41:34 UTC
(pulsar) branch branch-3.0 updated (bb909e60e9c -> 489628d4853)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from bb909e60e9c Revert "[improve][offload] Use filesystemURI as the storage path (#23591)"
new e2bd42311f7 [fix][broker] Continue closing even when executor is shut down (#22599)
new 7429cb47ca3 [improve][broker] Close TopicPoliciesService to allow Pulsar broker graceful shutdown (#22589)
new 489628d4853 [improve][broker] Don't use forkjoin pool by default for deleting partitioned topics (#22598)
The 3 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:
.../broker/resources/NamespaceResources.java | 15 ++++--
.../pulsar/broker/resources/PulsarResources.java | 12 ++++-
.../org/apache/pulsar/broker/PulsarService.java | 7 ++-
.../SystemTopicBasedTopicPoliciesService.java | 63 +++++++++++++++++++---
.../SystemTopicTxnBufferSnapshotService.java | 20 ++++++-
.../broker/service/TopicPoliciesService.java | 7 ++-
.../broker/service/persistent/PersistentTopic.java | 10 +++-
7 files changed, 119 insertions(+), 15 deletions(-)
(pulsar) 03/03: [improve][broker] Don't use forkjoin pool by default for deleting partitioned topics (#22598)
Posted by lh...@apache.org.
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 489628d4853971542343ac48f24ad8f216444e1f
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Apr 26 23:17:51 2024 +0300
[improve][broker] Don't use forkjoin pool by default for deleting partitioned topics (#22598)
(cherry picked from commit 8323a3c49912976aee723787fa67bee4d7d8d846)
---
.../pulsar/broker/resources/NamespaceResources.java | 15 ++++++++++++---
.../apache/pulsar/broker/resources/PulsarResources.java | 12 ++++++++++--
.../main/java/org/apache/pulsar/broker/PulsarService.java | 2 +-
3 files changed, 23 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
index e01e31b96dc..9c1539bbb8e 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
@@ -24,6 +24,8 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -56,10 +58,15 @@ public class NamespaceResources extends BaseResources<Policies> {
private static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data";
public NamespaceResources(MetadataStore localStore, MetadataStore configurationStore, int operationTimeoutSec) {
+ this(localStore, configurationStore, operationTimeoutSec, ForkJoinPool.commonPool());
+ }
+
+ public NamespaceResources(MetadataStore localStore, MetadataStore configurationStore, int operationTimeoutSec,
+ Executor executor) {
super(configurationStore, Policies.class, operationTimeoutSec);
this.configurationStore = configurationStore;
isolationPolicies = new IsolationPolicyResources(configurationStore, operationTimeoutSec);
- partitionedTopicResources = new PartitionedTopicResources(configurationStore, operationTimeoutSec);
+ partitionedTopicResources = new PartitionedTopicResources(configurationStore, operationTimeoutSec, executor);
this.localStore = localStore;
}
@@ -245,9 +252,11 @@ public class NamespaceResources extends BaseResources<Policies> {
public static class PartitionedTopicResources extends BaseResources<PartitionedTopicMetadata> {
private static final String PARTITIONED_TOPIC_PATH = "/admin/partitioned-topics";
+ private final Executor executor;
- public PartitionedTopicResources(MetadataStore configurationStore, int operationTimeoutSec) {
+ public PartitionedTopicResources(MetadataStore configurationStore, int operationTimeoutSec, Executor executor) {
super(configurationStore, PartitionedTopicMetadata.class, operationTimeoutSec);
+ this.executor = executor;
}
public CompletableFuture<Void> updatePartitionedTopicAsync(TopicName tn, Function<PartitionedTopicMetadata,
@@ -385,7 +394,7 @@ public class NamespaceResources extends BaseResources<Policies> {
future.complete(deleteResult);
}
});
- });
+ }, executor);
return future;
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
index a3c5633a6db..1f973a61a52 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
@@ -19,6 +19,8 @@
package org.apache.pulsar.broker.resources;
import java.util.Optional;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
import lombok.Getter;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
@@ -55,13 +57,19 @@ public class PulsarResources {
public PulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore) {
this(localMetadataStore, configurationMetadataStore, DEFAULT_OPERATION_TIMEOUT_SEC);
}
+
+ public PulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore,
+ int operationTimeoutSec) {
+ this(localMetadataStore, configurationMetadataStore, operationTimeoutSec, ForkJoinPool.commonPool());
+ }
+
public PulsarResources(MetadataStore localMetadataStore, MetadataStore configurationMetadataStore,
- int operationTimeoutSec) {
+ int operationTimeoutSec, Executor executor) {
if (configurationMetadataStore != null) {
tenantResources = new TenantResources(configurationMetadataStore, operationTimeoutSec);
clusterResources = new ClusterResources(configurationMetadataStore, operationTimeoutSec);
namespaceResources = new NamespaceResources(localMetadataStore, configurationMetadataStore
- , operationTimeoutSec);
+ , operationTimeoutSec, executor);
resourcegroupResources = new ResourceGroupResources(configurationMetadataStore, operationTimeoutSec);
} else {
tenantResources = null;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 096ae985caa..4da5e114a08 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -1011,7 +1011,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
@VisibleForTesting
protected PulsarResources newPulsarResources() {
PulsarResources pulsarResources = new PulsarResources(localMetadataStore, configurationMetadataStore,
- config.getMetadataStoreOperationTimeoutSeconds());
+ config.getMetadataStoreOperationTimeoutSeconds(), getExecutor());
pulsarResources.getClusterResources().getStore().registerListener(this::handleDeleteCluster);
return pulsarResources;
(pulsar) 01/03: [fix][broker] Continue closing even when executor is shut down (#22599)
Posted by lh...@apache.org.
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 e2bd42311f7d3a5f895f63426fc3538aa6c54e05
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Apr 26 23:16:54 2024 +0300
[fix][broker] Continue closing even when executor is shut down (#22599)
(cherry picked from commit f8f256cfbdcd780c81442dc5566b6ed071141645)
---
.../pulsar/broker/service/persistent/PersistentTopic.java | 10 +++++++++-
1 file changed, 9 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 337fad5e6b6..4828857878b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -45,6 +45,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -1452,7 +1453,14 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
FutureUtil.waitForAll(futures).thenRunAsync(() -> {
closeClientFuture.complete(null);
- }, getOrderedExecutor()).exceptionally(ex -> {
+ }, command -> {
+ try {
+ getOrderedExecutor().execute(command);
+ } catch (RejectedExecutionException e) {
+ // executor has been shut down, execute in current thread
+ command.run();
+ }
+ }).exceptionally(ex -> {
log.error("[{}] Error closing clients", topic, ex);
alreadyUnFenced.set(true);
unfenceTopicToResume();
(pulsar) 02/03: [improve][broker] Close TopicPoliciesService to allow Pulsar broker graceful shutdown (#22589)
Posted by lh...@apache.org.
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 7429cb47ca31678dcd95fe8786653c8a18ed14c3
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Apr 26 23:17:18 2024 +0300
[improve][broker] Close TopicPoliciesService to allow Pulsar broker graceful shutdown (#22589)
(cherry picked from commit 7a44c801f86c4276533b0f008e768fb8deba4abc)
---
.../org/apache/pulsar/broker/PulsarService.java | 5 ++
.../SystemTopicBasedTopicPoliciesService.java | 63 +++++++++++++++++++---
.../SystemTopicTxnBufferSnapshotService.java | 20 ++++++-
.../broker/service/TopicPoliciesService.java | 7 ++-
4 files changed, 87 insertions(+), 8 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 689a861ffd7..096ae985caa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -538,6 +538,11 @@ public class PulsarService implements AutoCloseable, ShutdownService {
transactionBufferClient.close();
}
+ if (topicPoliciesService != null) {
+ topicPoliciesService.close();
+ topicPoliciesService = null;
+ }
+
if (client != null) {
client.close();
client = null;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index c5fd889881b..8882732be02 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -32,6 +32,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nonnull;
import org.apache.commons.lang3.concurrent.ConcurrentInitializer;
@@ -72,6 +73,7 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
private final PulsarService pulsarService;
private final HashSet localCluster;
private final String clusterName;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
private final ConcurrentInitializer<NamespaceEventsSystemTopicFactory>
namespaceEventsSystemTopicFactoryLazyInitializer = new LazyInitializer<>() {
@@ -110,12 +112,18 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
this.writerCaches = Caffeine.newBuilder()
.expireAfterAccess(5, TimeUnit.MINUTES)
.removalListener((namespaceName, writer, cause) -> {
- ((SystemTopicClient.Writer) writer).closeAsync().exceptionally(ex -> {
- log.error("[{}] Close writer error.", namespaceName, ex);
- return null;
- });
+ try {
+ ((SystemTopicClient.Writer) writer).close();
+ } catch (Exception e) {
+ log.error("[{}] Close writer error.", namespaceName, e);
+ }
})
+ .executor(pulsarService.getExecutor())
.buildAsync((namespaceName, executor) -> {
+ if (closed.get()) {
+ return CompletableFuture.failedFuture(
+ new BrokerServiceException(getClass().getName() + " is closed."));
+ }
SystemTopicClient<PulsarEvent> systemTopicClient = getNamespaceEventsSystemTopicFactory()
.createTopicPoliciesSystemTopicClient(namespaceName);
return systemTopicClient.newWriterAsync();
@@ -391,6 +399,10 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
protected CompletableFuture<SystemTopicClient.Reader<PulsarEvent>> createSystemTopicClient(
NamespaceName namespace) {
+ if (closed.get()) {
+ return CompletableFuture.failedFuture(
+ new BrokerServiceException(getClass().getName() + " is closed."));
+ }
try {
createSystemTopicFactoryIfNeeded();
} catch (PulsarServerException ex) {
@@ -439,6 +451,11 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
private void initPolicesCache(SystemTopicClient.Reader<PulsarEvent> reader, CompletableFuture<Void> future) {
+ if (closed.get()) {
+ future.completeExceptionally(new BrokerServiceException(getClass().getName() + " is closed."));
+ cleanCacheAndCloseReader(reader.getSystemTopic().getTopicName().getNamespaceObject(), false);
+ return;
+ }
reader.hasMoreEventsAsync().whenComplete((hasMore, ex) -> {
if (ex != null) {
log.error("[{}] Failed to check the move events for the system topic",
@@ -520,6 +537,10 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
* #{@link SystemTopicBasedTopicPoliciesService#getTopicPoliciesAsync(TopicName)} method to block loading topic.
*/
private void readMorePoliciesAsync(SystemTopicClient.Reader<PulsarEvent> reader) {
+ if (closed.get()) {
+ cleanCacheAndCloseReader(reader.getSystemTopic().getTopicName().getNamespaceObject(), false);
+ return;
+ }
reader.readNextAsync()
.thenAccept(msg -> {
refreshTopicPoliciesCache(msg);
@@ -638,11 +659,20 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
private void fetchTopicPoliciesAsyncAndCloseReader(SystemTopicClient.Reader<PulsarEvent> reader,
TopicName topicName, TopicPolicies policies,
CompletableFuture<TopicPolicies> future) {
+ if (closed.get()) {
+ future.completeExceptionally(new BrokerServiceException(getClass().getName() + " is closed."));
+ reader.closeAsync().whenComplete((v, e) -> {
+ if (e != null) {
+ log.error("[{}] Close reader error.", topicName, e);
+ }
+ });
+ return;
+ }
reader.hasMoreEventsAsync().whenComplete((hasMore, ex) -> {
if (ex != null) {
future.completeExceptionally(ex);
}
- if (hasMore) {
+ if (hasMore != null && hasMore) {
reader.readNextAsync().whenComplete((msg, e) -> {
if (e != null) {
future.completeExceptionally(e);
@@ -666,7 +696,9 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
});
} else {
- future.complete(policies);
+ if (!future.isDone()) {
+ future.complete(policies);
+ }
reader.closeAsync().whenComplete((v, e) -> {
if (e != null) {
log.error("[{}] Close reader error.", topicName, e);
@@ -750,4 +782,23 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
private static final Logger log = LoggerFactory.getLogger(SystemTopicBasedTopicPoliciesService.class);
+
+ @Override
+ public void close() throws Exception {
+ if (closed.compareAndSet(false, true)) {
+ writerCaches.synchronous().invalidateAll();
+ readerCaches.values().forEach(future -> {
+ if (future != null && !future.isCompletedExceptionally()) {
+ future.thenAccept(reader -> {
+ try {
+ reader.close();
+ } catch (Exception e) {
+ log.error("Failed to close reader.", e);
+ }
+ });
+ }
+ });
+ readerCaches.clear();
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java
index c7fbd14220a..ba6cbee3557 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java
@@ -151,8 +151,26 @@ public class SystemTopicTxnBufferSnapshotService<T> {
public void close() throws Exception {
for (Map.Entry<NamespaceName, SystemTopicClient<T>> entry : clients.entrySet()) {
- entry.getValue().close();
+ try {
+ entry.getValue().close();
+ } catch (Exception e) {
+ log.error("Failed to close system topic client for namespace {}", entry.getKey(), e);
+ }
+ }
+ clients.clear();
+ for (Map.Entry<NamespaceName, ReferenceCountedWriter<T>> entry : refCountedWriterMap.entrySet()) {
+ CompletableFuture<SystemTopicClient.Writer<T>> future = entry.getValue().getFuture();
+ if (!future.isCompletedExceptionally()) {
+ future.thenAccept(writer -> {
+ try {
+ writer.close();
+ } catch (Exception e) {
+ log.error("Failed to close writer for namespace {}", entry.getKey(), e);
+ }
+ });
+ }
}
+ refCountedWriterMap.clear();
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java
index cfac6b396e8..3b6d468a5ec 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPoliciesService.java
@@ -37,7 +37,7 @@ import org.apache.pulsar.common.util.FutureUtil;
* Topic policies service.
*/
@InterfaceStability.Evolving
-public interface TopicPoliciesService {
+public interface TopicPoliciesService extends AutoCloseable {
TopicPoliciesService DISABLED = new TopicPoliciesServiceDisabled();
long DEFAULT_GET_TOPIC_POLICY_TIMEOUT = 30_000;
@@ -238,5 +238,10 @@ public interface TopicPoliciesService {
public void unregisterListener(TopicName topicName, TopicPolicyListener<TopicPolicies> listener) {
//No-op
}
+
+ @Override
+ public void close() {
+ //No-op
+ }
}
}