Updated Branches: refs/heads/master 09ceee600 -> ea3d7dc8e
CAMEL-6407: Include message history in stacktraces from error handler, making it easier to know where the problem was. Work in progress. Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/ca007a55 Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/ca007a55 Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/ca007a55 Branch: refs/heads/master Commit: ca007a55f5274ade7ee517ba69f7dea8b01a2a7c Parents: 09ceee6 Author: Claus Ibsen <davscl...@apache.org> Authored: Fri May 31 08:31:48 2013 +0200 Committer: Claus Ibsen <davscl...@apache.org> Committed: Fri May 31 08:31:59 2013 +0200 ---------------------------------------------------------------------- .../org/apache/camel/RuntimeConfiguration.java | 14 +++++ .../org/apache/camel/impl/DefaultCamelContext.java | 9 +++ .../org/apache/camel/impl/DefaultRouteContext.java | 14 +++++ .../apache/camel/model/OnExceptionDefinition.java | 16 ++++++ .../camel/model/ProcessorDefinitionHelper.java | 35 +++++++++++++ .../camel/model/RedeliveryPolicyDefinition.java | 37 +++++++++++++ .../org/apache/camel/model/RouteDefinition.java | 40 +++++++++++++++ .../camel/processor/CamelInternalProcessor.java | 27 ++++++++++ .../camel/processor/RedeliveryErrorHandler.java | 27 +++++++++- .../apache/camel/processor/RedeliveryPolicy.java | 22 ++++++++ .../processor/interceptor/DefaultChannel.java | 5 ++ 11 files changed, 243 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/RuntimeConfiguration.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/RuntimeConfiguration.java b/camel-core/src/main/java/org/apache/camel/RuntimeConfiguration.java index 96afdc0..7da882a 100644 --- a/camel-core/src/main/java/org/apache/camel/RuntimeConfiguration.java +++ b/camel-core/src/main/java/org/apache/camel/RuntimeConfiguration.java @@ -53,6 +53,20 @@ public interface RuntimeConfiguration { Boolean isTracing(); /** + * Sets whether message history is enabled or not (default is disabled). + * + * @param messageHistory whether message history is enabled + */ + void setMessageHistory(Boolean messageHistory); + + /** + * Returns whether stream cache is enabled + * + * @return <tt>true</tt> if stream cache is enabled + */ + Boolean isMessageHistory(); + + /** * Sets whether fault handling is enabled or not (default is disabled). * * @param handleFault whether to enable fault handling. http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java index 4067efb..33f45c5 100644 --- a/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java +++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java @@ -173,6 +173,7 @@ public class DefaultCamelContext extends ServiceSupport implements ModelCamelCon private final ThreadLocal<Boolean> isStartingRoutes = new ThreadLocal<Boolean>(); private Boolean autoStartup = Boolean.TRUE; private Boolean trace = Boolean.FALSE; + private Boolean messageHistory = Boolean.FALSE; private Boolean streamCache = Boolean.FALSE; private Boolean handleFault = Boolean.FALSE; private Boolean disableJMX = Boolean.FALSE; @@ -1290,6 +1291,14 @@ public class DefaultCamelContext extends ServiceSupport implements ModelCamelCon return trace; } + public Boolean isMessageHistory() { + return messageHistory; + } + + public void setMessageHistory(Boolean messageHistory) { + this.messageHistory = messageHistory; + } + public Boolean isHandleFault() { return handleFault; } http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java index 40b5415..2680e84 100644 --- a/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java +++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java @@ -58,6 +58,7 @@ public class DefaultRouteContext implements RouteContext { private InterceptStrategy managedInterceptStrategy; private boolean routeAdded; private Boolean trace; + private Boolean messageHistory; private Boolean streamCache; private Boolean handleFault; private Long delay; @@ -251,6 +252,19 @@ public class DefaultRouteContext implements RouteContext { } } + public void setMessageHistory(Boolean messageHistory) { + this.messageHistory = messageHistory; + } + + public Boolean isMessageHistory() { + if (messageHistory != null) { + return messageHistory; + } else { + // fallback to the option from camel context + return getCamelContext().isMessageHistory(); + } + } + public void setStreamCaching(Boolean cache) { this.streamCache = cache; } http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/model/OnExceptionDefinition.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/model/OnExceptionDefinition.java b/camel-core/src/main/java/org/apache/camel/model/OnExceptionDefinition.java index 622f856..bff027f 100644 --- a/camel-core/src/main/java/org/apache/camel/model/OnExceptionDefinition.java +++ b/camel-core/src/main/java/org/apache/camel/model/OnExceptionDefinition.java @@ -575,6 +575,22 @@ public class OnExceptionDefinition extends ProcessorDefinition<OnExceptionDefini } /** + * Sets whether to log exhausted exceptions with message history + */ + public OnExceptionDefinition logExhaustedMessageHistory(boolean logExhaustedMessageHistory) { + getOrCreateRedeliveryPolicy().logExhaustedMessageHistory(logExhaustedMessageHistory); + return this; + } + + /** + * Sets whether to log exhausted exceptions with message history + */ + public OnExceptionDefinition logExhaustedMessageHistory(String logExhaustedMessageHistory) { + getOrCreateRedeliveryPolicy().logExhaustedMessageHistory(logExhaustedMessageHistory); + return this; + } + + /** * Sets the maximum redeliveries * <ul> * <li>5 = default value</li> http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java index 6395ec6..4e1d153 100644 --- a/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java +++ b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java @@ -24,6 +24,7 @@ import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; +import org.apache.camel.Exchange; import org.apache.camel.spi.ExecutorServiceManager; import org.apache.camel.spi.RouteContext; import org.apache.camel.util.ObjectHelper; @@ -466,4 +467,38 @@ public final class ProcessorDefinitionHelper { return null; } + @SuppressWarnings("unchecked") + public static String dumpMessageHistoryStacktrace(Exchange exchange, boolean logStackTrace) { + List<ProcessorDefinition<?>> list = exchange.getProperty("CamelMessageHistory", List.class); + if (list == null || list.isEmpty()) { + return null; + } + + final String HEADER_FORMAT = "%-20s %-20s %-80s"; + final String OUTPUT_FORMAT = "[%-18.18s] [%-18.18s] [%-78.78s]"; + + StringBuilder sb = new StringBuilder(); + sb.append("\n"); + sb.append("Message History\n"); + sb.append("--------------------------------------------------------------------------------------------------------------------------\n"); + sb.append(String.format(HEADER_FORMAT, "RouteId", "ProcessorId", "Processor")); + sb.append("\n"); + + for (ProcessorDefinition<?> node : list) { + + String routeId = getRouteId(node); + String id = node.getId(); + String label = node.getLabel(); + + sb.append(String.format(OUTPUT_FORMAT, routeId, id, label)); + sb.append("\n"); + } + + if (logStackTrace) { + sb.append("\nStacktrace\n"); + sb.append("--------------------------------------------------------------------------------------------------------------------------"); + } + return sb.toString(); + } + } http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/model/RedeliveryPolicyDefinition.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/model/RedeliveryPolicyDefinition.java b/camel-core/src/main/java/org/apache/camel/model/RedeliveryPolicyDefinition.java index 475c629..1a2f878 100644 --- a/camel-core/src/main/java/org/apache/camel/model/RedeliveryPolicyDefinition.java +++ b/camel-core/src/main/java/org/apache/camel/model/RedeliveryPolicyDefinition.java @@ -68,6 +68,8 @@ public class RedeliveryPolicyDefinition { @XmlAttribute private String logExhausted; @XmlAttribute + private String logExhaustedMessageHistory; + @XmlAttribute private String disableRedelivery; @XmlAttribute private String delayPattern; @@ -136,6 +138,9 @@ public class RedeliveryPolicyDefinition { if (logExhausted != null) { answer.setLogExhausted(CamelContextHelper.parseBoolean(context, logExhausted)); } + if (logExhaustedMessageHistory != null) { + answer.setLogExhaustedMessageHistory(CamelContextHelper.parseBoolean(context, logExhaustedMessageHistory)); + } if (disableRedelivery != null) { if (CamelContextHelper.parseBoolean(context, disableRedelivery)) { answer.setMaximumRedeliveries(0); @@ -428,6 +433,30 @@ public class RedeliveryPolicyDefinition { } /** + * Sets whether exhausted exceptions should be logged including message history or not (supports property placeholders). + * Can be used to include or reduce verbose. + * + * @param logExhaustedMessageHistory whether exhausted exceptions should be logged with message history + * @return the builder + */ + public RedeliveryPolicyDefinition logExhaustedMessageHistory(boolean logExhaustedMessageHistory) { + setLogExhaustedMessageHistory(Boolean.toString(logExhaustedMessageHistory)); + return this; + } + + /** + * Sets whether exhausted exceptions should be logged including message history or not (supports property placeholders). + * Can be used to include or reduce verbose. + * + * @param logExhaustedMessageHistory whether exhausted exceptions should be logged with message history + * @return the builder + */ + public RedeliveryPolicyDefinition logExhaustedMessageHistory(String logExhaustedMessageHistory) { + setLogExhaustedMessageHistory(logExhaustedMessageHistory); + return this; + } + + /** * Sets the maximum redeliveries * <ul> * <li>x = redeliver at most x times</li> @@ -653,6 +682,14 @@ public class RedeliveryPolicyDefinition { this.logExhausted = logExhausted; } + public String getLogExhaustedMessageHistory() { + return logExhaustedMessageHistory; + } + + public void setLogExhaustedMessageHistory(String logExhaustedMessageHistory) { + this.logExhaustedMessageHistory = logExhaustedMessageHistory; + } + public String getDisableRedelivery() { return disableRedelivery; } http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java b/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java index 6e86431..d4af9b3 100644 --- a/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java +++ b/camel-core/src/main/java/org/apache/camel/model/RouteDefinition.java @@ -68,6 +68,7 @@ public class RouteDefinition extends ProcessorDefinition<RouteDefinition> { private String group; private String streamCache; private String trace; + private String messageHistory; private String handleFault; private String delayer; private String autoStartup; @@ -402,6 +403,26 @@ public class RouteDefinition extends ProcessorDefinition<RouteDefinition> { } /** + * Enable message history for this route. + * + * @return the builder + */ + public RouteDefinition messageHistory() { + setMessageHistory("true"); + return this; + } + + /** + * Disable message history for this route. + * + * @return the builder + */ + public RouteDefinition noMessageHistory() { + setMessageHistory("false"); + return this; + } + + /** * Disable handle fault for this route. * * @return the builder @@ -615,6 +636,14 @@ public class RouteDefinition extends ProcessorDefinition<RouteDefinition> { this.trace = trace; } + public String getMessageHistory() { + return messageHistory; + } + + public void setMessageHistory(String messageHistory) { + this.messageHistory = messageHistory; + } + public String getHandleFault() { return handleFault; } @@ -777,6 +806,17 @@ public class RouteDefinition extends ProcessorDefinition<RouteDefinition> { } } + // configure message history + if (messageHistory != null) { + Boolean isMessageHistory = CamelContextHelper.parseBoolean(camelContext, getMessageHistory()); + if (isMessageHistory != null) { + routeContext.setMessageHistory(isMessageHistory); + if (isMessageHistory) { + log.debug("Message history is enabled on route: {}", getId()); + } + } + } + // configure stream caching if (streamCache != null) { Boolean isStreamCache = CamelContextHelper.parseBoolean(camelContext, getStreamCache()); http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java index 217c8d7..6424099 100644 --- a/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java +++ b/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java @@ -681,4 +681,31 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor { } } + public static class MessageHistoryAdvice implements CamelInternalProcessorAdvice { + + private final ProcessorDefinition<?> definition; + + public MessageHistoryAdvice(ProcessorDefinition<?> definition) { + this.definition = definition; + } + + @Override + @SuppressWarnings("unchecked") + public Object before(Exchange exchange) throws Exception { + List<ProcessorDefinition<?>> history = exchange.getProperty("CamelMessageHistory", List.class); + if (history == null) { + history = new ArrayList<ProcessorDefinition<?>>(); + exchange.setProperty("CamelMessageHistory", history); + } + history.add(definition); + return null; + } + + @Override + public void after(Exchange exchange, Object data) throws Exception { + // noop + } + + } + } http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java b/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java index 19acacb..e038c66 100644 --- a/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java +++ b/camel-core/src/main/java/org/apache/camel/processor/RedeliveryErrorHandler.java @@ -30,6 +30,7 @@ import org.apache.camel.Message; import org.apache.camel.Predicate; import org.apache.camel.Processor; import org.apache.camel.model.OnExceptionDefinition; +import org.apache.camel.model.ProcessorDefinitionHelper; import org.apache.camel.spi.ShutdownPrepared; import org.apache.camel.spi.SubUnitOfWorkCallback; import org.apache.camel.spi.UnitOfWork; @@ -955,6 +956,15 @@ public abstract class RedeliveryErrorHandler extends ErrorHandlerSupport impleme if (cause != null) { msg = msg + " due: " + cause.getMessage(); } + + // should we include message history + if (!shouldRedeliver && data.currentRedeliveryPolicy.isLogExhaustedMessageHistory()) { + String routeStackTrace = ProcessorDefinitionHelper.dumpMessageHistoryStacktrace(exchange, false); + if (routeStackTrace != null) { + msg = msg + "\n" + routeStackTrace; + } + } + if (newLogLevel == LoggingLevel.ERROR) { // log intended rollback on maximum WARN level (no ERROR) logger.log(msg, LoggingLevel.WARN); @@ -962,10 +972,21 @@ public abstract class RedeliveryErrorHandler extends ErrorHandlerSupport impleme // otherwise use the desired logging level logger.log(msg, newLogLevel); } - } else if (e != null && logStackTrace) { - logger.log(message, e, newLogLevel); } else { - logger.log(message, newLogLevel); + String msg = message; + // should we include message history + if (!shouldRedeliver && data.currentRedeliveryPolicy.isLogExhaustedMessageHistory()) { + String routeStackTrace = ProcessorDefinitionHelper.dumpMessageHistoryStacktrace(exchange, e != null && logStackTrace); + if (routeStackTrace != null) { + msg = msg + "\n" + routeStackTrace; + } + } + + if (e != null && logStackTrace) { + logger.log(msg, e, newLogLevel); + } else { + logger.log(msg, newLogLevel); + } } } http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java b/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java index 279524a..1242168 100644 --- a/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java +++ b/camel-core/src/main/java/org/apache/camel/processor/RedeliveryPolicy.java @@ -49,6 +49,7 @@ import org.slf4j.LoggerFactory; * <li>logStackTrace = true</li> * <li>logHandled = false</li> * <li>logExhausted = true</li> + * <li>logExhaustedMessageHistory = true</li> * </ul> * <p/> * Setting the maximumRedeliveries to a negative value such as -1 will then always redeliver (unlimited). @@ -93,6 +94,7 @@ public class RedeliveryPolicy implements Cloneable, Serializable { protected boolean logHandled; protected boolean logContinued; protected boolean logExhausted = true; + protected boolean logExhaustedMessageHistory = true; protected boolean logRetryAttempted = true; protected String delayPattern; protected boolean asyncDelayedRedelivery; @@ -116,6 +118,7 @@ public class RedeliveryPolicy implements Cloneable, Serializable { + ", logHandled=" + logHandled + ", logContinued=" + logContinued + ", logExhausted=" + logExhausted + + ", logExhaustedMessageHistory=" + logExhaustedMessageHistory + ", useExponentialBackOff=" + useExponentialBackOff + ", backOffMultiplier=" + backOffMultiplier + ", useCollisionAvoidance=" + useCollisionAvoidance @@ -380,6 +383,14 @@ public class RedeliveryPolicy implements Cloneable, Serializable { } /** + * Sets whether to log exhausted errors including message history + */ + public RedeliveryPolicy logExhaustedMessageHistory(boolean logExhaustedMessageHistory) { + setLogExhaustedMessageHistory(logExhaustedMessageHistory); + return this; + } + + /** * Sets the delay pattern with delay intervals. */ public RedeliveryPolicy delayPattern(String delayPattern) { @@ -639,6 +650,17 @@ public class RedeliveryPolicy implements Cloneable, Serializable { this.logExhausted = logExhausted; } + public boolean isLogExhaustedMessageHistory() { + return logExhaustedMessageHistory; + } + + /** + * Sets whether exhausted exceptions should be logged with message history included. + */ + public void setLogExhaustedMessageHistory(boolean logExhaustedMessageHistory) { + this.logExhaustedMessageHistory = logExhaustedMessageHistory; + } + public boolean isAsyncDelayedRedelivery() { return asyncDelayedRedelivery; } http://git-wip-us.apache.org/repos/asf/camel/blob/ca007a55/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java b/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java index 725a3bd..f9d8c10 100644 --- a/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java +++ b/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java @@ -220,6 +220,11 @@ public class DefaultChannel extends CamelInternalProcessor implements ModelChann } } + if (routeContext.isMessageHistory()) { + // add message history advice + addAdvice(new MessageHistoryAdvice(targetOutputDef)); + } + // the regular tracer is not a task on internalProcessor as this is not really needed // end users have to explicit enable the tracer to use it, and then its okay if we wrap // the processors (but by default tracer is disabled, and therefore we do not wrap processors)
