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 2021/12/09 12:19:44 UTC
[pulsar] 01/05: Catch exceptions in scheduled tasks to prevent unintended cancellation (#12853)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5a00c02f19adaae9ef27394aa3cdf20a3b1bc2d1
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Nov 19 08:23:27 2021 +0200
Catch exceptions in scheduled tasks to prevent unintended cancellation (#12853)
* Catch exceptions in scheduled tasks to prevent unintended cancellation
- ScheduledExecutorService#scheduleAtFixedRate won't schedule the next execution if running the task
throws an exception. This can lead to unintended cancellation of the scheduled task in failure scenarios.
* Address review feedback: use private constructor
* Address review feedback: Use private inner class instead of Lambda
(cherry picked from commit afdfe199be73aad7b061e0d5b625787d249f1530)
---
.../mledger/impl/ManagedLedgerFactoryImpl.java | 5 +-
.../impl/GenericBrokerHostUsageImpl.java | 6 ++-
.../loadbalance/impl/LinuxBrokerHostUsageImpl.java | 4 +-
.../broker/resourcegroup/ResourceGroupService.java | 9 ++--
.../ResourceUsageTopicTransportManager.java | 3 +-
.../pulsar/broker/service/BrokerService.java | 2 +-
.../ReplicatedSubscriptionsController.java | 3 +-
.../service/persistent/SubscribeRateLimiter.java | 3 +-
.../metrics/PrometheusMetricsProvider.java | 6 +--
.../org/apache/pulsar/client/impl/ClientCnx.java | 6 ++-
.../apache/pulsar/client/impl/ConsumerImpl.java | 9 ++--
.../PersistentAcknowledgmentsGroupingTracker.java | 3 +-
.../apache/pulsar/client/impl/ProducerImpl.java | 43 ++++++++--------
.../pulsar/common/protocol/PulsarHandler.java | 6 ++-
.../org/apache/pulsar/common/stats/JvmMetrics.java | 3 +-
.../org/apache/pulsar/common/util/RateLimiter.java | 4 +-
.../org/apache/pulsar/common/util/Runnables.java | 59 ++++++++++++++++++++++
.../apache/pulsar/common/util/RunnablesTest.java | 32 ++++++++++++
.../windowing/WaterMarkEventGenerator.java | 5 +-
.../windowing/triggers/TimeTriggerPolicy.java | 5 +-
.../pulsar/functions/runtime/RuntimeSpawner.java | 37 +++++++-------
.../functions/runtime/process/ProcessRuntime.java | 22 ++++----
.../worker/ClusterServiceCoordinator.java | 5 +-
.../pulsar/functions/worker/SchedulerManager.java | 9 ++--
.../pulsar/metadata/impl/ZKSessionWatcher.java | 7 ++-
.../apache/pulsar/websocket/stats/JvmMetrics.java | 4 +-
.../apache/pulsar/websocket/stats/ProxyStats.java | 4 +-
27 files changed, 218 insertions(+), 86 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
index 94dee65..e176181 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.mledger.impl;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.bookkeeper.mledger.ManagedLedgerException.getManagedLedgerException;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.base.Predicates;
import com.google.common.collect.Maps;
import io.netty.util.concurrent.DefaultThreadFactory;
@@ -187,9 +188,9 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
this.config = config;
this.mbean = new ManagedLedgerFactoryMBeanImpl(this);
this.entryCacheManager = new EntryCacheManager(this);
- this.statsTask = scheduledExecutor.scheduleAtFixedRate(this::refreshStats,
+ this.statsTask = scheduledExecutor.scheduleAtFixedRate(catchingAndLoggingThrowables(this::refreshStats),
0, config.getStatsPeriodSeconds(), TimeUnit.SECONDS);
- this.flushCursorsTask = scheduledExecutor.scheduleAtFixedRate(this::flushCursors,
+ this.flushCursorsTask = scheduledExecutor.scheduleAtFixedRate(catchingAndLoggingThrowables(this::flushCursors),
config.getCursorPositionFlushSeconds(), config.getCursorPositionFlushSeconds(), TimeUnit.SECONDS);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java
index f55d75a..72d36ab 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.loadbalance.impl;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.sun.management.OperatingSystemMXBean;
import java.lang.management.ManagementFactory;
import java.util.concurrent.ScheduledExecutorService;
@@ -53,9 +54,10 @@ public class GenericBrokerHostUsageImpl implements BrokerHostUsage {
this.totalCpuLimit = getTotalCpuLimit();
// Call now to initialize values before the constructor returns
calculateBrokerHostUsage();
- executorService.scheduleAtFixedRate(this::checkCpuLoad, CPU_CHECK_MILLIS,
+ executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::checkCpuLoad), CPU_CHECK_MILLIS,
CPU_CHECK_MILLIS, TimeUnit.MILLISECONDS);
- executorService.scheduleAtFixedRate(this::doCalculateBrokerHostUsage, hostUsageCheckIntervalMin,
+ executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::doCalculateBrokerHostUsage),
+ hostUsageCheckIntervalMin,
hostUsageCheckIntervalMin, TimeUnit.MINUTES);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
index 8e4e3c5..ccbe8e3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.loadbalance.impl;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.base.Charsets;
import com.sun.management.OperatingSystemMXBean;
import java.io.IOException;
@@ -86,7 +87,8 @@ public class LinuxBrokerHostUsageImpl implements BrokerHostUsage {
// Call now to initialize values before the constructor returns
calculateBrokerHostUsage();
- executorService.scheduleAtFixedRate(this::calculateBrokerHostUsage, hostUsageCheckIntervalMin,
+ executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::calculateBrokerHostUsage),
+ hostUsageCheckIntervalMin,
hostUsageCheckIntervalMin, TimeUnit.MINUTES);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
index 677c04a..4900027 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupService.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.resourcegroup;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import io.prometheus.client.Counter;
import io.prometheus.client.Summary;
import java.util.Map;
@@ -575,7 +576,7 @@ public class ResourceGroupService {
cancelStatus, this.aggregateLocalUsagePeriodInSeconds, newPeriodInSeconds, timeUnitScale);
}
this.aggreagteLocalUsagePeriodicTask = pulsar.getExecutor().scheduleAtFixedRate(
- this::aggregateResourceGroupLocalUsages,
+ catchingAndLoggingThrowables(this::aggregateResourceGroupLocalUsages),
newPeriodInSeconds,
newPeriodInSeconds,
timeUnitScale);
@@ -665,7 +666,7 @@ public class ResourceGroupService {
cancelStatus, this.resourceUsagePublishPeriodInSeconds, newPeriodInSeconds, timeUnitScale);
}
this.calculateQuotaPeriodicTask = pulsar.getExecutor().scheduleAtFixedRate(
- this::calculateQuotaForAllResourceGroups,
+ catchingAndLoggingThrowables(this::calculateQuotaForAllResourceGroups),
newPeriodInSeconds,
newPeriodInSeconds,
timeUnitScale);
@@ -680,12 +681,12 @@ public class ResourceGroupService {
long periodInSecs = config.getResourceUsageTransportPublishIntervalInSecs();
this.aggregateLocalUsagePeriodInSeconds = this.resourceUsagePublishPeriodInSeconds = periodInSecs;
this.aggreagteLocalUsagePeriodicTask = this.pulsar.getExecutor().scheduleAtFixedRate(
- this::aggregateResourceGroupLocalUsages,
+ catchingAndLoggingThrowables(this::aggregateResourceGroupLocalUsages),
periodInSecs,
periodInSecs,
this.timeUnitScale);
this.calculateQuotaPeriodicTask = this.pulsar.getExecutor().scheduleAtFixedRate(
- this::calculateQuotaForAllResourceGroups,
+ catchingAndLoggingThrowables(this::calculateQuotaForAllResourceGroups),
periodInSecs,
periodInSecs,
this.timeUnitScale);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTopicTransportManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTopicTransportManager.java
index d3584a8..9826991 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTopicTransportManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTopicTransportManager.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.resourcegroup;
import static org.apache.pulsar.client.api.CompressionType.LZ4;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.collect.Sets;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -79,7 +80,7 @@ public class ResourceUsageTopicTransportManager implements ResourceUsageTranspor
public ResourceUsageWriterTask() throws PulsarClientException {
producer = createProducer();
resourceUsagePublishTask = pulsarService.getExecutor().scheduleAtFixedRate(
- this,
+ catchingAndLoggingThrowables(this),
pulsarService.getConfig().getResourceUsageTransportPublishIntervalInSecs(),
pulsarService.getConfig().getResourceUsageTransportPublishIntervalInSecs(),
TimeUnit.SECONDS);
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 e592336..0835eb0 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
@@ -331,7 +331,7 @@ public class BrokerService implements Closeable {
log.info("Enabling per-broker unack-message limit {} and dispatcher-limit {} on blocked-broker",
maxUnackedMessages, maxUnackedMsgsPerDispatcher);
// block misbehaving dispatcher by checking periodically
- pulsar.getExecutor().scheduleAtFixedRate(() -> checkUnAckMessageDispatching(),
+ pulsar.getExecutor().scheduleAtFixedRate(safeRun(this::checkUnAckMessageDispatching),
600, 30, TimeUnit.SECONDS);
} else {
this.maxUnackedMessages = 0;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java
index e0501ea..2b1ae4b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service.persistent;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import io.netty.buffer.ByteBuf;
import io.prometheus.client.Gauge;
import java.io.IOException;
@@ -78,7 +79,7 @@ public class ReplicatedSubscriptionsController implements AutoCloseable, Topic.P
this.topic = topic;
this.localCluster = localCluster;
timer = topic.getBrokerService().pulsar().getExecutor()
- .scheduleAtFixedRate(this::startNewSnapshot, 0,
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(this::startNewSnapshot), 0,
topic.getBrokerService().pulsar().getConfiguration()
.getReplicatedSubscriptionsSnapshotFrequencyMillis(),
TimeUnit.MILLISECONDS);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SubscribeRateLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SubscribeRateLimiter.java
index e1d39ae..93707b4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SubscribeRateLimiter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SubscribeRateLimiter.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.service.persistent;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.base.MoreObjects;
import java.util.Objects;
import java.util.Optional;
@@ -271,7 +272,7 @@ public class SubscribeRateLimiter {
}
private ScheduledFuture<?> createTask() {
- return executorService.scheduleAtFixedRate(this::closeAndClearRateLimiters,
+ return executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::closeAndClearRateLimiters),
this.subscribeRate.ratePeriodInSecond,
this.subscribeRate.ratePeriodInSecond,
TimeUnit.SECONDS);
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 dfb27f4..0e59286 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
@@ -18,6 +18,7 @@
*/
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;
@@ -95,9 +96,8 @@ public class PrometheusMetricsProvider implements StatsProvider {
DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS);
cluster = conf.getString(CLUSTER_NAME, DEFAULT_CLUSTER_NAME);
- executor.scheduleAtFixedRate(() -> {
- rotateLatencyCollection();
- }, 1, latencyRolloverSeconds, TimeUnit.SECONDS);
+ executor.scheduleAtFixedRate(catchingAndLoggingThrowables(this::rotateLatencyCollection),
+ 1, latencyRolloverSeconds, TimeUnit.SECONDS);
}
@Override
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index b0895f4..e2386a9 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.client.impl;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static org.apache.pulsar.client.impl.TransactionMetaStoreHandler.getExceptionByServerError;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.collect.Queues;
import io.netty.buffer.ByteBuf;
@@ -217,8 +218,9 @@ public class ClientCnx extends PulsarHandler {
this.localAddress = ctx.channel().localAddress();
this.remoteAddress = ctx.channel().remoteAddress();
- this.timeoutTask = this.eventLoopGroup.scheduleAtFixedRate(() -> checkRequestTimeout(), operationTimeoutMs,
- operationTimeoutMs, TimeUnit.MILLISECONDS);
+ this.timeoutTask = this.eventLoopGroup
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(this::checkRequestTimeout), operationTimeoutMs,
+ operationTimeoutMs, TimeUnit.MILLISECONDS);
if (proxyToTargetBrokerAddress == null) {
if (log.isDebugEnabled()) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 7e42d0d..11b1def 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.pulsar.common.protocol.Commands.hasChecksum;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.collect.ComparisonChain;
import com.google.common.collect.Iterables;
import com.scurrilous.circe.checksum.Crc32cIntChecksum;
@@ -1248,10 +1249,10 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
// Lazy task scheduling to expire incomplete chunk message
if (!expireChunkMessageTaskScheduled && expireTimeOfIncompleteChunkedMessageMillis > 0) {
- internalPinnedExecutor.scheduleAtFixedRate(() -> {
- removeExpireIncompleteChunkedMessages();
- }, expireTimeOfIncompleteChunkedMessageMillis, expireTimeOfIncompleteChunkedMessageMillis,
- TimeUnit.MILLISECONDS);
+ internalPinnedExecutor
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(this::removeExpireIncompleteChunkedMessages),
+ expireTimeOfIncompleteChunkedMessageMillis, expireTimeOfIncompleteChunkedMessageMillis,
+ TimeUnit.MILLISECONDS);
expireChunkMessageTaskScheduled = true;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
index 21ef9b3..aa65c61 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import io.netty.buffer.ByteBuf;
import io.netty.channel.EventLoopGroup;
import java.util.ArrayList;
@@ -102,7 +103,7 @@ public class PersistentAcknowledgmentsGroupingTracker implements Acknowledgments
this.currentCumulativeAckFuture = new TimedCompletableFuture<>();
if (acknowledgementGroupTimeMicros > 0) {
- scheduledTask = eventLoopGroup.next().scheduleWithFixedDelay(this::flush, acknowledgementGroupTimeMicros,
+ scheduledTask = eventLoopGroup.next().scheduleWithFixedDelay(catchingAndLoggingThrowables(this::flush), acknowledgementGroupTimeMicros,
acknowledgementGroupTimeMicros, TimeUnit.MICROSECONDS);
} else {
scheduledTask = null;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index c2227c4..e33fb5f 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -29,6 +29,7 @@ import static org.apache.pulsar.client.impl.ProducerBase.MultiSchemaMode.Auto;
import static org.apache.pulsar.client.impl.ProducerBase.MultiSchemaMode.Enabled;
import static org.apache.pulsar.common.protocol.Commands.hasChecksum;
import static org.apache.pulsar.common.protocol.Commands.readChecksum;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.annotations.VisibleForTesting;
import io.netty.buffer.ByteBuf;
import io.netty.util.Recycler;
@@ -204,7 +205,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
if (this.msgCrypto != null) {
// Regenerate data key cipher at fixed interval
- keyGeneratorTask = client.eventLoopGroup().scheduleWithFixedDelay(() -> {
+ keyGeneratorTask = client.eventLoopGroup().scheduleWithFixedDelay(catchingAndLoggingThrowables(() -> {
try {
msgCrypto.addPublicKeyCipher(conf.getEncryptionKeys(), conf.getCryptoKeyReader());
} catch (CryptoException e) {
@@ -217,7 +218,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
producerName, topic)));
}
}
- }, 0L, 4L, TimeUnit.HOURS);
+ }), 0L, 4L, TimeUnit.HOURS);
}
if (conf.getSendTimeoutMs() > 0) {
@@ -1450,24 +1451,26 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
if (!producerCreatedFuture.isDone() && isBatchMessagingEnabled()) {
// schedule the first batch message task
- batchTimerTask = cnx.ctx().executor().scheduleAtFixedRate(() -> {
- if (log.isTraceEnabled()) {
- log.trace(
- "[{}] [{}] Batching the messages from the batch container from timer thread",
- topic,
- producerName);
- }
- // semaphore acquired when message was enqueued to container
- synchronized (ProducerImpl.this) {
- // If it's closing/closed we need to ignore the send batch timer and not
- // schedule next timeout.
- if (getState() == State.Closing || getState() == State.Closed) {
- return;
- }
-
- batchMessageAndSend();
- }
- }, 0, conf.getBatchingMaxPublishDelayMicros(), TimeUnit.MICROSECONDS);
+ batchTimerTask = cnx.ctx().executor()
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(() -> {
+ if (log.isTraceEnabled()) {
+ log.trace(
+ "[{}] [{}] Batching the messages from the batch container from "
+ + "timer thread",
+ topic,
+ producerName);
+ }
+ // semaphore acquired when message was enqueued to container
+ synchronized (ProducerImpl.this) {
+ // If it's closing/closed we need to ignore the send batch timer and not
+ // schedule next timeout.
+ if (getState() == State.Closing || getState() == State.Closed) {
+ return;
+ }
+
+ batchMessageAndSend();
+ }
+ }), 0, conf.getBatchingMaxPublishDelayMicros(), TimeUnit.MICROSECONDS);
}
resendMessages(cnx, epoch);
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarHandler.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarHandler.java
index 481517c..557dbfa 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarHandler.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarHandler.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.common.protocol;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.concurrent.ScheduledFuture;
import java.net.SocketAddress;
@@ -65,8 +66,9 @@ public abstract class PulsarHandler extends PulsarDecoder {
log.debug("[{}] Scheduling keep-alive task every {} s", ctx.channel(), keepAliveIntervalSeconds);
}
if (keepAliveIntervalSeconds > 0) {
- this.keepAliveTask = ctx.executor().scheduleAtFixedRate(this::handleKeepAliveTimeout,
- keepAliveIntervalSeconds, keepAliveIntervalSeconds, TimeUnit.SECONDS);
+ this.keepAliveTask = ctx.executor()
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(this::handleKeepAliveTimeout),
+ keepAliveIntervalSeconds, keepAliveIntervalSeconds, TimeUnit.SECONDS);
}
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/stats/JvmMetrics.java b/pulsar-common/src/main/java/org/apache/pulsar/common/stats/JvmMetrics.java
index 6780563..83b1406 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/stats/JvmMetrics.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/stats/JvmMetrics.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.common.stats;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import io.netty.buffer.PoolArenaMetric;
@@ -96,7 +97,7 @@ public class JvmMetrics {
public JvmMetrics(ScheduledExecutorService executor, String componentName, JvmGCMetricsLogger gcLogger) {
this.gcLogger = gcLogger;
if (executor != null) {
- executor.scheduleAtFixedRate(gcLogger::refresh, 0, 1, TimeUnit.MINUTES);
+ executor.scheduleAtFixedRate(catchingAndLoggingThrowables(gcLogger::refresh), 0, 1, TimeUnit.MINUTES);
}
this.componentName = componentName;
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java
index 6f2d899..20ca181 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/RateLimiter.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.common.util;
import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.base.MoreObjects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@@ -256,7 +257,8 @@ public class RateLimiter implements AutoCloseable{
}
protected ScheduledFuture<?> createTask() {
- return executorService.scheduleAtFixedRate(this::renew, this.rateTime, this.rateTime, this.timeUnit);
+ return executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this::renew), this.rateTime,
+ this.rateTime, this.timeUnit);
}
synchronized void renew() {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Runnables.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Runnables.java
new file mode 100644
index 0000000..b720c4f
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Runnables.java
@@ -0,0 +1,59 @@
+/**
+ * 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.common.util;
+
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class Runnables {
+ private static final Logger LOGGER = LoggerFactory.getLogger(Runnables.class);
+
+ private Runnables() {}
+
+ /**
+ * Wraps a Runnable so that throwables are caught and logged when a Runnable is run.
+ *
+ * The main usecase for this method is to be used in {@link java.util.concurrent.ScheduledExecutorService#scheduleAtFixedRate(Runnable, long, long, TimeUnit)}
+ * calls to ensure that the scheduled task doesn't get cancelled as a result of an uncaught exception.
+ *
+ * @param runnable The runnable to wrap
+ * @return a wrapped Runnable
+ */
+ public static Runnable catchingAndLoggingThrowables(Runnable runnable) {
+ return new CatchingAndLoggingRunnable(runnable);
+ }
+
+ private static final class CatchingAndLoggingRunnable implements Runnable {
+ private final Runnable runnable;
+
+ private CatchingAndLoggingRunnable(Runnable runnable) {
+ this.runnable = runnable;
+ }
+
+ @Override
+ public void run() {
+ try {
+ runnable.run();
+ } catch (Throwable t) {
+ LOGGER.error("Unexpected throwable caught", t);
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/RunnablesTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/RunnablesTest.java
new file mode 100644
index 0000000..6dfbe1b
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/RunnablesTest.java
@@ -0,0 +1,32 @@
+/**
+ * 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.common.util;
+
+import org.testng.annotations.Test;
+
+public class RunnablesTest {
+
+ @Test
+ public void shouldCatchAndLogException() {
+ Runnables.catchingAndLoggingThrowables(() -> {
+ throw new RuntimeException();
+ }).run();
+ }
+}
\ No newline at end of file
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WaterMarkEventGenerator.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WaterMarkEventGenerator.java
index d904202..0024586 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WaterMarkEventGenerator.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WaterMarkEventGenerator.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.functions.windowing;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.ThreadContext;
@@ -134,7 +135,9 @@ public class WaterMarkEventGenerator<T> implements Runnable {
}
public void start() {
- this.executorFuture = executorService.scheduleAtFixedRate(this, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
+ this.executorFuture =
+ executorService.scheduleAtFixedRate(catchingAndLoggingThrowables(this), intervalMs, intervalMs,
+ TimeUnit.MILLISECONDS);
}
public void shutdown() {
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/triggers/TimeTriggerPolicy.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/triggers/TimeTriggerPolicy.java
index bc40f53..3ae0c23 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/triggers/TimeTriggerPolicy.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/triggers/TimeTriggerPolicy.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.functions.windowing.triggers;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.ThreadContext;
@@ -75,7 +76,9 @@ public class TimeTriggerPolicy<T> implements TriggerPolicy<T, Void> {
@Override
public void start() {
- executorFuture = executor.scheduleAtFixedRate(newTriggerTask(), duration, duration, TimeUnit.MILLISECONDS);
+ executorFuture =
+ executor.scheduleAtFixedRate(catchingAndLoggingThrowables(newTriggerTask()), duration, duration,
+ TimeUnit.MILLISECONDS);
}
@Override
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
index 471af68..60c8697 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
@@ -23,6 +23,7 @@
*/
package org.apache.pulsar.functions.runtime;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledFuture;
@@ -81,23 +82,25 @@ public class RuntimeSpawner implements AutoCloseable {
// monitor function runtime to make sure it is running. If not, restart the function runtime
if (!runtimeFactory.externallyManaged() && instanceLivenessCheckFreqMs > 0) {
- processLivenessCheckTimer = InstanceCache.getInstanceCache().getScheduledExecutorService().scheduleAtFixedRate(() -> {
- Runtime runtime = RuntimeSpawner.this.runtime;
- if (runtime != null && !runtime.isAlive()) {
- log.error("{}/{}/{} Function Container is dead with following exception. Restarting.", details.getTenant(),
- details.getNamespace(), details.getName(), runtime.getDeathException());
- // Just for the sake of sanity, just destroy the runtime
- try {
- runtime.stop();
- runtimeDeathException = runtime.getDeathException();
- runtime.start();
- } catch (Exception e) {
- log.error("{}/{}/{}-{} Function Restart failed", details.getTenant(),
- details.getNamespace(), details.getName(), e, e);
- }
- numRestarts++;
- }
- }, instanceLivenessCheckFreqMs, instanceLivenessCheckFreqMs, TimeUnit.MILLISECONDS);
+ processLivenessCheckTimer = InstanceCache.getInstanceCache().getScheduledExecutorService()
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(() -> {
+ Runtime runtime = RuntimeSpawner.this.runtime;
+ if (runtime != null && !runtime.isAlive()) {
+ log.error("{}/{}/{} Function Container is dead with following exception. Restarting.",
+ details.getTenant(),
+ details.getNamespace(), details.getName(), runtime.getDeathException());
+ // Just for the sake of sanity, just destroy the runtime
+ try {
+ runtime.stop();
+ runtimeDeathException = runtime.getDeathException();
+ runtime.start();
+ } catch (Exception e) {
+ log.error("{}/{}/{}-{} Function Restart failed", details.getTenant(),
+ details.getNamespace(), details.getName(), e, e);
+ }
+ numRestarts++;
+ }
+ }), instanceLivenessCheckFreqMs, instanceLivenessCheckFreqMs, TimeUnit.MILLISECONDS);
}
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java
index 7d585bc..6135b6b 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.functions.runtime.process;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
@@ -180,16 +181,17 @@ class ProcessRuntime implements Runtime {
.build();
stub = InstanceControlGrpc.newFutureStub(channel);
- timer = InstanceCache.getInstanceCache().getScheduledExecutorService().scheduleAtFixedRate(() -> {
- CompletableFuture<InstanceCommunication.HealthCheckResult> result = healthCheck();
- try {
- result.get();
- } catch (Exception e) {
- log.error("Health check failed for {}-{}",
- instanceConfig.getFunctionDetails().getName(),
- instanceConfig.getInstanceId(), e);
- }
- }, expectedHealthCheckInterval, expectedHealthCheckInterval, TimeUnit.SECONDS);
+ timer = InstanceCache.getInstanceCache().getScheduledExecutorService()
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(() -> {
+ CompletableFuture<InstanceCommunication.HealthCheckResult> result = healthCheck();
+ try {
+ result.get();
+ } catch (Exception e) {
+ log.error("Health check failed for {}-{}",
+ instanceConfig.getFunctionDetails().getName(),
+ instanceConfig.getInstanceId(), e);
+ }
+ }), expectedHealthCheckInterval, expectedHealthCheckInterval, TimeUnit.SECONDS);
}
}
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 c1682a8..570408e 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
@@ -19,6 +19,7 @@
package org.apache.pulsar.functions.worker;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -69,7 +70,7 @@ public class ClusterServiceCoordinator implements AutoCloseable {
for (Map.Entry<String, TimerTaskInfo> entry : this.tasks.entrySet()) {
TimerTaskInfo timerTaskInfo = entry.getValue();
String taskName = entry.getKey();
- this.executor.scheduleAtFixedRate(() -> {
+ this.executor.scheduleAtFixedRate(catchingAndLoggingThrowables(() -> {
if (isLeader.get()) {
try {
timerTaskInfo.getTask().run();
@@ -77,7 +78,7 @@ public class ClusterServiceCoordinator implements AutoCloseable {
log.error("Cluster timer task {} failed with exception.", taskName, e);
}
}
- }, timerTaskInfo.getInterval(), timerTaskInfo.getInterval(), TimeUnit.MILLISECONDS);
+ }), timerTaskInfo.getInterval(), timerTaskInfo.getInterval(), TimeUnit.MILLISECONDS);
}
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
index b7fd0a4..130e839 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.functions.worker;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
@@ -532,19 +533,19 @@ public class SchedulerManager implements AutoCloseable {
private void scheduleCompaction(ScheduledExecutorService executor, long scheduleFrequencySec) {
if (executor != null) {
- executor.scheduleWithFixedDelay(() -> {
+ executor.scheduleWithFixedDelay(catchingAndLoggingThrowables(() -> {
if (leaderService.isLeader() && isCompactionNeeded.get()) {
compactAssignmentTopic();
isCompactionNeeded.set(false);
}
- }, scheduleFrequencySec, scheduleFrequencySec, TimeUnit.SECONDS);
+ }), scheduleFrequencySec, scheduleFrequencySec, TimeUnit.SECONDS);
- executor.scheduleWithFixedDelay(() -> {
+ executor.scheduleWithFixedDelay(catchingAndLoggingThrowables(() -> {
if (leaderService.isLeader() && metadataTopicLastMessage.compareTo(functionMetaDataManager.getLastMessageSeen()) != 0) {
metadataTopicLastMessage = functionMetaDataManager.getLastMessageSeen();
compactFunctionMetadataTopic();
}
- }, scheduleFrequencySec, scheduleFrequencySec, TimeUnit.SECONDS);
+ }), scheduleFrequencySec, scheduleFrequencySec, TimeUnit.SECONDS);
}
}
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java
index 9ba6be5..51b3df4 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.metadata.impl;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.concurrent.CompletableFuture;
@@ -67,8 +68,10 @@ public class ZKSessionWatcher implements AutoCloseable, Watcher {
this.scheduler = Executors
.newSingleThreadScheduledExecutor(new DefaultThreadFactory("metadata-store-zk-session-watcher"));
- this.task = scheduler.scheduleAtFixedRate(this::checkConnectionStatus, tickTimeMillis, tickTimeMillis,
- TimeUnit.MILLISECONDS);
+ this.task =
+ scheduler.scheduleAtFixedRate(catchingAndLoggingThrowables(this::checkConnectionStatus), tickTimeMillis,
+ tickTimeMillis,
+ TimeUnit.MILLISECONDS);
this.currentStatus = SessionEvent.SessionReestablished;
}
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/JvmMetrics.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/JvmMetrics.java
index 905c49b..112fc205 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/JvmMetrics.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/JvmMetrics.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.websocket.stats;
import static org.apache.pulsar.common.stats.Metrics.create;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import java.lang.management.ManagementFactory;
import java.util.Map;
@@ -52,7 +53,8 @@ public class JvmMetrics {
private static final Logger log = LoggerFactory.getLogger(JvmMetrics.class);
public JvmMetrics(WebSocketService service) {
- service.getExecutor().scheduleAtFixedRate(this::updateGcStats, 0, 1, TimeUnit.MINUTES);
+ service.getExecutor()
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(this::updateGcStats), 0, 1, TimeUnit.MINUTES);
}
public Metrics generate() {
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/ProxyStats.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/ProxyStats.java
index ef7523b..9bf5f4a 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/ProxyStats.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/stats/ProxyStats.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.websocket.stats;
+import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import static org.apache.pulsar.websocket.ProducerHandler.ENTRY_LATENCY_BUCKETS_USEC;
import java.util.List;
@@ -55,7 +56,8 @@ public class ProxyStats {
this.metricsCollection = Lists.newArrayList();
this.tempMetricsCollection = Lists.newArrayList();
// schedule stat generation task every 1 minute
- service.getExecutor().scheduleAtFixedRate(() -> generate(), 120, 60, TimeUnit.SECONDS);
+ service.getExecutor()
+ .scheduleAtFixedRate(catchingAndLoggingThrowables(this::generate), 120, 60, TimeUnit.SECONDS);
}
/**