This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 8002509c386f CAMEL-26190: camel-jgroups - a stream or a file body is
sent as its content, and another body that is not Serializable fails with a
hint (#29018)
8002509c386f is described below
commit 8002509c386f481a627cfb6e1cd4cf1a951c776b
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 07:42:57 2026 +0100
CAMEL-26190: camel-jgroups - a stream or a file body is sent as its
content, and another body that is not Serializable fails with a hint (#29018)
…ontent, and another body that is not Serializable fails with a hint
The producer wrapped the body as is in an ObjectMessage, so a stream, a
stream
cache, a GenericFile or a Path failed with a NotSerializableException of
that
class, with no hint what to do. These bodies are now converted to a byte[]
of
their content. Any other body that is neither Serializable nor a JGroups
SizeStreamable fails with an InvalidPayloadException that says to convert it
first. Serializable bodies (including java.io.File) are sent as before.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/components/jgroups.json | 2 +-
.../camel/catalog/docs/jgroups-component.adoc | 7 +
.../apache/camel/component/jgroups/jgroups.json | 2 +-
.../src/main/docs/jgroups-component.adoc | 7 +
.../camel/component/jgroups/JGroupsEndpoint.java | 11 +-
.../camel/component/jgroups/JGroupsProducer.java | 25 ++++
.../component/jgroups/JGroupsProducerBodyTest.java | 165 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 ++
8 files changed, 222 insertions(+), 6 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/jgroups.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/jgroups.json
index 79a55abbdfa9..655872db4bcc 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/jgroups.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/jgroups.json
@@ -41,7 +41,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "unknown" ], "dependsOn": [
"enableViewMessages" ], "description": "The object the sender put in the
message, Java-deserialized: what a jgroups producer sent, such as a String, a
byte array, a Map or a Serializable POJO, if the deserialization filter accepts
its class. With enableViewMessages an org.jgroups.View also arrives on each
change of the members of the cluster" },
- "producerIn": { "carriers": [ "text", "bytes", "value", "map", "list",
"file", "pojo" ], "description": "The body as is, Java-serialized into the
message, so it must be Serializable: a stream or another object that is not
fails with a NotSerializableException. A null body is not sent. The
CamelJGroupsDest header sends it to one member instead of the cluster" },
+ "producerIn": { "carriers": [ "text", "bytes", "value", "map", "list",
"stream", "file", "pojo" ], "description": "The body as is, Java-serialized
into the message, so it must be Serializable. A stream (an InputStream, a
Reader or a stream cache), a file from the file component or a Path is sent as
a byte array of its content, and a java.io.File as is; another object that is
not Serializable fails with an InvalidPayloadException. A null body is not
sent. The CamelJGroupsDest header s [...]
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jgroups-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jgroups-component.adoc
index ed05c593ccfa..e75c9ad64acd 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jgroups-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jgroups-component.adoc
@@ -183,6 +183,13 @@ Using `jgroups` component on the producer side of the
route will forward
body of the Camel exchanges to the `JChannel` instance managed by the
endpoint.
+The body is Java-serialized into the JGroups message, so it must be
`Serializable`.
+A stream body (such as an `InputStream`, a `Reader` or a stream cache), a file
+from the `file` component and a `java.nio.file.Path` are read and sent as a
`byte[]`
+of their content. A `java.io.File` is `Serializable` and is sent as is. Any
other
+body that is not `Serializable` fails with an `InvalidPayloadException` that
says to
+convert it first, for example to a `String` or a `byte[]`.
+
[tabs]
====
Java::
diff --git
a/components/camel-jgroups/src/generated/resources/META-INF/org/apache/camel/component/jgroups/jgroups.json
b/components/camel-jgroups/src/generated/resources/META-INF/org/apache/camel/component/jgroups/jgroups.json
index 79a55abbdfa9..655872db4bcc 100644
---
a/components/camel-jgroups/src/generated/resources/META-INF/org/apache/camel/component/jgroups/jgroups.json
+++
b/components/camel-jgroups/src/generated/resources/META-INF/org/apache/camel/component/jgroups/jgroups.json
@@ -41,7 +41,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "unknown" ], "dependsOn": [
"enableViewMessages" ], "description": "The object the sender put in the
message, Java-deserialized: what a jgroups producer sent, such as a String, a
byte array, a Map or a Serializable POJO, if the deserialization filter accepts
its class. With enableViewMessages an org.jgroups.View also arrives on each
change of the members of the cluster" },
- "producerIn": { "carriers": [ "text", "bytes", "value", "map", "list",
"file", "pojo" ], "description": "The body as is, Java-serialized into the
message, so it must be Serializable: a stream or another object that is not
fails with a NotSerializableException. A null body is not sent. The
CamelJGroupsDest header sends it to one member instead of the cluster" },
+ "producerIn": { "carriers": [ "text", "bytes", "value", "map", "list",
"stream", "file", "pojo" ], "description": "The body as is, Java-serialized
into the message, so it must be Serializable. A stream (an InputStream, a
Reader or a stream cache), a file from the file component or a Path is sent as
a byte array of its content, and a java.io.File as is; another object that is
not Serializable fails with an InvalidPayloadException. A null body is not
sent. The CamelJGroupsDest header s [...]
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git a/components/camel-jgroups/src/main/docs/jgroups-component.adoc
b/components/camel-jgroups/src/main/docs/jgroups-component.adoc
index ed05c593ccfa..e75c9ad64acd 100644
--- a/components/camel-jgroups/src/main/docs/jgroups-component.adoc
+++ b/components/camel-jgroups/src/main/docs/jgroups-component.adoc
@@ -183,6 +183,13 @@ Using `jgroups` component on the producer side of the
route will forward
body of the Camel exchanges to the `JChannel` instance managed by the
endpoint.
+The body is Java-serialized into the JGroups message, so it must be
`Serializable`.
+A stream body (such as an `InputStream`, a `Reader` or a stream cache), a file
+from the `file` component and a `java.nio.file.Path` are read and sent as a
`byte[]`
+of their content. A `java.io.File` is `Serializable` and is sent as is. Any
other
+body that is not `Serializable` fails with an `InvalidPayloadException` that
says to
+convert it first, for example to a `String` or a `byte[]`.
+
[tabs]
====
Java::
diff --git
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
index 5be04b8b9910..6dbd0d58c75a 100644
---
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
+++
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsEndpoint.java
@@ -58,11 +58,14 @@ import org.slf4j.LoggerFactory;
+ " members of the cluster"),
producerIn = @BodyType(value = {
Carrier.TEXT, Carrier.BYTES, Carrier.VALUE, Carrier.MAP,
Carrier.LIST,
- Carrier.FILE, Carrier.POJO },
+ Carrier.STREAM, Carrier.FILE, Carrier.POJO },
description = "The body as is,
Java-serialized into the message, so it must be"
- + " Serializable: a stream or
another object that is not fails with a"
- + " NotSerializableException.
A null body is not sent. The CamelJGroupsDest header"
- + " sends it to one member
instead of the cluster"),
+ + " Serializable. A stream (an
InputStream, a Reader or a stream cache), a"
+ + " file from the file
component or a Path is sent as a byte array of its"
+ + " content, and a
java.io.File as is;"
+ + " another object that is not
Serializable fails with an"
+ + " InvalidPayloadException. A
null body is not sent. The CamelJGroupsDest"
+ + " header sends it to one
member instead of the cluster"),
producerOut = @BodyType(Carrier.UNCHANGED))
public class JGroupsEndpoint extends DefaultEndpoint {
diff --git
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsProducer.java
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsProducer.java
index e635c33d4efb..7fa042e7d050 100644
---
a/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsProducer.java
+++
b/components/camel-jgroups/src/main/java/org/apache/camel/component/jgroups/JGroupsProducer.java
@@ -16,11 +16,20 @@
*/
package org.apache.camel.component.jgroups;
+import java.io.InputStream;
+import java.io.Reader;
+import java.io.Serializable;
+import java.nio.file.Path;
+
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.StreamCache;
+import org.apache.camel.WrappedFile;
import org.apache.camel.support.DefaultProducer;
import org.jgroups.Address;
import org.jgroups.Message;
import org.jgroups.ObjectMessage;
+import org.jgroups.util.SizeStreamable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -73,6 +82,7 @@ public class JGroupsProducer extends DefaultProducer {
public void process(Exchange exchange) throws Exception {
Object body = exchange.getIn().getBody();
if (body != null) {
+ body = toSerializableBody(exchange, body);
Address destinationAddress =
exchange.getIn().getHeader(JGroupsConstants.HEADER_JGROUPS_DEST, Address.class);
Address sourceAddress =
exchange.getIn().getHeader(JGroupsConstants.HEADER_JGROUPS_SRC, Address.class);
@@ -91,4 +101,19 @@ public class JGroupsProducer extends DefaultProducer {
}
}
+ private static Object toSerializableBody(Exchange exchange, Object body)
throws InvalidPayloadException {
+ if (body instanceof Serializable || body instanceof SizeStreamable) {
+ return body;
+ }
+ if (body instanceof StreamCache || body instanceof InputStream || body
instanceof Reader
+ || body instanceof WrappedFile || body instanceof Path) {
+ // a stream or a file has no serialized form, so send its content
+ return exchange.getIn().getMandatoryBody(byte[].class);
+ }
+ throw new InvalidPayloadException(
+ exchange, Serializable.class, exchange.getIn(),
+ "the body is Java-serialized into the JGroups message, so
convert it first, for example to a String"
+ + " or a
byte[]");
+ }
+
}
diff --git
a/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsProducerBodyTest.java
b/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsProducerBodyTest.java
new file mode 100644
index 000000000000..cf82ce4b22f8
--- /dev/null
+++
b/components/camel-jgroups/src/test/java/org/apache/camel/component/jgroups/JGroupsProducerBodyTest.java
@@ -0,0 +1,165 @@
+/*
+ * 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.jgroups;
+
+import java.io.ByteArrayInputStream;
+import java.io.StringReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.StreamCache;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A body that has no serialized form (a stream, a stream cache, a file from
the file component or a Path) is sent as
+ * its content, and another body that is not Serializable fails with a hint to
convert it.
+ */
+public class JGroupsProducerBodyTest extends CamelTestSupport {
+
+ static final String CLUSTER_NAME = "JGroupsProducerBodyTest";
+
+ static final String CONTENT = "Hello JGroups";
+
+ @TempDir
+ Path inbox;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("jgroups:" + CLUSTER_NAME +
"?acceptAllObjects=true").to("mock:received");
+
+ from("direct:cached").streamCaching().to("jgroups:" +
CLUSTER_NAME);
+ from("direct:stream").noStreamCaching().to("jgroups:" +
CLUSTER_NAME);
+ from("file:" + inbox +
"?initialDelay=0&delay=10").noAutoStartup().routeId("file")
+ .to("jgroups:" + CLUSTER_NAME);
+ }
+ };
+ }
+
+ @Test
+ public void shouldSendStreamCacheContent() throws Exception {
+ MockEndpoint received = getMockEndpoint("mock:received");
+ received.expectedMessageCount(1);
+
+ template.send("direct:cached", exchange -> {
+ exchange.getIn().setBody(new
ByteArrayInputStream(CONTENT.getBytes(StandardCharsets.UTF_8)));
+ });
+
+ received.assertIsSatisfied();
+ assertArrayEquals(CONTENT.getBytes(StandardCharsets.UTF_8),
+
received.getReceivedExchanges().get(0).getIn().getBody(byte[].class));
+ }
+
+ @Test
+ public void shouldKeepStreamCacheReadableAfterSend() throws Exception {
+ getMockEndpoint("mock:received").expectedMessageCount(1);
+
+ Object body = template.send("direct:cached", exchange -> {
+ exchange.getIn().setBody(new
ByteArrayInputStream(CONTENT.getBytes(StandardCharsets.UTF_8)));
+ }).getMessage().getBody();
+
+ MockEndpoint.assertIsSatisfied(context);
+ StreamCache cache = assertInstanceOf(StreamCache.class, body);
+ assertArrayEquals(CONTENT.getBytes(StandardCharsets.UTF_8),
+ context.getTypeConverter().convertTo(byte[].class, cache));
+ }
+
+ @Test
+ public void shouldSendInputStreamContent() throws Exception {
+ MockEndpoint received = getMockEndpoint("mock:received");
+ received.expectedMessageCount(1);
+
+ template.sendBody("direct:stream", new
ByteArrayInputStream(CONTENT.getBytes(StandardCharsets.UTF_8)));
+
+ received.assertIsSatisfied();
+ assertArrayEquals(CONTENT.getBytes(StandardCharsets.UTF_8),
+
received.getReceivedExchanges().get(0).getIn().getBody(byte[].class));
+ }
+
+ @Test
+ public void shouldSendReaderContent() throws Exception {
+ MockEndpoint received = getMockEndpoint("mock:received");
+ received.expectedMessageCount(1);
+
+ template.sendBody("direct:stream", new StringReader(CONTENT));
+
+ received.assertIsSatisfied();
+ assertArrayEquals(CONTENT.getBytes(StandardCharsets.UTF_8),
+
received.getReceivedExchanges().get(0).getIn().getBody(byte[].class));
+ }
+
+ @Test
+ public void shouldSendFileContent() throws Exception {
+ MockEndpoint received = getMockEndpoint("mock:received");
+ received.expectedMessageCount(1);
+
+ Files.writeString(inbox.resolve("hello.txt"), CONTENT);
+ context.getRouteController().startRoute("file");
+
+ received.assertIsSatisfied();
+ assertArrayEquals(CONTENT.getBytes(StandardCharsets.UTF_8),
+
received.getReceivedExchanges().get(0).getIn().getBody(byte[].class));
+ }
+
+ @Test
+ public void shouldSendPathContent() throws Exception {
+ MockEndpoint received = getMockEndpoint("mock:received");
+ received.expectedMessageCount(1);
+
+ Path file = Files.writeString(inbox.resolve("hello.txt"), CONTENT);
+ template.sendBody("direct:stream", file);
+
+ received.assertIsSatisfied();
+ assertArrayEquals(CONTENT.getBytes(StandardCharsets.UTF_8),
+
received.getReceivedExchanges().get(0).getIn().getBody(byte[].class));
+ }
+
+ @Test
+ public void shouldFailWithHintForBodyThatIsNotSerializable() throws
Exception {
+ getMockEndpoint("mock:received").expectedMessageCount(0);
+
+ CamelExecutionException e = assertThrows(CamelExecutionException.class,
+ () -> template.sendBody("direct:stream", new
NotSerializableBody()));
+
+ InvalidPayloadException cause =
assertInstanceOf(InvalidPayloadException.class, e.getCause());
+
assertTrue(cause.getMessage().contains(NotSerializableBody.class.getCanonicalName()),
cause.getMessage());
+ assertTrue(cause.getMessage().contains("convert it first"),
cause.getMessage());
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ static final class NotSerializableBody {
+ @Override
+ public String toString() {
+ return "NotSerializableBody";
+ }
+ }
+
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 3994cf3b7d42..158a61a0e8ab 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -2924,6 +2924,15 @@ header of the incoming message was gone after the
producer. The headers are now
`CamelJcrNodeName` and `CamelJcrNodeType`. Remove these headers before a later
`jcr` producer in the same route
if it must not use them.
+=== camel-jgroups - a stream or a file body is sent as its content
+
+The `jgroups` producer sent the body as is, so a body that is not
`Serializable` failed with a
+`RuntimeException` caused by a `java.io.NotSerializableException`. A stream
body (an `InputStream`, a `Reader` or a
+stream cache), a file from the `file` component and a `java.nio.file.Path` are
now sent as a `byte[]` of their
+content, so the members of the cluster receive a `byte[]`. Any other body that
is not `Serializable` now fails with an
+`InvalidPayloadException` that says to convert it first, before it reaches
JGroups. An `onException` for
+`NotSerializableException` must handle `InvalidPayloadException` instead.
+
=== camel-jira - the producers wait for the answer of Jira
The `addComment`, `addIssueLink`, `addWorkLog`, `attach`, `deleteIssue`,
`transitionIssue` and `watchers` producers
