You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2022/04/11 13:37:56 UTC
[camel] branch main updated: CAMEL-17762: allow adjusting log level for exception messages during resume updates
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 2c9ff6ec87f CAMEL-17762: allow adjusting log level for exception messages during resume updates
2c9ff6ec87f is described below
commit 2c9ff6ec87f7995abf485ee59d113ba541002b8d
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Mon Apr 11 14:41:19 2022 +0200
CAMEL-17762: allow adjusting log level for exception messages during resume updates
---
.../apache/camel/catalog/schemas/camel-spring.xsd | 4 ++-
.../KafkaConsumerWithResumeRouteStrategyIT.java | 7 +++--
.../src/test/resources/log4j2.properties | 4 +++
.../apache/camel/model/ResumableDefinition.java | 30 ++++++++++++++++++
.../processor/resume/ResumableCompletion.java | 36 ++++++++++++++++++++--
.../camel/processor/resume/ResumableProcessor.java | 7 +++--
.../org/apache/camel/reifier/ResumableReifier.java | 15 ++++++++-
.../java/org/apache/camel/xml/in/ModelParser.java | 8 ++++-
8 files changed, 101 insertions(+), 10 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
index 10a5a508e54..94a28c4d2f4 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
@@ -10416,7 +10416,9 @@ Reference to the rest-dsl.
<xs:complexType name="resumableDefinition">
<xs:complexContent>
<xs:extension base="tns:noOutputDefinition">
- <xs:sequence/>
+ <xs:sequence>
+ <xs:element minOccurs="0" name="loggingLevel" type="xs:string"/>
+ </xs:sequence>
<xs:attribute name="resumeStrategy" type="xs:string" use="required">
<xs:annotation>
<xs:documentation xml:lang="en"><![CDATA[
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerWithResumeRouteStrategyIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerWithResumeRouteStrategyIT.java
index 02bccecfb5a..181009da26c 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerWithResumeRouteStrategyIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerWithResumeRouteStrategyIT.java
@@ -222,10 +222,11 @@ public class KafkaConsumerWithResumeRouteStrategyIT extends BaseEmbeddedKafkaTes
public void configure() {
from("kafka:" + TOPIC + "?groupId=" + TOPIC + "_GROUP&autoCommitIntervalMs=1000"
+ "&autoOffsetReset=earliest&consumersCount=1")
+ .resumable().resumeStrategy("resumeStrategy", "DEBUG")
.routeId("resume-strategy-route")
- .setHeader(Exchange.OFFSET,
- constant(Resumables.of("key", RANDOM_VALUE)))
- .resumable().resumeStrategy("resumeStrategy")
+ .setHeader(Exchange.OFFSET, constant(Resumables.of("key", RANDOM_VALUE)))
+ // Note: this is for manually testing the ResumableCompletion onFailure exception logging. Uncomment it for testing it
+ // .process(e -> e.setException(new RuntimeCamelException("Mock error in test")))
.to("mock:result");
}
};
diff --git a/components/camel-kafka/src/test/resources/log4j2.properties b/components/camel-kafka/src/test/resources/log4j2.properties
index 72b19cf9e80..61f9e8e29b4 100644
--- a/components/camel-kafka/src/test/resources/log4j2.properties
+++ b/components/camel-kafka/src/test/resources/log4j2.properties
@@ -43,3 +43,7 @@ logger.resume.level=INFO
logger.kafka.name=org.apache.kafka
logger.kafka.level=INFO
+# To verify the ResumableCompletion onFailure
+# logger.resume-processor.name=org.apache.camel.processor.resume
+# logger.resume-processor.level=DEBUG
+
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/ResumableDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/ResumableDefinition.java
index 63ed0f198bb..4e2d62deb86 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/ResumableDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/ResumableDefinition.java
@@ -40,6 +40,10 @@ public class ResumableDefinition extends NoOutputDefinition<ResumableDefinition>
@Metadata(required = true, javaType = "org.apache.camel.ResumeStrategy")
private String resumeStrategy;
+ @Metadata(label = "advanced", javaType = "org.apache.camel.LoggingLevel", defaultValue = "ERROR",
+ enums = "TRACE,DEBUG,INFO,WARN,ERROR,OFF")
+ private String loggingLevel;
+
@Override
public String getShortName() {
return "resumable";
@@ -66,6 +70,14 @@ public class ResumableDefinition extends NoOutputDefinition<ResumableDefinition>
this.resumeStrategyBean = resumeStrategyBean;
}
+ public String getLoggingLevel() {
+ return loggingLevel;
+ }
+
+ public void setLoggingLevel(String loggingLevelRef) {
+ this.loggingLevel = loggingLevelRef;
+ }
+
// Fluent API
// -------------------------------------------------------------------------
@@ -77,6 +89,15 @@ public class ResumableDefinition extends NoOutputDefinition<ResumableDefinition>
return this;
}
+ /**
+ * Sets the resume strategy to use
+ */
+ public ResumableDefinition resumeStrategy(String resumeStrategyRef, String loggingLevelRef) {
+ setResumeStrategy(resumeStrategyRef);
+ setLoggingLevel(loggingLevelRef);
+ return this;
+ }
+
/**
* Sets the resume strategy to use
*/
@@ -84,4 +105,13 @@ public class ResumableDefinition extends NoOutputDefinition<ResumableDefinition>
setResumeStrategy(resumeStrategy);
return this;
}
+
+ /**
+ * Sets the resume strategy to use
+ */
+ public ResumableDefinition resumeStrategy(ResumeStrategy resumeStrategy, String loggingLevelRef) {
+ setResumeStrategy(resumeStrategy);
+ setLoggingLevel(loggingLevelRef);
+ return this;
+ }
}
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/ResumableCompletion.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/ResumableCompletion.java
index f80f71edde4..a09d04591e6 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/ResumableCompletion.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/ResumableCompletion.java
@@ -18,9 +18,11 @@
package org.apache.camel.processor.resume;
import org.apache.camel.Exchange;
+import org.apache.camel.LoggingLevel;
import org.apache.camel.Resumable;
import org.apache.camel.ResumeStrategy;
import org.apache.camel.UpdatableConsumerResumeStrategy;
+import org.apache.camel.spi.CamelLogger;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.support.ExchangeHelper;
import org.slf4j.Logger;
@@ -30,9 +32,11 @@ public class ResumableCompletion implements Synchronization {
private static final Logger LOG = LoggerFactory.getLogger(ResumableCompletion.class);
private final ResumeStrategy resumeStrategy;
+ private final LoggingLevel loggingLevel;
- public ResumableCompletion(ResumeStrategy resumeStrategy) {
+ public ResumableCompletion(ResumeStrategy resumeStrategy, LoggingLevel loggingLevel) {
this.resumeStrategy = resumeStrategy;
+ this.loggingLevel = loggingLevel;
}
@Override
@@ -70,6 +74,34 @@ public class ResumableCompletion implements Synchronization {
@Override
public void onFailure(Exchange exchange) {
- LOG.warn("Skipping offset update for due to failure in processing");
+ Exception e = exchange.getException();
+ Object offset = exchange.getMessage().getHeader(Exchange.OFFSET);
+
+ if (offset instanceof Resumable) {
+ Resumable<?, ?> resumable = (Resumable<?, ?>) offset;
+
+ String logMessage = String.format(
+ "Skipping offset update with address '%s' and offset value '%s' due to failure in processing: %s",
+ resumable.getAddressable(), resumable.getLastOffset().offset(), e.getMessage());
+
+ if (LOG.isDebugEnabled() || CamelLogger.shouldLog(LOG, loggingLevel)) {
+ CamelLogger.log(LOG, LoggingLevel.DEBUG, logMessage, e);
+ } else {
+ logMessage += " (stacktrace available in DEBUG logging level)";
+
+ CamelLogger.log(LOG, loggingLevel, logMessage);
+ }
+ } else {
+ String logMessage = String.format("Skipping offset update of '%s' due to failure in processing: %s",
+ offset == null ? "type null" : "unspecified type", e.getMessage());
+
+ if (LOG.isDebugEnabled() || CamelLogger.shouldLog(LOG, loggingLevel)) {
+ CamelLogger.log(LOG, LoggingLevel.DEBUG, logMessage, e);
+ } else {
+ logMessage += " (stacktrace available in DEBUG logging level)";
+
+ CamelLogger.log(LOG, loggingLevel, logMessage);
+ }
+ }
}
}
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/ResumableProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/ResumableProcessor.java
index 422161481b7..240376d5acf 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/ResumableProcessor.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/ResumableProcessor.java
@@ -26,6 +26,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.Exchange;
import org.apache.camel.ExtendedExchange;
+import org.apache.camel.LoggingLevel;
import org.apache.camel.Navigate;
import org.apache.camel.Processor;
import org.apache.camel.ResumeStrategy;
@@ -48,12 +49,14 @@ public class ResumableProcessor extends AsyncProcessorSupport
private CamelContext camelContext;
private final ResumeStrategy resumeStrategy;
private final AsyncProcessor processor;
+ private final LoggingLevel loggingLevel;
private String id;
private String routeId;
- public ResumableProcessor(ResumeStrategy resumeStrategy, Processor processor) {
+ public ResumableProcessor(ResumeStrategy resumeStrategy, Processor processor, LoggingLevel loggingLevel) {
this.resumeStrategy = Objects.requireNonNull(resumeStrategy);
this.processor = AsyncProcessorConverterHelper.convert(processor);
+ this.loggingLevel = loggingLevel;
}
@Override
@@ -66,7 +69,7 @@ public class ResumableProcessor extends AsyncProcessorSupport
@Override
public boolean process(final Exchange exchange, final AsyncCallback callback) {
- final Synchronization onCompletion = new ResumableCompletion(resumeStrategy);
+ final Synchronization onCompletion = new ResumableCompletion(resumeStrategy, loggingLevel);
exchange.adapt(ExtendedExchange.class).addOnCompletion(onCompletion);
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/ResumableReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/ResumableReifier.java
index 08c3c7943b8..c96b14a941c 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/ResumableReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/ResumableReifier.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.reifier;
+import org.apache.camel.LoggingLevel;
import org.apache.camel.Processor;
import org.apache.camel.ResumeStrategy;
import org.apache.camel.Route;
@@ -38,7 +39,8 @@ public class ResumableReifier extends ProcessorReifier<ResumableDefinition> {
ObjectHelper.notNull(resumeStrategy, "resumeStrategy", definition);
route.setResumeStrategy(resumeStrategy);
- return new ResumableProcessor(resumeStrategy, childProcessor);
+ LoggingLevel loggingLevel = resolveLoggingLevel();
+ return new ResumableProcessor(resumeStrategy, childProcessor, loggingLevel);
}
protected ResumeStrategy resolveResumeStrategy() {
@@ -47,6 +49,17 @@ public class ResumableReifier extends ProcessorReifier<ResumableDefinition> {
String ref = parseString(definition.getResumeStrategy());
strategy = mandatoryLookup(ref, ResumeStrategy.class);
}
+
return strategy;
}
+
+ protected LoggingLevel resolveLoggingLevel() {
+ LoggingLevel loggingLevel = parse(LoggingLevel.class, definition.getLoggingLevel());
+
+ if (loggingLevel == null) {
+ loggingLevel = LoggingLevel.ERROR;
+ }
+
+ return loggingLevel;
+ }
}
diff --git a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
index dee460b426c..24df3c4b80c 100644
--- a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
+++ b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
@@ -922,7 +922,13 @@ public class ModelParser extends BaseParser {
return true;
}
return processorDefinitionAttributeHandler().accept(def, key, val);
- }, optionalIdentifiedDefinitionElementHandler(), noValueHandler());
+ }, (def, key) -> {
+ if ("loggingLevel".equals(key)) {
+ def.setLoggingLevel(doParseText());
+ return true;
+ }
+ return optionalIdentifiedDefinitionElementHandler().accept(def, key);
+ }, noValueHandler());
}
protected RollbackDefinition doParseRollbackDefinition() throws IOException, XmlPullParserException {
return doParse(new RollbackDefinition(), (def, key, val) -> {