This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 85ed489 CAMEL-12636: camel-jmx consumer should use a dedicated thread
pool for routing JMX notifications
85ed489 is described below
commit 85ed4893b7e41e02d04767302b8bb2b65d2c65d9
Author: Claus Ibsen <claus.ib...@gmail.com>
AuthorDate: Wed Jul 11 09:09:36 2018 +0200
CAMEL-12636: camel-jmx consumer should use a dedicated thread pool for
routing JMX notifications
---
.../camel-jmx/src/main/docs/jmx-component.adoc | 5 +-
.../apache/camel/component/jmx/JMXConsumer.java | 53 ++++++++++++++++----
.../apache/camel/component/jmx/JMXEndpoint.java | 19 +++++++-
.../camel/component/jmx/CamelJmxConsumerTest.java | 57 ++++++++++++++++++++++
4 files changed, 120 insertions(+), 14 deletions(-)
diff --git a/components/camel-jmx/src/main/docs/jmx-component.adoc
b/components/camel-jmx/src/main/docs/jmx-component.adoc
index 3e03c70..47517dc 100644
--- a/components/camel-jmx/src/main/docs/jmx-component.adoc
+++ b/components/camel-jmx/src/main/docs/jmx-component.adoc
@@ -41,11 +41,11 @@ with the following path and query parameters:
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *serverURL* | server url comes from the remaining endpoint | | String
+| *serverURL* | Server url comes from the remaining endpoint. Use platform to
connect to local JVM. | | String
|===
-==== Query Parameters (29 parameters):
+==== Query Parameters (30 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -60,6 +60,7 @@ with the following path and query parameters:
| *observedAttribute* (consumer) | URI Property: monitor types only The
attribute to observe for the monitor bean. | | String
| *exceptionHandler* (consumer) | To let the consumer use a custom
ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this
options is not in use. By default the consumer will deal with exceptions, that
will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer
creates an exchange. | | ExchangePattern
+| *executorService* (consumer) | To use a custom shared thread pool for the
consumers. By default each consume has their own thread-pool to process and
route notifications. | | ExecutorService
| *handback* (advanced) | URI Property: Value to handback to the listener when
a notification is received. This value will be put in the message header with
the key jmx.handback | | Object
| *notificationFilter* (advanced) | URI Property: Reference to a bean that
implements the NotificationFilter. | | NotificationFilter
| *objectProperties* (advanced) | URI Property: properties for the object
name. These values will be used if the objectName param is not set | | Map
diff --git
a/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXConsumer.java
b/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXConsumer.java
index b7ea707..d2db3ae 100644
---
a/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXConsumer.java
+++
b/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXConsumer.java
@@ -20,6 +20,7 @@ import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.Collections;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServerConnection;
@@ -33,7 +34,6 @@ import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import org.apache.camel.Exchange;
-import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
@@ -53,6 +53,12 @@ public class JMXConsumer extends DefaultConsumer implements
NotificationListener
private JMXEndpoint mJmxEndpoint;
private JMXConnector mConnector;
private String mConnectionId;
+
+ /**
+ * Used for processing notifications (should not block notification thread)
+ */
+ private ExecutorService executorService;
+ private boolean shutdownExecutorService;
/**
* Used to schedule delayed connection attempts
@@ -74,13 +80,17 @@ public class JMXConsumer extends DefaultConsumer implements
NotificationListener
*/
private NotificationXmlFormatter mFormatter;
-
public JMXConsumer(JMXEndpoint endpoint, Processor processor) {
super(endpoint, processor);
this.mJmxEndpoint = endpoint;
this.mFormatter = new NotificationXmlFormatter();
}
+ @Override
+ public JMXEndpoint getEndpoint() {
+ return (JMXEndpoint) super.getEndpoint();
+ }
+
/**
* Initializes the mbean server connection and starts listening for
* Notification events from the object.
@@ -89,6 +99,18 @@ public class JMXConsumer extends DefaultConsumer implements
NotificationListener
protected void doStart() throws Exception {
ServiceHelper.startService(mFormatter);
+ if (executorService == null) {
+ if (getEndpoint().getExecutorService() != null) {
+ // use shared thread-pool
+ executorService = getEndpoint().getExecutorService();
+ } else {
+ // lets just use a single threaded thread-pool to process
these notifications
+ String name = "JMXConsumer[" +
getEndpoint().getJMXObjectName().getCanonicalName() + "]";
+ executorService =
getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this,
name);
+ shutdownExecutorService = true;
+ }
+ }
+
// connect to the mbean server
if (mJmxEndpoint.isPlatformServer()) {
setServerConnection(ManagementFactory.getPlatformMBeanServer());
@@ -181,7 +203,7 @@ public class JMXConsumer extends DefaultConsumer implements
NotificationListener
if (mJmxEndpoint.getReconnectOnConnectionFailure()) {
scheduleReconnect();
} else {
- LOG.warn("The JMX consumer will not be reconnected. Use
'reconnectOnConnectionFailure' to "
+ LOG.warn("The JMX consumer will not be reconnected. Use
'reconnectOnConnectionFailure' to "
+ "enable reconnections.");
}
}
@@ -223,10 +245,9 @@ public class JMXConsumer extends DefaultConsumer
implements NotificationListener
/**
* Adds a notification listener to the target bean.
- * @throws Exception
*/
protected void addNotificationListener() throws Exception {
- JMXEndpoint ep = (JMXEndpoint) getEndpoint();
+ JMXEndpoint ep = getEndpoint();
NotificationFilter nf = ep.getNotificationFilter();
ObjectName objectName = ep.getJMXObjectName();
@@ -253,6 +274,11 @@ public class JMXConsumer extends DefaultConsumer
implements NotificationListener
}
ServiceHelper.stopService(mFormatter);
+
+ if (shutdownExecutorService && executorService != null) {
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
+ executorService = null;
+ }
}
/**
@@ -287,8 +313,8 @@ public class JMXConsumer extends DefaultConsumer implements
NotificationListener
* @see
javax.management.NotificationListener#handleNotification(javax.management.Notification,
java.lang.Object)
*/
public void handleNotification(Notification aNotification, Object
aHandback) {
- JMXEndpoint ep = (JMXEndpoint) getEndpoint();
- Exchange exchange =
getEndpoint().createExchange(ExchangePattern.InOnly);
+ JMXEndpoint ep = getEndpoint();
+ Exchange exchange = getEndpoint().createExchange();
Message message = exchange.getIn();
message.setHeader("jmx.handback", aHandback);
try {
@@ -297,11 +323,18 @@ public class JMXConsumer extends DefaultConsumer
implements NotificationListener
} else {
message.setBody(aNotification);
}
- getProcessor().process(exchange);
+
+ // process the notification from thred pool to not block this
notification callback thread from the JVM
+ executorService.submit(() -> {
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ getExceptionHandler().handleException("Failed to process
notification", e);
+ }
+ });
+
} catch (NotificationFormatException e) {
getExceptionHandler().handleException("Failed to marshal
notification", e);
- } catch (Exception e) {
- getExceptionHandler().handleException("Failed to process
notification", e);
}
}
diff --git
a/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXEndpoint.java
b/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXEndpoint.java
index d3161c1..6891f20 100644
---
a/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXEndpoint.java
+++
b/components/camel-jmx/src/main/java/org/apache/camel/component/jmx/JMXEndpoint.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.jmx;
import java.util.Hashtable;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import javax.management.MalformedObjectNameException;
import javax.management.NotificationFilter;
import javax.management.ObjectName;
@@ -60,7 +61,7 @@ public class JMXEndpoint extends DefaultEndpoint {
protected static final String ERR_OBSERVED_ATTRIBUTE = "Observed attribute
must be specified";
/**
- * server url comes from the remaining endpoint
+ * Server url comes from the remaining endpoint. Use platform to connect
to local JVM.
*/
@UriPath
private String serverURL;
@@ -219,6 +220,12 @@ public class JMXEndpoint extends DefaultEndpoint {
private Map<String, String> objectProperties;
/**
+ * To use a custom shared thread pool for the consumers. By default each
consume has their own thread-pool to process and route notifications.
+ */
+ @UriParam(label = "consumer,advanced")
+ private ExecutorService executorService;
+
+ /**
* cached object name that was built from the objectName param or the
hashtable
*/
private transient ObjectName jmxObjectName;
@@ -518,7 +525,15 @@ public class JMXEndpoint extends DefaultEndpoint {
public void setReconnectDelay(int reconnectDelay) {
this.reconnectDelay = reconnectDelay;
}
-
+
+ public ExecutorService getExecutorService() {
+ return executorService;
+ }
+
+ public void setExecutorService(ExecutorService executorService) {
+ this.executorService = executorService;
+ }
+
private ObjectName buildObjectName() throws MalformedObjectNameException {
ObjectName objectName;
if (getObjectProperties() == null) {
diff --git
a/components/camel-jmx/src/test/java/org/apache/camel/component/jmx/CamelJmxConsumerTest.java
b/components/camel-jmx/src/test/java/org/apache/camel/component/jmx/CamelJmxConsumerTest.java
new file mode 100644
index 0000000..eb50602
--- /dev/null
+++
b/components/camel-jmx/src/test/java/org/apache/camel/component/jmx/CamelJmxConsumerTest.java
@@ -0,0 +1,57 @@
+/**
+ * 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.camel.component.jmx;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.api.management.mbean.ManagedRouteMBean;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class CamelJmxConsumerTest extends CamelTestSupport {
+
+ @Override
+ protected boolean useJmx() {
+ return true;
+ }
+
+ @Test
+ public void testJmxConsumer() throws Exception {
+ getMockEndpoint("mock:result").expectedMinimumMessageCount(1);
+
getMockEndpoint("mock:result").message(0).body().contains("<newValue>true</newValue>");
+
+ // change the attribute so JMX triggers
+ ManagedRouteMBean mr = context.getManagedRoute("foo",
ManagedRouteMBean.class);
+ mr.setTracing(true);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
from("jmx:platform?objectDomain=org.apache.camel&key.context=camel-1&key.type=routes&key.name=\"foo\"").routeId("jmxRoute")
+ .to("log:jmx")
+ .to("mock:result");
+
+ from("direct:foo").routeId("foo").to("log:foo", "mock:foo");
+ }
+ };
+ }
+}
