This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26515-grpc-route-controlled-reply-type in repository https://gitbox.apache.org/repos/asf/camel.git
commit ba3328b8297f8185d233381df8719723ef60e385 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Nov 3 15:48:45 2026 +0100 CAMEL-26515: camel-grpc - the toRouteControlledStreamObserver producer converts the reply to the response type of the method, or fails the call With routeControlledStreamObserver=true the route sends the replies through a producer with toRouteControlledStreamObserver=true, which called onNext with the body as is. The unchanged request was decoded by the client as the response. The consumer now sets the response message type of the method as the grpcResponseType exchange property (unary, server streaming and DELEGATION). The producer converts each reply (each element of a List) to it, like the consumer does since CAMEL-26187, and when a message does not convert it ends the call with an INTERNAL error and fails the exchange. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/catalog/docs/grpc-component.adoc | 11 +- .../camel-grpc/src/main/docs/grpc-component.adoc | 11 +- .../apache/camel/component/grpc/GrpcConstants.java | 5 + ...rpcProducerToRouteControlledStreamObserver.java | 23 ++- .../component/grpc/server/GrpcMethodHandler.java | 40 +++- .../GrpcRequestDelegationStreamObserver.java | 3 + .../grpc/GrpcRouteControlledReplyTypeTest.java | 224 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 5 + 8 files changed, 308 insertions(+), 14 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/grpc-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/grpc-component.adoc index 9cc188ba41a3..a57b34a1c5a8 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/grpc-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/grpc-component.adoc @@ -80,8 +80,15 @@ method, or a `List` of them for a streaming response. A message without a body s A body of another type is converted to the response message type with a Camel type converter. When there is no such conversion, nothing is sent, and the call ends with an `INTERNAL` status error. For example, a route that leaves the -request message unchanged fails the call instead of sending the request as the response. This does not apply with -`routeControlledStreamObserver=true`, where the route sends the responses itself. +request message unchanged fails the call instead of sending the request as the response. + +With `routeControlledStreamObserver=true` the route sends the responses itself, with a producer endpoint with +`toRouteControlledStreamObserver=true`. That producer applies the same conversion to the body of each `onNext` event +(each element of a `List` body is sent as a message). When a message cannot be converted, nothing is sent, the call +ends with an `INTERNAL` status error and the exchange fails. The error description is muted unless the producer +endpoint has `muteException=false`. The consumer passes the response message type in the `grpcResponseType` exchange +property; when the property is not on the exchange, the body is sent as is. Calls on the `StreamObserver` from the +`grpcResponseObserver` exchange property in the route's own code are not checked. === gRPC producer resource type mapping diff --git a/components/camel-grpc/src/main/docs/grpc-component.adoc b/components/camel-grpc/src/main/docs/grpc-component.adoc index 9cc188ba41a3..a57b34a1c5a8 100644 --- a/components/camel-grpc/src/main/docs/grpc-component.adoc +++ b/components/camel-grpc/src/main/docs/grpc-component.adoc @@ -80,8 +80,15 @@ method, or a `List` of them for a streaming response. A message without a body s A body of another type is converted to the response message type with a Camel type converter. When there is no such conversion, nothing is sent, and the call ends with an `INTERNAL` status error. For example, a route that leaves the -request message unchanged fails the call instead of sending the request as the response. This does not apply with -`routeControlledStreamObserver=true`, where the route sends the responses itself. +request message unchanged fails the call instead of sending the request as the response. + +With `routeControlledStreamObserver=true` the route sends the responses itself, with a producer endpoint with +`toRouteControlledStreamObserver=true`. That producer applies the same conversion to the body of each `onNext` event +(each element of a `List` body is sent as a message). When a message cannot be converted, nothing is sent, the call +ends with an `INTERNAL` status error and the exchange fails. The error description is muted unless the producer +endpoint has `muteException=false`. The consumer passes the response message type in the `grpcResponseType` exchange +property; when the property is not on the exchange, the body is sent as is. Calls on the `StreamObserver` from the +`grpcResponseObserver` exchange property in the route's own code are not checked. === gRPC producer resource type mapping diff --git a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConstants.java b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConstants.java index bbb140e20d81..f13f46b12174 100644 --- a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConstants.java +++ b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcConstants.java @@ -70,4 +70,9 @@ public interface GrpcConstants { */ String GRPC_BINDABLE_SERVICE_FACTORY_NAME = "grpcBindableServiceFactory"; String GRPC_RESPONSE_OBSERVER = "grpcResponseObserver"; + /* + * The exchange property with the response message type of the called method, set by a consumer with + * routeControlledStreamObserver=true: the toRouteControlledStreamObserver producer converts each reply to it + */ + String GRPC_RESPONSE_TYPE = "grpcResponseType"; } diff --git a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java index efaad79a2ab6..97c45550e9af 100644 --- a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java +++ b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/GrpcProducerToRouteControlledStreamObserver.java @@ -16,9 +16,12 @@ */ package org.apache.camel.component.grpc; +import java.util.List; + import io.grpc.stub.StreamObserver; import org.apache.camel.Endpoint; import org.apache.camel.Exchange; +import org.apache.camel.component.grpc.server.GrpcMethodHandler; import org.apache.camel.support.DefaultProducer; public class GrpcProducerToRouteControlledStreamObserver extends DefaultProducer { @@ -33,7 +36,7 @@ public class GrpcProducerToRouteControlledStreamObserver extends DefaultProducer String eventType = exchange.getMessage().getHeader( GrpcConstants.GRPC_EVENT_TYPE_HEADER, String.class); switch (eventType) { - case GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT -> observer.onNext(exchange.getMessage().getBody()); + case GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT -> onNext(exchange, observer); case GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED -> observer.onCompleted(); case GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR -> observer.onError((Throwable) exchange.getMessage().getBody()); default -> { @@ -41,4 +44,22 @@ public class GrpcProducerToRouteControlledStreamObserver extends DefaultProducer } } } + + /** + * Sends the body to the client: each element of a List body, or the body itself, converted to the response type of + * the method that the consumer set as exchange property. When a message cannot be converted, nothing is sent, the + * call ends with an error and the exchange fails, as the client would otherwise decode the bytes of another message + * type (for example the unchanged request) as the response. + */ + private void onNext(Exchange exchange, StreamObserver<Object> observer) { + Class<?> responseType = exchange.getProperty(GrpcConstants.GRPC_RESPONSE_TYPE, Class.class); + List<Object> responses; + try { + responses = GrpcMethodHandler.toResponses(exchange, exchange.getMessage().getBody(), responseType); + } catch (IllegalArgumentException e) { + observer.onError(GrpcMethodHandler.toStatusException((GrpcEndpoint) getEndpoint(), e)); + throw e; + } + responses.forEach(observer::onNext); + } } diff --git a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java index 91404959c5f2..4ae897735137 100644 --- a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java +++ b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcMethodHandler.java @@ -76,6 +76,7 @@ public class GrpcMethodHandler { if (endpoint.getConfiguration().isRouteControlledStreamObserver()) { exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER, responseObserver); + exchange.setProperty(GrpcConstants.GRPC_RESPONSE_TYPE, getResponseType(methodName)); invokeRoute(endpoint, exchange); return; } @@ -106,6 +107,30 @@ public class GrpcMethodHandler { static boolean sendResponses( GrpcEndpoint endpoint, Exchange exchange, Object body, Class<?> responseType, StreamObserver<Object> responseObserver) { + List<Object> responses; + try { + responses = toResponses(exchange, body, responseType); + } catch (IllegalArgumentException e) { + LOG.warn("Cannot send the reply of exchange {}: {}", exchange.getExchangeId(), e.getMessage()); + responseObserver.onError(toStatusException(endpoint, e)); + return false; + } + responses.forEach(responseObserver::onNext); + return true; + } + + /** + * The response messages to send to the client: each element of a List body, or the body itself, converted to the + * response type of the method. + * + * @param exchange the exchange of the reply + * @param body the reply of the route + * @param responseType the response message type of the method, or null when unknown to keep the body + * as is + * @return the response messages + * @throws IllegalArgumentException when a message is not of the response type and cannot be converted to it + */ + public static List<Object> toResponses(Exchange exchange, Object body, Class<?> responseType) { List<?> messages = body instanceof List<?> list ? list : Collections.singletonList(body); List<Object> responses = new ArrayList<>(messages.size()); for (Object message : messages) { @@ -117,16 +142,13 @@ public class GrpcMethodHandler { try { responses.add(exchange.getContext().getTypeConverter().mandatoryConvertTo(responseType, exchange, message)); } catch (NoTypeConversionAvailableException | TypeConversionException e) { - String reason = "The reply of the gRPC method must be a " + responseType.getName() - + " or a List of them, but was a " + message.getClass().getName(); - IllegalArgumentException cause = new IllegalArgumentException(reason, e); - LOG.warn("Cannot send the reply of exchange {}: {}", exchange.getExchangeId(), cause.getMessage()); - responseObserver.onError(toStatusException(endpoint, cause)); - return false; + throw new IllegalArgumentException( + "The reply of the gRPC method must be a " + responseType.getName() + + " or a List of them, but was a " + message.getClass().getName(), + e); } } - responses.forEach(responseObserver::onNext); - return true; + return responses; } /** @@ -165,7 +187,7 @@ public class GrpcMethodHandler { /** * The error to send to the client when the exchange failed. */ - static StatusRuntimeException toStatusException(GrpcEndpoint endpoint, Exception cause) { + public static StatusRuntimeException toStatusException(GrpcEndpoint endpoint, Exception cause) { // the description IS transmitted to the client, so it carries the route's exception message only when // the endpoint opted out of muting; the cause below stays local either way String description = endpoint.getConfiguration().isMuteException() ? MUTED_DESCRIPTION : cause.getMessage(); diff --git a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestDelegationStreamObserver.java b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestDelegationStreamObserver.java index bb7d2bd2018b..4400287455a5 100644 --- a/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestDelegationStreamObserver.java +++ b/components/camel-grpc/src/main/java/org/apache/camel/component/grpc/server/GrpcRequestDelegationStreamObserver.java @@ -40,6 +40,7 @@ public class GrpcRequestDelegationStreamObserver extends GrpcRequestAbstractStre exchange.getIn().setBody(request); exchange.getIn().setHeaders(headers); exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER, responseObserver); + exchange.setProperty(GrpcConstants.GRPC_RESPONSE_TYPE, responseType); consumer.process(exchange, doneSync -> { }); } @@ -49,6 +50,7 @@ public class GrpcRequestDelegationStreamObserver extends GrpcRequestAbstractStre var exchange = endpoint.createExchange(); exchange.getIn().setHeaders(headers); exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER, responseObserver); + exchange.setProperty(GrpcConstants.GRPC_RESPONSE_TYPE, responseType); consumer.onError(exchange, throwable); } @@ -57,6 +59,7 @@ public class GrpcRequestDelegationStreamObserver extends GrpcRequestAbstractStre var exchange = endpoint.createExchange(); exchange.getIn().setHeaders(headers); exchange.setProperty(GrpcConstants.GRPC_RESPONSE_OBSERVER, responseObserver); + exchange.setProperty(GrpcConstants.GRPC_RESPONSE_TYPE, responseType); consumer.onCompleted(exchange); } } diff --git a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java new file mode 100644 index 000000000000..0256c29c4494 --- /dev/null +++ b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java @@ -0,0 +1,224 @@ +/* + * 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.grpc; + +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.stub.StreamObserver; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.grpc.GrpcConsumerReplyTypeTest.Pong; +import org.apache.camel.component.grpc.GrpcConsumerReplyTypeTest.PongResponseStreamObserver; +import org.apache.camel.support.TypeConverterSupport; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * With routeControlledStreamObserver=true, a reply sent by the toRouteControlledStreamObserver producer must be of the + * response type of the called method: a body of another type is converted to it, or the call ends with an error. It + * must never be serialized as is, as the client would decode the bytes of, for example, the unchanged request as a + * response. + */ +public class GrpcRouteControlledReplyTypeTest extends GrpcTestSupport { + + private static final String PING_NAME = "PING"; + private static final int PING_ID = 7; + + private static final String CONSUMER = "grpc://localhost:0/org.apache.camel.component.grpc.PingPong?synchronous=true" + + "&routeControlledStreamObserver=true"; + private static final String REPLY = "grpc://dummy:0/?toRouteControlledStreamObserver=true"; + + private final Map<String, ManagedChannel> channels = new ConcurrentHashMap<>(); + + @AfterEach + public void stopGrpcChannels() { + channels.values().forEach(channel -> channel.shutdown().shutdownNow()); + channels.clear(); + } + + @Test + public void unchangedRequestIsNotSentAsResponse() { + StatusRuntimeException e = assertThrows(StatusRuntimeException.class, + () -> blockingStub("rc-unchanged").pingSyncSync(pingRequest())); + + assertReplyTypeError(e.getStatus(), PingRequest.class); + } + + @Test + public void replyTypeErrorIsMutedByDefault() { + StatusRuntimeException e = assertThrows(StatusRuntimeException.class, + () -> blockingStub("rc-unchanged-muted").pingSyncSync(pingRequest())); + + assertEquals(Status.Code.INTERNAL, e.getStatus().getCode()); + assertEquals("Exchange processing failed", e.getStatus().getDescription()); + } + + @Test + public void replyIsConvertedToTheResponseType() { + PongResponse pongResponse = blockingStub("rc-convert").pingSyncSync(pingRequest()); + + assertEquals(PING_NAME + "PONG", pongResponse.getPongName()); + assertEquals(PING_ID, pongResponse.getPongId()); + } + + @Test + public void serverStreamingSendsEachMessageOfAList() { + Iterator<PongResponse> responses = blockingStub("rc-streaming-list").pingSyncAsync(pingRequest()); + + assertEquals(PING_NAME + "PONG", responses.next().getPongName()); + assertEquals(PING_NAME + "PONG2", responses.next().getPongName()); + assertFalse(responses.hasNext()); + } + + @Test + public void delegationUnchangedRequestIsNotSentAsResponse() throws Exception { + PongResponseStreamObserver responseObserver = callBidiStreaming("rc-delegation-unchanged"); + + assertTrue(responseObserver.pongResponses.isEmpty(), "the request must not be sent as response"); + StatusRuntimeException e = assertInstanceOf(StatusRuntimeException.class, responseObserver.error); + assertReplyTypeError(e.getStatus(), PingRequest.class); + assertEquals(0, responseObserver.completed.get()); + } + + @Test + public void delegationConvertsTheReply() throws Exception { + PongResponseStreamObserver responseObserver = callBidiStreaming("rc-delegation-convert"); + + assertEquals(1, responseObserver.pongResponses.size()); + assertEquals(PING_NAME + "PONG", responseObserver.pongResponses.get(0).getPongName()); + assertEquals(1, responseObserver.completed.get()); + } + + private static void assertReplyTypeError(Status status, Class<?> replyType) { + assertEquals(Status.Code.INTERNAL, status.getCode()); + assertTrue(status.getDescription().contains(PongResponse.class.getName()), + "the description must name the response type: " + status); + assertTrue(status.getDescription().contains(replyType.getName()), + "the description must name the type of the reply: " + status); + } + + private static PingRequest pingRequest() { + return PingRequest.newBuilder().setPingName(PING_NAME).setPingId(PING_ID).build(); + } + + private ManagedChannel channel(String routeId) { + return channels.computeIfAbsent(routeId, + id -> ManagedChannelBuilder.forAddress("localhost", getRoutePort(id)).usePlaintext().build()); + } + + private PingPongGrpc.PingPongBlockingStub blockingStub(String routeId) { + return PingPongGrpc.newBlockingStub(channel(routeId)).withDeadlineAfter(5, TimeUnit.SECONDS); + } + + private PongResponseStreamObserver callBidiStreaming(String routeId) throws InterruptedException { + PongResponseStreamObserver responseObserver = new PongResponseStreamObserver(); + StreamObserver<PingRequest> requestObserver = PingPongGrpc.newStub(channel(routeId)).pingAsyncAsync(responseObserver); + requestObserver.onNext(pingRequest()); + requestObserver.onCompleted(); + assertTrue(responseObserver.latch.await(5, TimeUnit.SECONDS)); + return responseObserver; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + getContext().getTypeConverterRegistry().addTypeConverter(PongResponse.class, Pong.class, + new TypeConverterSupport() { + @Override + public <T> T convertTo(Class<T> type, Exchange exchange, Object value) { + Pong pong = (Pong) value; + return type.cast(PongResponse.newBuilder().setPongName(pong.name()).setPongId(pong.id()) + .build()); + } + }); + + // unary and server streaming calls: the route sends the reply, then completes the call + from(CONSUMER + "&hash=1") + .routeId("rc-unchanged") + .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT)) + .to(REPLY + "&muteException=false") + .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, + constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED)) + .to(REPLY); + + from(CONSUMER + "&hash=2") + .routeId("rc-unchanged-muted") + .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT)) + .to(REPLY) + .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, + constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED)) + .to(REPLY); + + from(CONSUMER + "&hash=3") + .routeId("rc-convert") + .process(e -> e.getMessage().setBody(toPong(e.getMessage().getBody(PingRequest.class), "PONG"))) + .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT)) + .to(REPLY) + .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, + constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED)) + .to(REPLY); + + from(CONSUMER + "&hash=4") + .routeId("rc-streaming-list") + .process(e -> { + PingRequest request = e.getMessage().getBody(PingRequest.class); + e.getMessage().setBody(List.of(toPong(request, "PONG"), toPong(request, "PONG2"))); + }) + .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, constant(GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT)) + .to(REPLY) + .setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, + constant(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED)) + .to(REPLY); + + // client and bidirectional streaming calls: the consumer forwards each event with its type header + from(CONSUMER + "&hash=5&consumerStrategy=DELEGATION&forwardOnCompleted=true") + .routeId("rc-delegation-unchanged") + .to(REPLY + "&muteException=false"); + + from(CONSUMER + "&hash=6&consumerStrategy=DELEGATION&forwardOnCompleted=true") + .routeId("rc-delegation-convert") + .process(e -> { + PingRequest request = e.getMessage().getBody(PingRequest.class); + if (request != null) { + e.getMessage().setBody(toPong(request, "PONG")); + } + }) + .to(REPLY); + } + }; + } + + private static Pong toPong(PingRequest request, String suffix) { + return new Pong(request.getPingName() + suffix, request.getPingId()); + } +} 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 7c82ebd18930..c811e2758aa4 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 @@ -928,6 +928,11 @@ of the called method, and when that is not possible it sends nothing and ends th A route must set the response message of the method as the body. A reply without a body still sends an empty response message. +The same applies to the replies that a producer with `toRouteControlledStreamObserver=true` sends for a consumer with +`routeControlledStreamObserver=true`: a reply that does not convert ends the call with an `INTERNAL` status error and +fails the exchange. Such a producer also sends each element of a `List` body as a message, where a `List` failed +before. + === camel-hivemq - a body that does not convert to a byte array fails the exchange The `hivemq` producer published a message body that has no conversion to `byte[]`, such as a `Map`, a POJO or a
