This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26190-jgroups-stream-body in repository https://gitbox.apache.org/repos/asf/camel.git
commit 71286774e24571e02c313d098004b7ec683562fb Author: Claus Ibsen <[email protected]> AuthorDate: Tue Nov 3 00:04:23 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 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 10f5a47c1b89..dc481012756e 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 @@ -2868,6 +2868,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
