You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2018/06/01 02:06:52 UTC
[rocketmq] branch develop updated: Shutdown all thread pools when
broker quits
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 8422e74 Shutdown all thread pools when broker quits
8422e74 is described below
commit 8422e74fa5d1f6c8a29770e568d9bb228ac1f6c7
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Fri Jun 1 10:04:56 2018 +0800
Shutdown all thread pools when broker quits
---
.../org/apache/rocketmq/broker/out/BrokerOuterAPI.java | 1 +
.../rocketmq/producer/AbstractOMSProducer.java | 16 +++++++++++++---
.../apache/rocketmq/store/stats/BrokerStatsManager.java | 1 +
3 files changed, 15 insertions(+), 3 deletions(-)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
index 262e2d2..2825a34 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
@@ -80,6 +80,7 @@ public class BrokerOuterAPI {
public void shutdown() {
this.remotingClient.shutdown();
+ this.brokerOuterExecutor.shutdown();
}
public String fetchNameServerAddr() {
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
index 53fc0f9..3db8590 100644
--- a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
@@ -99,9 +99,19 @@ abstract class AbstractOMSProducer implements ServiceLifecycle, MessageFactory {
return new OMSTimeOutException("-1", String.format("Send message to broker timeout, %dms, Topic=%s, msgId=%s",
this.rocketmqProducer.getSendMsgTimeout(), topic, msgId), e);
} else if (e.getCause() instanceof MQBrokerException || e.getCause() instanceof RemotingConnectException) {
- MQBrokerException brokerException = (MQBrokerException) e.getCause();
- return new OMSRuntimeException("-1", String.format("Received a broker exception, Topic=%s, msgId=%s, %s",
- topic, msgId, brokerException.getErrorMessage()), e);
+ if (e.getCause() instanceof MQBrokerException) {
+ MQBrokerException brokerException = (MQBrokerException) e.getCause();
+ return new OMSRuntimeException("-1", String.format("Received a broker exception, Topic=%s, msgId=%s, %s",
+ topic, msgId, brokerException.getErrorMessage()), e);
+ }
+
+ if (e.getCause() instanceof RemotingConnectException) {
+ RemotingConnectException connectException = (RemotingConnectException)e.getCause();
+ return new OMSRuntimeException("-1",
+ String.format("Network connection experiences failures. Topic=%s, msgId=%s, %s",
+ topic, msgId, connectException.getMessage()),
+ e);
+ }
}
}
// Exception thrown by local.
diff --git a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
index ac8ae3c..4adbed7 100644
--- a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
+++ b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
@@ -109,6 +109,7 @@ public class BrokerStatsManager {
public void shutdown() {
this.scheduledExecutorService.shutdown();
+ this.commercialExecutor.shutdown();
}
public StatsItem getStatsItem(final String statsName, final String statsKey) {
--
To stop receiving notification emails like this one, please contact
lizhanhui@apache.org.