You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/02/08 06:53:13 UTC
[pulsar] branch branch-2.11 updated: [improve][broker] Add UncaughtExceptionHandler for every thread pool (#18211)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new bfaf39c4dd0 [improve][broker] Add UncaughtExceptionHandler for every thread pool (#18211)
bfaf39c4dd0 is described below
commit bfaf39c4dd0ac495edab3c948b372305c2567762
Author: feynmanlin <31...@qq.com>
AuthorDate: Thu Oct 27 11:29:53 2022 +0800
[improve][broker] Add UncaughtExceptionHandler for every thread pool (#18211)
---
.../org/apache/pulsar/broker/PulsarService.java | 11 ++++----
.../broker/TransactionMetadataStoreService.java | 4 +--
.../loadbalance/impl/ModularLoadManagerImpl.java | 7 +++--
.../loadbalance/impl/SimpleLoadManagerImpl.java | 4 +--
.../pulsar/broker/service/BrokerService.java | 33 ++++++++++++----------
.../metrics/PrometheusMetricsProvider.java | 5 ++--
.../pulsar/broker/tools/LoadReportCommand.java | 4 ++-
.../pulsar/client/impl/AutoClusterFailover.java | 4 +--
.../client/impl/ControlledClusterFailover.java | 4 +--
.../pulsar/client/impl/PulsarClientImpl.java | 5 ++--
.../pulsar/client/util/ExecutorProvider.java | 8 ++++--
.../pulsar/functions/instance/InstanceCache.java | 4 +--
.../worker/ClusterServiceCoordinator.java | 13 +++++++++
13 files changed, 65 insertions(+), 41 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 6cb761d7761..2cb48ed1549 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
@@ -314,13 +314,13 @@ public class PulsarService implements AutoCloseable, ShutdownService {
this.config = config;
this.processTerminator = processTerminator;
this.loadManagerExecutor = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-load-manager"));
+ .newSingleThreadScheduledExecutor(new ExecutorProvider.ExtendedThreadFactory("pulsar-load-manager"));
this.workerConfig = workerConfig;
this.functionWorkerService = functionWorkerService;
this.executor = Executors.newScheduledThreadPool(config.getNumExecutorThreadPoolSize(),
- new DefaultThreadFactory("pulsar"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar"));
this.cacheExecutor = Executors.newScheduledThreadPool(config.getNumCacheExecutorThreadPoolSize(),
- new DefaultThreadFactory("zk-cache-callback"));
+ new ExecutorProvider.ExtendedThreadFactory("zk-cache-callback"));
if (config.isTransactionCoordinatorEnabled()) {
this.transactionExecutorProvider = new ExecutorProvider(this.getConfiguration()
@@ -613,7 +613,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ new ExecutorProvider.ExtendedThreadFactory(getClass().getSimpleName() + "-shutdown"));
FutureUtil.addTimeoutHandling(future,
Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
@@ -1423,7 +1423,8 @@ public class PulsarService implements AutoCloseable, ShutdownService {
protected synchronized ScheduledExecutorService getCompactorExecutor() {
if (this.compactorExecutor == null) {
- compactorExecutor = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("compaction"));
+ compactorExecutor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("compaction"));
}
return this.compactorExecutor;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
index a15039b3ff8..fb56e8e6857 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
@@ -24,7 +24,6 @@ import static org.apache.pulsar.transaction.coordinator.proto.TxnStatus.COMMITTI
import com.google.common.annotations.VisibleForTesting;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.Collections;
import java.util.Deque;
import java.util.List;
@@ -52,6 +51,7 @@ import org.apache.pulsar.client.api.transaction.TransactionBufferClient;
import org.apache.pulsar.client.api.transaction.TransactionBufferClientException.ReachMaxPendingOpsException;
import org.apache.pulsar.client.api.transaction.TransactionBufferClientException.RequestTimeoutException;
import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.api.proto.TxnAction;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.util.FutureUtil;
@@ -93,7 +93,7 @@ public class TransactionMetadataStoreService {
private static final long HANDLE_PENDING_CONNECT_TIME_OUT = 30000L;
private final ThreadFactory threadFactory =
- new DefaultThreadFactory("transaction-coordinator-thread-factory");
+ new ExecutorProvider.ExtendedThreadFactory("transaction-coordinator-thread-factory");
public TransactionMetadataStoreService(TransactionMetadataStoreProvider transactionMetadataStoreProvider,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index 9444737fb79..3b0834027df 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -22,7 +22,6 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.ConcurrentModificationException;
@@ -62,6 +61,7 @@ import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.BrokerTopicLo
import org.apache.pulsar.broker.resources.ClusterResources;
import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
@@ -213,8 +213,9 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
loadData = new LoadData();
loadSheddingPipeline = new ArrayList<>();
preallocatedBundleToBroker = new ConcurrentHashMap<>();
- scheduler = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-modular-load-manager"));
- this.brokerToFailureDomainMap = Maps.newHashMap();
+ scheduler = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-modular-load-manager"));
+ this.brokerToFailureDomainMap = new HashMap<>();
this.brokerTopicLoadingPredicate = new BrokerTopicLoadingPredicate() {
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
index 06c87cba627..c21d939e017 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
@@ -28,7 +28,6 @@ import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.collect.TreeMultimap;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.HashMap;
@@ -55,6 +54,7 @@ import org.apache.pulsar.broker.loadbalance.LoadManager;
import org.apache.pulsar.broker.loadbalance.PlacementStrategy;
import org.apache.pulsar.broker.loadbalance.ResourceUnit;
import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.BrokerTopicLoadingPredicate;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
import org.apache.pulsar.common.policies.data.ResourceQuota;
@@ -189,7 +189,7 @@ public class SimpleLoadManagerImpl implements LoadManager, Consumer<Notification
// Perform initializations which may be done without a PulsarService.
public SimpleLoadManagerImpl() {
scheduler = Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("pulsar-simple-load-manager"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-simple-load-manager"));
this.sortedRankings.set(new TreeMap<>());
this.currentLoadReports = new HashMap<>();
this.resourceUnitRankings = new HashMap<>();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 4151887be3f..6dc78736c9e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -130,6 +130,7 @@ import org.apache.pulsar.client.api.SizeUnit;
import org.apache.pulsar.client.impl.ClientBuilderImpl;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.internal.PropertiesUtils;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.configuration.BindAddress;
import org.apache.pulsar.common.configuration.FieldContext;
@@ -313,13 +314,14 @@ public class BrokerService implements Closeable {
this.topicOrderedExecutor = OrderedExecutor.newBuilder()
.numThreads(pulsar.getConfiguration().getNumWorkerThreadsForNonPersistentTopic())
.name("broker-topic-workers").build();
- final DefaultThreadFactory acceptorThreadFactory = new DefaultThreadFactory("pulsar-acceptor");
+ final DefaultThreadFactory acceptorThreadFactory =
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-acceptor");
this.acceptorGroup = EventLoopUtil.newEventLoopGroup(
pulsar.getConfiguration().getNumAcceptorThreads(), false, acceptorThreadFactory);
this.workerGroup = eventLoopGroup;
- this.statsUpdater = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-stats-updater"));
+ this.statsUpdater = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-stats-updater"));
this.authorizationService = new AuthorizationService(
pulsar.getConfiguration(), pulsar().getPulsarResources());
if (!pulsar.getConfiguration().getEntryFilterNames().isEmpty()) {
@@ -329,22 +331,22 @@ public class BrokerService implements Closeable {
pulsar.getLocalMetadataStore().registerListener(this::handleMetadataChanges);
pulsar.getConfigurationMetadataStore().registerListener(this::handleMetadataChanges);
- this.inactivityMonitor = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-inactivity-monitor"));
- this.messageExpiryMonitor = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-msg-expiry-monitor"));
+ this.inactivityMonitor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-inactivity-monitor"));
+ this.messageExpiryMonitor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-msg-expiry-monitor"));
this.compactionMonitor =
Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("pulsar-compaction-monitor"));
- this.consumedLedgersMonitor = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("consumed-Ledgers-monitor"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-compaction-monitor"));
+ this.consumedLedgersMonitor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("consumed-Ledgers-monitor"));
this.topicPublishRateLimiterMonitor =
new PublishRateLimiterMonitor("pulsar-topic-publish-rate-limiter-monitor");
this.brokerPublishRateLimiterMonitor =
new PublishRateLimiterMonitor("pulsar-broker-publish-rate-limiter-monitor");
this.backlogQuotaManager = new BacklogQuotaManager(pulsar);
- this.backlogQuotaChecker = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-backlog-quota-checker"));
+ this.backlogQuotaChecker = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-backlog-quota-checker"));
this.authenticationService = new AuthenticationService(pulsar.getConfiguration());
this.blockedDispatchers =
ConcurrentOpenHashSet.<PersistentDispatcherMultipleConsumers>newBuilder().build();
@@ -431,7 +433,8 @@ public class BrokerService implements Closeable {
bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024));
EventLoopUtil.enableTriggeredMode(bootstrap);
- DefaultThreadFactory defaultThreadFactory = new DefaultThreadFactory("pulsar-ph-" + protocol);
+ DefaultThreadFactory defaultThreadFactory =
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-ph-" + protocol);
EventLoopGroup dedicatedWorkerGroup =
EventLoopUtil.newEventLoopGroup(configuration.getNumIOThreads(), false, defaultThreadFactory);
bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(dedicatedWorkerGroup));
@@ -536,7 +539,7 @@ public class BrokerService implements Closeable {
int interval = pulsar().getConfiguration().getBrokerDeduplicationSnapshotFrequencyInSeconds();
if (interval > 0 && pulsar().getConfiguration().isBrokerDeduplicationEnabled()) {
this.deduplicationSnapshotMonitor =
- Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory(
+ Executors.newSingleThreadScheduledExecutor(new ExecutorProvider.ExtendedThreadFactory(
"deduplication-snapshot-monitor"));
deduplicationSnapshotMonitor.scheduleAtFixedRate(safeRun(() -> forEachTopic(
Topic::checkDeduplicationSnapshot))
@@ -663,7 +666,7 @@ public class BrokerService implements Closeable {
stop();
}
//start monitor.
- scheduler = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory(name));
+ scheduler = Executors.newSingleThreadScheduledExecutor(new ExecutorProvider.ExtendedThreadFactory(name));
// schedule task that sums up publish-rate across all cnx on a topic ,
// and check the rate limit exceeded or not.
scheduler.scheduleAtFixedRate(safeRun(checkTask), tickTimeMs, tickTimeMs, TimeUnit.MILLISECONDS);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java
index 0e59286861a..18625a62a23 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java
@@ -20,7 +20,6 @@ package org.apache.pulsar.broker.stats.prometheus.metrics;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.annotations.VisibleForTesting;
-import io.netty.util.concurrent.DefaultThreadFactory;
import io.prometheus.client.Collector;
import java.io.IOException;
import java.io.Writer;
@@ -34,6 +33,7 @@ import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stats.StatsProvider;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.util.ExecutorProvider;
/**
* A <i>Prometheus</i> based {@link StatsProvider} implementation.
@@ -90,7 +90,8 @@ public class PrometheusMetricsProvider implements StatsProvider {
@Override
public void start(Configuration conf) {
- executor = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("metrics"));
+ executor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("metrics"));
int latencyRolloverSeconds = conf.getInt(PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS,
DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/tools/LoadReportCommand.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/tools/LoadReportCommand.java
index 34340c3b0c5..3a77a8dccef 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/tools/LoadReportCommand.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/tools/LoadReportCommand.java
@@ -32,6 +32,7 @@ import org.apache.pulsar.broker.loadbalance.BrokerHostUsage;
import org.apache.pulsar.broker.loadbalance.impl.GenericBrokerHostUsageImpl;
import org.apache.pulsar.broker.loadbalance.impl.LinuxBrokerHostUsageImpl;
import org.apache.pulsar.broker.tools.LoadReportCommand.Flags;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
@@ -88,7 +89,8 @@ public class LoadReportCommand extends CliCommand<CliFlags, Flags> {
spec.console().println("--------------------------------------");
spec.console().println();
- ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+ ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("load-report"));
BrokerHostUsage hostUsage;
try {
if (isLinux) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
index 1f34de71979..e6f363ad28a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java
@@ -20,7 +20,6 @@ package org.apache.pulsar.client.impl;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.base.Strings;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
@@ -38,6 +37,7 @@ import org.apache.pulsar.client.api.AutoClusterFailoverBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.ServiceUrlProvider;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
@Slf4j
@Data
@@ -80,7 +80,7 @@ public class AutoClusterFailover implements ServiceUrlProvider {
this.intervalMs = builder.checkIntervalMs;
this.resolver = new PulsarServiceNameResolver();
this.executor = Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("pulsar-service-provider"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-service-provider"));
}
@Override
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java
index 98ab5362a56..2e5e24fecd5 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java
@@ -25,7 +25,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
@@ -44,6 +43,7 @@ import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.ControlledClusterFailoverBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.ServiceUrlProvider;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.asynchttpclient.AsyncHttpClient;
import org.asynchttpclient.AsyncHttpClientConfig;
@@ -74,7 +74,7 @@ public class ControlledClusterFailover implements ServiceUrlProvider {
this.currentPulsarServiceUrl = builder.defaultServiceUrl;
this.interval = builder.interval;
this.executor = Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("pulsar-service-provider"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-service-provider"));
this.httpClient = buildHttpClient();
this.requestBuilder = httpClient.prepareGet(builder.urlProvider)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index 8ba26e0d15e..27204abe437 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -26,7 +26,6 @@ import com.google.common.cache.LoadingCache;
import io.netty.channel.EventLoopGroup;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.time.Clock;
@@ -756,7 +755,7 @@ public class PulsarClientImpl implements PulsarClient {
// would happen
CompletableFuture<Void> combinedFuture = FutureUtil.waitForAll(futures);
ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("pulsar-client-shutdown-timeout-scheduler"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-client-shutdown-timeout-scheduler"));
FutureUtil.addTimeoutHandling(combinedFuture, Duration.ofSeconds(CLOSE_TIMEOUT_SECONDS),
shutdownExecutor, () -> FutureUtil.createTimeoutException("Closing producers and consumers timed out.",
PulsarClientImpl.class, "closeAsync"));
@@ -1082,7 +1081,7 @@ public class PulsarClientImpl implements PulsarClient {
}
private static ThreadFactory getThreadFactory(String poolName) {
- return new DefaultThreadFactory(poolName, Thread.currentThread().isDaemon());
+ return new ExecutorProvider.ExtendedThreadFactory(poolName, Thread.currentThread().isDaemon());
}
void cleanupProducer(ProducerBase<?> producer) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java b/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
index 67606af63a7..548c0a2abc4 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
@@ -40,10 +40,12 @@ public class ExecutorProvider {
private final String poolName;
private volatile boolean isShutdown;
- protected static class ExtendedThreadFactory extends DefaultThreadFactory {
-
+ public static class ExtendedThreadFactory extends DefaultThreadFactory {
@Getter
private Thread thread;
+ public ExtendedThreadFactory(String poolName) {
+ super(poolName, false);
+ }
public ExtendedThreadFactory(String poolName, boolean daemon) {
super(poolName, daemon);
}
@@ -51,6 +53,8 @@ public class ExecutorProvider {
@Override
public Thread newThread(Runnable r) {
thread = super.newThread(r);
+ thread.setUncaughtExceptionHandler((t, e) ->
+ log.error("Thread {} got uncaught Exception", t.getName(), e));
return thread;
}
}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java
index 93dd30325b0..eac21f7f07a 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java
@@ -18,11 +18,11 @@
*/
package org.apache.pulsar.functions.instance;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import lombok.Getter;
+import org.apache.pulsar.client.util.ExecutorProvider;
public class InstanceCache {
@@ -33,7 +33,7 @@ public class InstanceCache {
private InstanceCache() {
ThreadFactory namedThreadFactory =
- new DefaultThreadFactory("function-timer-thread");
+ new ExecutorProvider.ExtendedThreadFactory("function-timer-thread");
scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(namedThreadFactory);
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java
index 7e1c86a18fc..3c1ed981030 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.functions.worker;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.common.annotations.VisibleForTesting;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
@@ -30,6 +31,7 @@ import java.util.function.Supplier;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.util.ExecutorProvider;
@Slf4j
public class ClusterServiceCoordinator implements AutoCloseable {
@@ -53,6 +55,17 @@ public class ClusterServiceCoordinator implements AutoCloseable {
private final Supplier<Boolean> isLeader;
public ClusterServiceCoordinator(String workerId, LeaderService leaderService, Supplier<Boolean> isLeader) {
+ this(workerId, leaderService, isLeader, Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("cluster-service-coordinator-timer")));
+ }
+
+ @VisibleForTesting
+ ClusterServiceCoordinator(
+ String workerId,
+ LeaderService leaderService,
+ Supplier<Boolean> isLeader,
+ ScheduledExecutorService executor
+ ) {
this.workerId = workerId;
this.leaderService = leaderService;
this.isLeader = isLeader;