You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2014/12/21 08:51:03 UTC
[2/5] camel git commit: CAMEL-8165: Async routing engine - Add
insight into threads blocked waiting for callbacks
CAMEL-8165: Async routing engine - Add insight into threads blocked waiting for callbacks
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/36c2d70d
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/36c2d70d
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/36c2d70d
Branch: refs/heads/master
Commit: 36c2d70d5f952e4b8eaa9cf9ed0b3af26e5f66ef
Parents: 057fb60
Author: Claus Ibsen <da...@apache.org>
Authored: Sat Dec 20 13:55:07 2014 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Sat Dec 20 14:35:09 2014 +0100
----------------------------------------------------------------------
.../management/mbean/CamelOpenMBeanTypes.java | 8 +-
.../impl/DefaultAsyncProcessorAwaitManager.java | 120 ++++++++++++++++---
.../ManagedAsyncProcessorAwaitManager.java | 24 ++--
.../camel/spi/AsyncProcessorAwaitManager.java | 28 +++++
...AsyncProcessorAwaitManagerInterruptTest.java | 86 +++++++++++++
.../async/AsyncProcessorAwaitManagerTest.java | 80 +++++++++++++
6 files changed, 311 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/36c2d70d/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java b/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java
index 7c88dd9..070f5c8 100644
--- a/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java
+++ b/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java
@@ -90,13 +90,13 @@ public final class CamelOpenMBeanTypes {
public static TabularType listAwaitThreadsTabularType() throws OpenDataException {
CompositeType ct = listAwaitThreadsCompositeType();
- return new TabularType("listAwaitThreads", "Lists blocked threads by the routing engine", ct, new String[]{"name"});
+ return new TabularType("listAwaitThreads", "Lists blocked threads by the routing engine", ct, new String[]{"id"});
}
public static CompositeType listAwaitThreadsCompositeType() throws OpenDataException {
- return new CompositeType("threads", "Threads", new String[]{"name", "exchangeId", "duration"},
- new String[]{"Thread name", "ExchangeId", "Duration"},
- new OpenType[]{SimpleType.STRING, SimpleType.STRING, SimpleType.STRING});
+ return new CompositeType("threads", "Threads", new String[]{"id", "name", "exchangeId", "routeId", "nodeId", "duration"},
+ new String[]{"Thread Id", "Thread name", "ExchangeId", "RouteId", "NodeId", "Duration"},
+ new OpenType[]{SimpleType.STRING, SimpleType.STRING, SimpleType.STRING, SimpleType.STRING, SimpleType.STRING, SimpleType.STRING});
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/36c2d70d/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java
index 9d78260..20c2927 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultAsyncProcessorAwaitManager.java
@@ -18,29 +18,41 @@ package org.apache.camel.impl;
import java.util.Collection;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.RejectedExecutionException;
import org.apache.camel.Exchange;
+import org.apache.camel.MessageHistory;
+import org.apache.camel.processor.DefaultExchangeFormatter;
import org.apache.camel.spi.AsyncProcessorAwaitManager;
+import org.apache.camel.spi.ExchangeFormatter;
import org.apache.camel.support.ServiceSupport;
+import org.apache.camel.util.MessageHelper;
+import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class DefaultAsyncProcessorAwaitManager extends ServiceSupport implements AsyncProcessorAwaitManager {
- // TODO: capture message history of the exchange when it was interrupted
- // TODO: capture route id, node id where thread is blocked
- // TODO: rename to AsyncInflightRepository?
-
private static final Logger LOG = LoggerFactory.getLogger(DefaultAsyncProcessorAwaitManager.class);
private final Map<Exchange, AwaitThread> inflight = new ConcurrentHashMap<Exchange, AwaitThread>();
-
+ private final ExchangeFormatter exchangeFormatter;
private boolean interruptThreadsWhileStopping = true;
+ public DefaultAsyncProcessorAwaitManager() {
+ // setup exchange formatter to be used for message history dump
+ DefaultExchangeFormatter formatter = new DefaultExchangeFormatter();
+ formatter.setShowExchangeId(true);
+ formatter.setMultiline(true);
+ formatter.setShowHeaders(true);
+ formatter.setStyle(DefaultExchangeFormatter.OutputStyle.Fixed);
+ this.exchangeFormatter = formatter;
+ }
+
@Override
public void await(Exchange exchange, CountDownLatch latch) {
LOG.trace("Waiting for asynchronous callback before continuing for exchangeId: {} -> {}",
@@ -77,13 +89,47 @@ public class DefaultAsyncProcessorAwaitManager extends ServiceSupport implements
}
@Override
+ public void interrupt(String exchangeId) {
+ // need to find the exchange with the given exchange id
+ Exchange found = null;
+ for (AsyncProcessorAwaitManager.AwaitThread entry : browse()) {
+ Exchange exchange = entry.getExchange();
+ if (exchangeId.equals(exchange.getExchangeId())) {
+ found = exchange;
+ break;
+ }
+ }
+
+ if (found != null) {
+ interrupt(found);
+ }
+ }
+
+ @Override
public void interrupt(Exchange exchange) {
- AwaitThreadEntry latch = (AwaitThreadEntry) inflight.get(exchange);
- if (latch != null) {
- LOG.warn("Interrupted while waiting for asynchronous callback, will continue routing exchangeId: {} -> {}",
- exchange.getExchangeId(), exchange);
- exchange.setException(new RejectedExecutionException("Interrupted while waiting for asynchronous callback"));
- latch.getLatch().countDown();
+ AwaitThreadEntry entry = (AwaitThreadEntry) inflight.get(exchange);
+ if (entry != null) {
+ try {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Interrupted while waiting for asynchronous callback, will release the following blocked thread which was waiting for exchange to finish processing with exchangeId: ");
+ sb.append(exchange.getExchangeId());
+ sb.append("\n");
+
+ sb.append(dumpBlockedThread(entry));
+
+ // dump a route stack trace of the exchange
+ String routeStackTrace = MessageHelper.dumpMessageHistoryStacktrace(exchange, exchangeFormatter, false);
+ if (routeStackTrace != null) {
+ sb.append(routeStackTrace);
+ }
+ LOG.warn(sb.toString());
+
+ } catch (Exception e) {
+ throw ObjectHelper.wrapRuntimeCamelException(e);
+ } finally {
+ exchange.setException(new RejectedExecutionException("Interrupted while waiting for asynchronous callback for exchangeId: " + exchange.getExchangeId()));
+ entry.getLatch().countDown();
+ }
}
}
@@ -109,9 +155,7 @@ public class DefaultAsyncProcessorAwaitManager extends ServiceSupport implements
StringBuilder sb = new StringBuilder();
for (AwaitThread entry : threads) {
- sb.append("\tBlocked thread: ").append(entry.getBlockedThread().getName())
- .append(", exchangeId=").append(entry.getExchange().getExchangeId())
- .append(", duration=").append(entry.getWaitDuration()).append(" msec.");
+ sb.append(dumpBlockedThread(entry));
}
if (isInterruptThreadsWhileStopping()) {
@@ -133,17 +177,50 @@ public class DefaultAsyncProcessorAwaitManager extends ServiceSupport implements
inflight.clear();
}
+ private static String dumpBlockedThread(AwaitThread entry) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("\n");
+ sb.append("Blocked Thread\n");
+ sb.append("---------------------------------------------------------------------------------------------------------------------------------------\n");
+
+ sb.append(style("Id:")).append(entry.getBlockedThread().getId()).append("\n");
+ sb.append(style("Name:")).append(entry.getBlockedThread().getName()).append("\n");
+ sb.append(style("RouteId:")).append(safeNull(entry.getRouteId())).append("\n");
+ sb.append(style("NodeId:")).append(safeNull(entry.getNodeId())).append("\n");
+ sb.append(style("Duration:")).append(entry.getWaitDuration()).append(" msec.\n");
+ return sb.toString();
+ }
+
+ private static String style(String label) {
+ return String.format("\t%-20s", label);
+ }
+
+ private static String safeNull(Object value) {
+ return value != null ? value.toString() : "";
+ }
+
private static final class AwaitThreadEntry implements AwaitThread {
private final Thread thread;
private final Exchange exchange;
private final CountDownLatch latch;
private final long start;
+ private String routeId;
+ private String nodeId;
private AwaitThreadEntry(Thread thread, Exchange exchange, CountDownLatch latch) {
this.thread = thread;
this.exchange = exchange;
this.latch = latch;
this.start = System.currentTimeMillis();
+
+ // capture details from message history if enabled
+ List<MessageHistory> list = exchange.getProperty(Exchange.MESSAGE_HISTORY, List.class);
+ if (list != null && !list.isEmpty()) {
+ // grab last part
+ MessageHistory history = list.get(list.size() - 1);
+ routeId = history.getRouteId();
+ nodeId = history.getNode() != null ? history.getNode().getId() : null;
+ }
}
@Override
@@ -161,9 +238,24 @@ public class DefaultAsyncProcessorAwaitManager extends ServiceSupport implements
return System.currentTimeMillis() - start;
}
+ @Override
+ public String getRouteId() {
+ return routeId;
+ }
+
+ @Override
+ public String getNodeId() {
+ return nodeId;
+ }
+
public CountDownLatch getLatch() {
return latch;
}
+
+ @Override
+ public String toString() {
+ return "AwaitThreadEntry[name=" + thread.getName() + ", exchangeId=" + exchange.getExchangeId() + "]";
+ }
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/36c2d70d/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedAsyncProcessorAwaitManager.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedAsyncProcessorAwaitManager.java b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedAsyncProcessorAwaitManager.java
index bd32b4e..a4759ef 100644
--- a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedAsyncProcessorAwaitManager.java
+++ b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedAsyncProcessorAwaitManager.java
@@ -24,7 +24,6 @@ import javax.management.openmbean.TabularData;
import javax.management.openmbean.TabularDataSupport;
import org.apache.camel.CamelContext;
-import org.apache.camel.Exchange;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.api.management.mbean.CamelOpenMBeanTypes;
import org.apache.camel.api.management.mbean.ManagedAsyncProcessorAwaitManagerMBean;
@@ -70,13 +69,16 @@ public class ManagedAsyncProcessorAwaitManager extends ManagedService implements
Collection<AsyncProcessorAwaitManager.AwaitThread> threads = manager.browse();
for (AsyncProcessorAwaitManager.AwaitThread entry : threads) {
CompositeType ct = CamelOpenMBeanTypes.listAwaitThreadsCompositeType();
+ String id = "" + entry.getBlockedThread().getId();
String name = entry.getBlockedThread().getName();
String exchangeId = entry.getExchange().getExchangeId();
+ String routeId = entry.getRouteId();
+ String nodeId = entry.getNodeId();
String duration = "" + entry.getWaitDuration();
- CompositeData data = new CompositeDataSupport(ct, new String[]
- {"name", "exchangeId", "duration"},
- new Object[]{name, exchangeId, duration});
+ CompositeData data = new CompositeDataSupport(ct,
+ new String[]{"id", "name", "exchangeId", "routeId", "nodeId", "duration"},
+ new Object[]{id, name, exchangeId, routeId, nodeId, duration});
answer.put(data);
}
return answer;
@@ -87,19 +89,7 @@ public class ManagedAsyncProcessorAwaitManager extends ManagedService implements
@Override
public void interrupt(String exchangeId) {
- // need to find the exchange with the given exchange id
- Exchange found = null;
- for (AsyncProcessorAwaitManager.AwaitThread entry : manager.browse()) {
- Exchange exchange = entry.getExchange();
- if (exchangeId.equals(exchange.getExchangeId())) {
- found = exchange;
- break;
- }
- }
-
- if (found != null) {
- manager.interrupt(found);
- }
+ manager.interrupt(exchangeId);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/36c2d70d/camel-core/src/main/java/org/apache/camel/spi/AsyncProcessorAwaitManager.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/spi/AsyncProcessorAwaitManager.java b/camel-core/src/main/java/org/apache/camel/spi/AsyncProcessorAwaitManager.java
index 7780e3f..0b39c1c 100644
--- a/camel-core/src/main/java/org/apache/camel/spi/AsyncProcessorAwaitManager.java
+++ b/camel-core/src/main/java/org/apache/camel/spi/AsyncProcessorAwaitManager.java
@@ -50,6 +50,21 @@ public interface AsyncProcessorAwaitManager extends StaticService {
* Time in millis the thread has been blocked waiting for the signal.
*/
long getWaitDuration();
+
+ /**
+ * The id of the route where the exchange was processed when the thread was set to block.
+ * <p/>
+ * Is <tt>null</tt> if message history is disabled.
+ */
+ String getRouteId();
+
+ /**
+ * The id of the node from the route where the exchange was processed when the thread was set to block.
+ * <p/>
+ * Is <tt>null</tt> if message history is disabled.
+ */
+ String getNodeId();
+
}
/**
@@ -90,6 +105,19 @@ public interface AsyncProcessorAwaitManager extends StaticService {
* so the blocked thread can continue. An exception is set on the exchange which allows Camel's error handler to deal
* with this malfunctioned exchange.
*
+ * @param exchangeId the exchange id to interrupt.
+ */
+ void interrupt(String exchangeId);
+
+ /**
+ * To interrupt an exchange which may seem as stuck, to force the exchange to continue,
+ * allowing any blocking thread to be released.
+ * <p/>
+ * <b>Important:</b> Use this with caution as the other thread is still assumed to be process the exchange. Though
+ * if it appears as the exchange is <i>stuck</i>, then this method can remedy this, by forcing the latch to count-down
+ * so the blocked thread can continue. An exception is set on the exchange which allows Camel's error handler to deal
+ * with this malfunctioned exchange.
+ *
* @param exchange the exchange to interrupt.
*/
void interrupt(Exchange exchange);
http://git-wip-us.apache.org/repos/asf/camel/blob/36c2d70d/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptTest.java b/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptTest.java
new file mode 100644
index 0000000..d78f7f1
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerInterruptTest.java
@@ -0,0 +1,86 @@
+/**
+ * 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.processor.async;
+
+import java.util.Collection;
+import java.util.concurrent.RejectedExecutionException;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.AsyncProcessorAwaitManager;
+
+/**
+ * @version
+ */
+public class AsyncProcessorAwaitManagerInterruptTest extends ContextTestSupport {
+
+ public void testAsyncAwaitInterrupt() throws Exception {
+ assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+
+ getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
+ getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel");
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ try {
+ template.requestBody("direct:start", "Hello Camel", String.class);
+ fail("Should have thrown exception");
+ } catch (CamelExecutionException e) {
+ RejectedExecutionException cause = assertIsInstanceOf(RejectedExecutionException.class, e.getCause());
+ assertTrue(cause.getMessage().startsWith("Interrupted while waiting for asynchronous callback"));
+ }
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:start").routeId("myRoute")
+ .to("mock:before")
+ .to("async:bye:camel").id("myAsync")
+ .to("mock:after")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ int size = context.getAsyncProcessorAwaitManager().size();
+ log.info("async inflight: {}", size);
+ assertEquals(1, size);
+
+ Collection<AsyncProcessorAwaitManager.AwaitThread> threads = context.getAsyncProcessorAwaitManager().browse();
+ AsyncProcessorAwaitManager.AwaitThread thread = threads.iterator().next();
+
+ // lets interrupt it
+ String id = thread.getExchange().getExchangeId();
+ context.getAsyncProcessorAwaitManager().interrupt(id);
+ }
+ })
+ .transform(constant("Hi Camel"))
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/36c2d70d/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerTest.java b/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerTest.java
new file mode 100644
index 0000000..8389069
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/async/AsyncProcessorAwaitManagerTest.java
@@ -0,0 +1,80 @@
+/**
+ * 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.processor.async;
+
+import java.util.Collection;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.AsyncProcessorAwaitManager;
+
+/**
+ * @version
+ */
+public class AsyncProcessorAwaitManagerTest extends ContextTestSupport {
+
+ public void testAsyncAwait() throws Exception {
+ assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+
+ getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
+ getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel");
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
+
+ String reply = template.requestBody("direct:start", "Hello Camel", String.class);
+ assertEquals("Bye Camel", reply);
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(0, context.getAsyncProcessorAwaitManager().size());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:start").routeId("myRoute")
+ .to("mock:before")
+ .to("async:bye:camel").id("myAsync")
+ .to("mock:after")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ int size = context.getAsyncProcessorAwaitManager().size();
+ log.info("async inflight: {}", size);
+ assertEquals(1, size);
+
+ Collection<AsyncProcessorAwaitManager.AwaitThread> threads = context.getAsyncProcessorAwaitManager().browse();
+ AsyncProcessorAwaitManager.AwaitThread thread = threads.iterator().next();
+
+ long wait = thread.getWaitDuration();
+ log.info("Thread {} has waited for {} msec.", thread.getBlockedThread().getName(), wait);
+
+ assertEquals("myRoute", thread.getRouteId());
+ assertEquals("myAsync", thread.getNodeId());
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file