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 2025/01/02 20:42:01 UTC
(camel) 01/01: CAMEL-21580: camel-core: Fix rare potential ConcurrentModificationException when using producer template result processor during header copying. Use specialized processor as the message body is the only desired output so we avoid the header copying entirely.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch con-rp
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 3d8a3d82ee02905073f6b53c10baeb9272ca322e
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Thu Jan 2 21:39:24 2025 +0100
CAMEL-21580: camel-core: Fix rare potential ConcurrentModificationException when using producer template result processor during header copying. Use specialized processor as the message body is the only desired output so we avoid the header copying entirely.
---
.../impl/engine/DefaultFluentProducerTemplate.java | 3 +-
.../camel/impl/engine/DefaultProducerTemplate.java | 19 ++++----
.../engine/ProducerTemplateResultProcessor.java | 51 ++++++++++++++++++++++
3 files changed, 63 insertions(+), 10 deletions(-)
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultFluentProducerTemplate.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultFluentProducerTemplate.java
index b1063041a6e..b7e85d5da83 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultFluentProducerTemplate.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultFluentProducerTemplate.java
@@ -34,7 +34,6 @@ import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.support.ExchangeHelper;
-import org.apache.camel.support.processor.ConvertBodyProcessor;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.ObjectHelper;
@@ -77,7 +76,7 @@ public class DefaultFluentProducerTemplate extends ServiceSupport implements Flu
this.resultProcessors = new ClassValue<>() {
@Override
protected Processor computeValue(Class<?> type) {
- return new ConvertBodyProcessor(type);
+ return new ProducerTemplateResultProcessor(type);
}
};
}
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultProducerTemplate.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultProducerTemplate.java
index 522969f5ce0..91f4f57f84b 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultProducerTemplate.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultProducerTemplate.java
@@ -38,7 +38,6 @@ import org.apache.camel.spi.Synchronization;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.cache.DefaultProducerCache;
-import org.apache.camel.support.processor.ConvertBodyProcessor;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.ObjectHelper;
@@ -401,7 +400,8 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
@Override
public <T> T requestBody(Endpoint endpoint, Object body, Class<T> type) {
Exchange exchange
- = send(endpoint, ExchangePattern.InOut, createSetBodyProcessor(body), new ConvertBodyProcessor(type));
+ = send(endpoint, ExchangePattern.InOut, createSetBodyProcessor(body),
+ new ProducerTemplateResultProcessor(type));
Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
@@ -410,7 +410,8 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
public <T> T requestBody(String endpointUri, Object body, Class<T> type) {
Endpoint endpoint = resolveMandatoryEndpoint(endpointUri);
Exchange exchange
- = send(endpoint, ExchangePattern.InOut, createSetBodyProcessor(body), new ConvertBodyProcessor(type));
+ = send(endpoint, ExchangePattern.InOut, createSetBodyProcessor(body),
+ new ProducerTemplateResultProcessor(type));
Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
@@ -418,7 +419,7 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
@Override
public <T> T requestBodyAndHeader(Endpoint endpoint, Object body, String header, Object headerValue, Class<T> type) {
Exchange exchange = send(endpoint, ExchangePattern.InOut, createBodyAndHeaderProcessor(body, header, headerValue),
- new ConvertBodyProcessor(type));
+ new ProducerTemplateResultProcessor(type));
Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
@@ -427,7 +428,7 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
public <T> T requestBodyAndHeader(String endpointUri, Object body, String header, Object headerValue, Class<T> type) {
Endpoint endpoint = resolveMandatoryEndpoint(endpointUri);
Exchange exchange = send(endpoint, ExchangePattern.InOut, createBodyAndHeaderProcessor(body, header, headerValue),
- new ConvertBodyProcessor(type));
+ new ProducerTemplateResultProcessor(type));
Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
@@ -436,7 +437,8 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
public <T> T requestBodyAndHeaders(String endpointUri, Object body, Map<String, Object> headers, Class<T> type) {
Endpoint endpoint = resolveMandatoryEndpoint(endpointUri);
Exchange exchange
- = send(endpoint, ExchangePattern.InOut, createBodyAndHeaders(body, headers), new ConvertBodyProcessor(type));
+ = send(endpoint, ExchangePattern.InOut, createBodyAndHeaders(body, headers),
+ new ProducerTemplateResultProcessor(type));
Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
@@ -444,7 +446,8 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
@Override
public <T> T requestBodyAndHeaders(Endpoint endpoint, Object body, Map<String, Object> headers, Class<T> type) {
Exchange exchange
- = send(endpoint, ExchangePattern.InOut, createBodyAndHeaders(body, headers), new ConvertBodyProcessor(type));
+ = send(endpoint, ExchangePattern.InOut, createBodyAndHeaders(body, headers),
+ new ProducerTemplateResultProcessor(type));
Object answer = extractResultBody(exchange);
return camelContext.getTypeConverter().convertTo(type, answer);
}
@@ -673,7 +676,7 @@ public class DefaultProducerTemplate extends ServiceSupport implements ProducerT
}
protected <T> CompletableFuture<T> asyncRequestBody(final Endpoint endpoint, Processor processor, final Class<T> type) {
- return asyncRequestBody(endpoint, processor, new ConvertBodyProcessor(type))
+ return asyncRequestBody(endpoint, processor, new ProducerTemplateResultProcessor(type))
.thenApply(answer -> camelContext.getTypeConverter().convertTo(type, answer));
}
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/ProducerTemplateResultProcessor.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/ProducerTemplateResultProcessor.java
new file mode 100644
index 00000000000..512f89bed6b
--- /dev/null
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/ProducerTemplateResultProcessor.java
@@ -0,0 +1,51 @@
+/*
+ * 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.impl.engine;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+
+/**
+ * To be used by producer template to prepare the message body for a given expected result type.
+ */
+class ProducerTemplateResultProcessor implements Processor {
+
+ private final Class<?> type;
+
+ public ProducerTemplateResultProcessor(Class<?> type) {
+ this.type = type;
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ if (exchange.getMessage().getBody() == null) {
+ // only convert if there is a body
+ return;
+ }
+
+ if (exchange.getException() != null) {
+ // do not convert if an exception has been thrown as if we attempt to convert and it also fails with a new
+ // exception then it will override the existing exception
+ return;
+ }
+
+ Object newBody = exchange.getMessage().getBody(type);
+ if (newBody != null) {
+ exchange.getMessage().setBody(newBody);
+ }
+ }
+}