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 de36cecfda63 CAMEL-26217: camel-tensorflow-serving - a body that is
not the request of the api fails instead of being dropped, and model-metadata
asks for signature_def once (#29064)
de36cecfda63 is described below
commit de36cecfda632092d5ee9fe827378ec73bf01648
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Nov 3 12:12:53 2026 +0100
CAMEL-26217: camel-tensorflow-serving - a body that is not the request of
the api fails instead of being dropped, and model-metadata asks for
signature_def once (#29064)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../catalog/components/tensorflow-serving.json | 2 +-
.../catalog/docs/tensorflow-serving-component.adoc | 9 +
.../tensorflow/serving/tensorflow-serving.json | 2 +-
.../main/docs/tensorflow-serving-component.adoc | 9 +
.../serving/TensorFlowServingEndpoint.java | 25 +-
.../serving/TensorFlowServingProducer.java | 71 ++++-
.../serving/TensorFlowServingProducerBodyTest.java | 326 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 16 +
8 files changed, 433 insertions(+), 27 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/tensorflow-serving.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/tensorflow-serving.json
index d7692d3f6acd..43fbf4183aac 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/tensorflow-serving.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/tensorflow-serving.json
@@ -43,7 +43,7 @@
"CamelTensorFlowServingSignatureName": { "index": 5, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "A named signature to evaluate. If
unspecified, the default signature will be used.", "constantName":
"org.apache.camel.component.tensorflow.serving.TensorFlowServingConstants#SIGNATURE_NAME"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "pojo" ], "byOption": [ { "option": "api",
"value": "model-status", "carriers": [ "any" ] }, { "option": "api", "value":
"model-metadata", "carriers": [ "any" ] } ], "description": "The protobuf
request of the api, or its Builder: a PredictRequest for predict, a
ClassificationRequest or RegressionRequest, or the Input of one, for classify
and regress. Another body, such as text, JSON or bytes, is not read: an empty
request is sent and the server rejects [...]
+ "producerIn": { "carriers": [ "pojo" ], "byOption": [ { "option": "api",
"value": "model-status", "carriers": [ "null", "pojo" ] }, { "option": "api",
"value": "model-metadata", "carriers": [ "null", "pojo" ] } ], "description":
"The protobuf request of the api, or its Builder: a PredictRequest for predict,
a ClassificationRequest or RegressionRequest, or the Input of one, for classify
and regress. Another body that is not empty, such as text, JSON, bytes or a
Map, is not parsed and [...]
"producerOut": { "carriers": [ "pojo" ], "byOption": [ { "option": "api",
"value": "model-status", "carriers": [ "pojo" ], "javaType":
"tensorflow.serving.GetModelStatus$GetModelStatusResponse" }, { "option":
"api", "value": "model-metadata", "carriers": [ "pojo" ], "javaType":
"tensorflow.serving.GetModelMetadata$GetModelMetadataResponse" }, { "option":
"api", "value": "classify", "carriers": [ "pojo" ], "javaType":
"tensorflow.serving.Classification$ClassificationResponse" }, { "op [...]
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tensorflow-serving-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tensorflow-serving-component.adoc
index 0f3d5b3b7e32..6b88daf4333b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tensorflow-serving-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tensorflow-serving-component.adoc
@@ -102,6 +102,12 @@ tensorflow-serving:<api>[?options]
| `PredictResponse` footnote:[`tensorflow.serving.Predict.PredictResponse`]
|===
+The input body is the protobuf request of the API or its `Builder`, and for
`classify` and `regress` also an `Input`
+or its `Builder`. The component does not parse any other body that is not
empty, such as text, JSON, bytes or a
+`Map`: the exchange fails with an `InvalidPayloadException` that names the
request type of the API, so build the
+protobuf request before the endpoint. Without a body, or with an empty one (a
blank `String` or an empty `byte[]`),
+the request is built from the options and headers only, which is all that
`model-status` and `model-metadata` need.
+
== Examples
=== Model status API
@@ -210,6 +216,9 @@ YAML::
Currently, TensorFlow Serving only supports `signature_def` as the metadata
field.
See:
https://github.com/tensorflow/serving/blob/2.18.0/tensorflow_serving/apis/get_model_metadata.proto#L26-L28[get_model_metadata.proto]
+The component asks for `signature_def`, unless the `GetModelMetadataRequest`
in the body names the metadata fields
+itself.
+
.Fetch model metadata
[tabs]
diff --git
a/components/camel-ai/camel-tensorflow-serving/src/generated/resources/META-INF/org/apache/camel/component/tensorflow/serving/tensorflow-serving.json
b/components/camel-ai/camel-tensorflow-serving/src/generated/resources/META-INF/org/apache/camel/component/tensorflow/serving/tensorflow-serving.json
index d7692d3f6acd..43fbf4183aac 100644
---
a/components/camel-ai/camel-tensorflow-serving/src/generated/resources/META-INF/org/apache/camel/component/tensorflow/serving/tensorflow-serving.json
+++
b/components/camel-ai/camel-tensorflow-serving/src/generated/resources/META-INF/org/apache/camel/component/tensorflow/serving/tensorflow-serving.json
@@ -43,7 +43,7 @@
"CamelTensorFlowServingSignatureName": { "index": 5, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "A named signature to evaluate. If
unspecified, the default signature will be used.", "constantName":
"org.apache.camel.component.tensorflow.serving.TensorFlowServingConstants#SIGNATURE_NAME"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "pojo" ], "byOption": [ { "option": "api",
"value": "model-status", "carriers": [ "any" ] }, { "option": "api", "value":
"model-metadata", "carriers": [ "any" ] } ], "description": "The protobuf
request of the api, or its Builder: a PredictRequest for predict, a
ClassificationRequest or RegressionRequest, or the Input of one, for classify
and regress. Another body, such as text, JSON or bytes, is not read: an empty
request is sent and the server rejects [...]
+ "producerIn": { "carriers": [ "pojo" ], "byOption": [ { "option": "api",
"value": "model-status", "carriers": [ "null", "pojo" ] }, { "option": "api",
"value": "model-metadata", "carriers": [ "null", "pojo" ] } ], "description":
"The protobuf request of the api, or its Builder: a PredictRequest for predict,
a ClassificationRequest or RegressionRequest, or the Input of one, for classify
and regress. Another body that is not empty, such as text, JSON, bytes or a
Map, is not parsed and [...]
"producerOut": { "carriers": [ "pojo" ], "byOption": [ { "option": "api",
"value": "model-status", "carriers": [ "pojo" ], "javaType":
"tensorflow.serving.GetModelStatus$GetModelStatusResponse" }, { "option":
"api", "value": "model-metadata", "carriers": [ "pojo" ], "javaType":
"tensorflow.serving.GetModelMetadata$GetModelMetadataResponse" }, { "option":
"api", "value": "classify", "carriers": [ "pojo" ], "javaType":
"tensorflow.serving.Classification$ClassificationResponse" }, { "op [...]
},
"properties": {
diff --git
a/components/camel-ai/camel-tensorflow-serving/src/main/docs/tensorflow-serving-component.adoc
b/components/camel-ai/camel-tensorflow-serving/src/main/docs/tensorflow-serving-component.adoc
index 0f3d5b3b7e32..6b88daf4333b 100644
---
a/components/camel-ai/camel-tensorflow-serving/src/main/docs/tensorflow-serving-component.adoc
+++
b/components/camel-ai/camel-tensorflow-serving/src/main/docs/tensorflow-serving-component.adoc
@@ -102,6 +102,12 @@ tensorflow-serving:<api>[?options]
| `PredictResponse` footnote:[`tensorflow.serving.Predict.PredictResponse`]
|===
+The input body is the protobuf request of the API or its `Builder`, and for
`classify` and `regress` also an `Input`
+or its `Builder`. The component does not parse any other body that is not
empty, such as text, JSON, bytes or a
+`Map`: the exchange fails with an `InvalidPayloadException` that names the
request type of the API, so build the
+protobuf request before the endpoint. Without a body, or with an empty one (a
blank `String` or an empty `byte[]`),
+the request is built from the options and headers only, which is all that
`model-status` and `model-metadata` need.
+
== Examples
=== Model status API
@@ -210,6 +216,9 @@ YAML::
Currently, TensorFlow Serving only supports `signature_def` as the metadata
field.
See:
https://github.com/tensorflow/serving/blob/2.18.0/tensorflow_serving/apis/get_model_metadata.proto#L26-L28[get_model_metadata.proto]
+The component asks for `signature_def`, unless the `GetModelMetadataRequest`
in the body names the metadata fields
+itself.
+
.Fetch model metadata
[tabs]
diff --git
a/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpoint.java
b/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpoint.java
index bbf106112137..204f13cec27a 100644
---
a/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpoint.java
+++
b/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingEndpoint.java
@@ -42,16 +42,23 @@ import tensorflow.serving.PredictionServiceGrpc;
category = { Category.AI }, headersClass =
TensorFlowServingConstants.class)
@BodyTypes(producerIn = @BodyType(value = Carrier.POJO,
byOption = {
- @BodyType.When(option = "api", value
= "model-status", carriers = Carrier.ANY),
- @BodyType.When(option = "api", value
= "model-metadata", carriers = Carrier.ANY) },
+ @BodyType.When(option = "api", value
= "model-status",
+ carriers = {
Carrier.NULL, Carrier.POJO }),
+ @BodyType.When(option = "api", value
= "model-metadata",
+ carriers = {
Carrier.NULL, Carrier.POJO }) },
description = "The protobuf request of the
api, or its Builder: a PredictRequest for"
- + " predict, a
ClassificationRequest or RegressionRequest, or the Input of"
- + " one, for classify and
regress. Another body, such as text, JSON or"
- + " bytes, is not read: an
empty request is sent and the server rejects it."
- + " model-status and
model-metadata need no body, and read a"
- + " GetModelStatusRequest or
GetModelMetadataRequest when it is one. The"
- + " model comes from the
CamelTensorFlowServingModelName header, or else"
- + " the modelName option"),
+ + " predict, a
ClassificationRequest or RegressionRequest, or the Input"
+ + " of one, for classify and
regress. Another body that is not empty,"
+ + " such as text, JSON, bytes
or a Map, is not parsed and fails the"
+ + " exchange with an
InvalidPayloadException that names the request of"
+ + " the api: build the
protobuf request before the endpoint. No body, a"
+ + " blank String or an empty
byte array builds the request from the"
+ + " options and headers only,
which is all model-status and"
+ + " model-metadata need; they
read a GetModelStatusRequest or"
+ + " GetModelMetadataRequest
when the body is one, and model-metadata"
+ + " asks for signature_def
unless the request names its metadata fields."
+ + " The model comes from the
CamelTensorFlowServingModelName header, or"
+ + " else the modelName
option"),
producerOut = @BodyType(value = Carrier.POJO,
byOption = {
@BodyType.When(option = "api",
value = "model-status", carriers = Carrier.POJO,
diff --git
a/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingProducer.java
b/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingProducer.java
index 398b3ef3281d..15e143280338 100644
---
a/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingProducer.java
+++
b/components/camel-ai/camel-tensorflow-serving/src/main/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingProducer.java
@@ -21,6 +21,7 @@ import java.util.Optional;
import com.google.protobuf.GeneratedMessage;
import com.google.protobuf.Int64Value;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.support.DefaultProducer;
import tensorflow.serving.Classification;
@@ -34,6 +35,12 @@ import tensorflow.serving.RegressionOuterClass;
public class TensorFlowServingProducer extends DefaultProducer {
+ /** The end of the error of an api that reads its request from the body. */
+ private static final String BUILD_THE_REQUEST = ": build the protobuf
request before the endpoint";
+ /** The end of the error of an api that needs no body. */
+ private static final String OR_NO_BODY
+ = ", and builds the request from the options and headers when
there is no body, or an empty one";
+
private final String api;
private final ModelServiceGrpc.ModelServiceBlockingStub modelService;
private final PredictionServiceGrpc.PredictionServiceBlockingStub
predictionService;
@@ -92,9 +99,35 @@ public class TensorFlowServingProducer extends
DefaultProducer {
return builder;
}
- private GetModelStatus.GetModelStatusResponse modelStatus(Exchange
exchange) {
+ /**
+ * The body as the request of the api. No body, or an empty one, gives
null: the request is then built from the
+ * options and headers only. Another body that is not the request and does
not convert to it, such as text, JSON,
+ * bytes or a Map, fails instead of being dropped, as the request would be
sent without it.
+ */
+ private <T> T requestOf(Exchange exchange, Class<T> type, String reads)
throws InvalidPayloadException {
Message message = exchange.getMessage();
- GetModelStatus.GetModelStatusRequest request =
message.getBody(GetModelStatus.GetModelStatusRequest.class);
+ if (isEmpty(message.getBody())) {
+ return null;
+ }
+ T request = message.getBody(type);
+ if (request == null) {
+ throw new InvalidPayloadException(exchange, type, message, "the "
+ api + " api reads " + reads);
+ }
+ return request;
+ }
+
+ /**
+ * No body, or an empty one that an earlier step can leave, such as a
blank String or an empty byte array.
+ */
+ private static boolean isEmpty(Object body) {
+ return body == null
+ || (body instanceof String text && text.isBlank())
+ || (body instanceof byte[] bytes && bytes.length == 0);
+ }
+
+ private GetModelStatus.GetModelStatusResponse modelStatus(Exchange
exchange) throws InvalidPayloadException {
+ GetModelStatus.GetModelStatusRequest request = requestOf(exchange,
GetModelStatus.GetModelStatusRequest.class,
+ "a GetModelStatusRequest or its Builder" + OR_NO_BODY);
GetModelStatus.GetModelStatusRequest.Builder builder =
GetModelStatus.GetModelStatusRequest.newBuilder()
.setModelSpec(modelSpec(exchange));
if (request != null) {
@@ -103,21 +136,26 @@ public class TensorFlowServingProducer extends
DefaultProducer {
return modelService.getModelStatus(builder.build());
}
- private GetModelMetadata.GetModelMetadataResponse modelMetadata(Exchange
exchange) {
- Message message = exchange.getMessage();
- GetModelMetadata.GetModelMetadataRequest request =
message.getBody(GetModelMetadata.GetModelMetadataRequest.class);
+ private GetModelMetadata.GetModelMetadataResponse modelMetadata(Exchange
exchange) throws InvalidPayloadException {
+ GetModelMetadata.GetModelMetadataRequest request = requestOf(exchange,
+ GetModelMetadata.GetModelMetadataRequest.class,
+ "a GetModelMetadataRequest or its Builder" + OR_NO_BODY);
GetModelMetadata.GetModelMetadataRequest.Builder builder =
GetModelMetadata.GetModelMetadataRequest.newBuilder()
- .setModelSpec(modelSpec(exchange))
- .addMetadataField("signature_def");
+ .setModelSpec(modelSpec(exchange));
if (request != null) {
builder.mergeFrom(request);
}
+ // the metadata fields of the request, or else signature_def, the only
field TensorFlow Serving supports
+ // (merging appends a repeated field, so adding it before the merge
would ask for it twice)
+ if (builder.getMetadataFieldCount() == 0) {
+ builder.addMetadataField("signature_def");
+ }
return predictionService.getModelMetadata(builder.build());
}
- private Classification.ClassificationResponse classify(Exchange exchange) {
- Message message = exchange.getMessage();
- Classification.ClassificationRequest request =
message.getBody(Classification.ClassificationRequest.class);
+ private Classification.ClassificationResponse classify(Exchange exchange)
throws InvalidPayloadException {
+ Classification.ClassificationRequest request = requestOf(exchange,
Classification.ClassificationRequest.class,
+ "a ClassificationRequest, an Input, or the Builder of one" +
BUILD_THE_REQUEST);
Classification.ClassificationRequest.Builder builder =
Classification.ClassificationRequest.newBuilder()
.setModelSpec(modelSpec(exchange));
if (request != null) {
@@ -126,9 +164,10 @@ public class TensorFlowServingProducer extends
DefaultProducer {
return predictionService.classify(builder.build());
}
- private RegressionOuterClass.RegressionResponse regress(Exchange exchange)
{
- Message message = exchange.getMessage();
- RegressionOuterClass.RegressionRequest request =
message.getBody(RegressionOuterClass.RegressionRequest.class);
+ private RegressionOuterClass.RegressionResponse regress(Exchange exchange)
throws InvalidPayloadException {
+ RegressionOuterClass.RegressionRequest request = requestOf(
+ exchange, RegressionOuterClass.RegressionRequest.class,
+ "a RegressionRequest, an Input, or the Builder of one" +
BUILD_THE_REQUEST);
RegressionOuterClass.RegressionRequest.Builder builder =
RegressionOuterClass.RegressionRequest.newBuilder()
.setModelSpec(modelSpec(exchange));
if (request != null) {
@@ -137,9 +176,9 @@ public class TensorFlowServingProducer extends
DefaultProducer {
return predictionService.regress(builder.build());
}
- private Predict.PredictResponse predict(Exchange exchange) {
- Message message = exchange.getMessage();
- Predict.PredictRequest request =
message.getBody(Predict.PredictRequest.class);
+ private Predict.PredictResponse predict(Exchange exchange) throws
InvalidPayloadException {
+ Predict.PredictRequest request = requestOf(exchange,
Predict.PredictRequest.class,
+ "a PredictRequest or its Builder" + BUILD_THE_REQUEST);
Predict.PredictRequest.Builder builder =
Predict.PredictRequest.newBuilder()
.setModelSpec(modelSpec(exchange));
if (request != null) {
diff --git
a/components/camel-ai/camel-tensorflow-serving/src/test/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingProducerBodyTest.java
b/components/camel-ai/camel-tensorflow-serving/src/test/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingProducerBodyTest.java
new file mode 100644
index 000000000000..a986a9ca257a
--- /dev/null
+++
b/components/camel-ai/camel-tensorflow-serving/src/test/java/org/apache/camel/component/tensorflow/serving/TensorFlowServingProducerBodyTest.java
@@ -0,0 +1,326 @@
+/*
+ * 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.tensorflow.serving;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import io.grpc.Grpc;
+import io.grpc.InsecureServerCredentials;
+import io.grpc.Server;
+import io.grpc.stub.StreamObserver;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.tensorflow.example.Example;
+import org.tensorflow.example.Feature;
+import org.tensorflow.example.Features;
+import org.tensorflow.example.FloatList;
+import org.tensorflow.framework.DataType;
+import org.tensorflow.framework.TensorProto;
+import tensorflow.serving.Classification;
+import tensorflow.serving.GetModelMetadata;
+import tensorflow.serving.GetModelStatus;
+import tensorflow.serving.InputOuterClass;
+import tensorflow.serving.Model;
+import tensorflow.serving.ModelServiceGrpc;
+import tensorflow.serving.Predict;
+import tensorflow.serving.PredictionServiceGrpc;
+import tensorflow.serving.RegressionOuterClass;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The body the producer sends, against a local gRPC server that records the
requests it receives.
+ */
+class TensorFlowServingProducerBodyTest extends CamelTestSupport {
+
+ private static final String MODEL = "half_plus_two";
+ private static final String JSON = "{\"inputs\": {\"x\": [1.0, 2.0,
5.0]}}";
+
+ private static final BlockingQueue<Object> REQUESTS = new
LinkedBlockingQueue<>();
+ private static Server server;
+
+ @BeforeAll
+ static void startServer() throws IOException {
+ server = Grpc.newServerBuilderForPort(0,
InsecureServerCredentials.create())
+ .addService(new RecordingModelService())
+ .addService(new RecordingPredictionService())
+ .build()
+ .start();
+ }
+
+ @AfterAll
+ static void stopServer() throws InterruptedException {
+ server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
+ }
+
+ @BeforeEach
+ void clearRequests() {
+ REQUESTS.clear();
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getComponent("tensorflow-serving",
TensorFlowServingComponent.class).getConfiguration()
+ .setTarget("localhost:" + server.getPort());
+ return context;
+ }
+
+ static Stream<Arguments> bodiesThatAreNotTheRequest() {
+ byte[] bytes = JSON.getBytes(StandardCharsets.UTF_8);
+ Map<String, Object> map = Map.of("inputs", Map.of("x", List.of(1.0,
2.0, 5.0)));
+ return Stream.of(
+ Arguments.of("predict", Predict.PredictRequest.class, JSON),
+ Arguments.of("predict", Predict.PredictRequest.class, bytes),
+ Arguments.of("predict", Predict.PredictRequest.class, map),
+ Arguments.of("classify",
Classification.ClassificationRequest.class, JSON),
+ // the request of another api
+ Arguments.of("classify",
Classification.ClassificationRequest.class, predictRequest()),
+ Arguments.of("regress",
RegressionOuterClass.RegressionRequest.class, map),
+ Arguments.of("model-status",
GetModelStatus.GetModelStatusRequest.class, MODEL),
+ Arguments.of("model-metadata",
GetModelMetadata.GetModelMetadataRequest.class, bytes));
+ }
+
+ @ParameterizedTest
+ @MethodSource("bodiesThatAreNotTheRequest")
+ void bodyThatIsNotTheRequestFails(String api, Class<?> requestType, Object
body) {
+ Exchange exchange = send(api, body);
+
+ InvalidPayloadException e =
assertInstanceOf(InvalidPayloadException.class, exchange.getException());
+ assertEquals(requestType, e.getType());
+ assertTrue(e.getMessage().contains(requestType.getCanonicalName()),
e.getMessage());
+ assertTrue(e.getMessage().contains("the " + api + " api reads"),
e.getMessage());
+ assertTrue(REQUESTS.isEmpty(), "No request must be sent");
+ }
+
+ @Test
+ void predictSendsTheRequestOfTheBody() {
+ Predict.PredictRequest request = predictRequest();
+
+ Exchange exchange = send("predict", request);
+
+ assertNull(exchange.getException());
+ Predict.PredictRequest sent = sent(Predict.PredictRequest.class);
+ assertEquals(MODEL, sent.getModelSpec().getName());
+ assertEquals(request.getInputsMap(), sent.getInputsMap());
+ assertInstanceOf(Predict.PredictResponse.class,
exchange.getMessage().getBody());
+ }
+
+ @Test
+ void predictSendsTheRequestOfABuilderBody() {
+ Predict.PredictRequest.Builder builder = predictRequest().toBuilder();
+
+ Exchange exchange = send("predict", builder);
+
+ assertNull(exchange.getException());
+ assertEquals(builder.getInputsMap(),
sent(Predict.PredictRequest.class).getInputsMap());
+ }
+
+ @Test
+ void predictWithoutBodySendsTheModelOnly() {
+ Exchange exchange = send("predict", null);
+
+ assertNull(exchange.getException());
+ Predict.PredictRequest sent = sent(Predict.PredictRequest.class);
+ assertEquals(MODEL, sent.getModelSpec().getName());
+ assertEquals(0, sent.getInputsCount());
+ }
+
+ @Test
+ void classifySendsTheInputOfTheBody() {
+ InputOuterClass.Input input = input();
+
+ Exchange exchange = send("classify", input);
+
+ assertNull(exchange.getException());
+ Classification.ClassificationRequest sent =
sent(Classification.ClassificationRequest.class);
+ assertEquals(MODEL, sent.getModelSpec().getName());
+ assertEquals(input, sent.getInput());
+ }
+
+ @Test
+ void regressSendsTheInputOfTheBody() {
+ InputOuterClass.Input input = input();
+
+ Exchange exchange = send("regress", input);
+
+ assertNull(exchange.getException());
+ assertEquals(input,
sent(RegressionOuterClass.RegressionRequest.class).getInput());
+ }
+
+ @Test
+ void modelStatusWithoutBodyAsksForTheModelOfTheOptions() {
+ Exchange exchange = send("model-status", null);
+
+ assertNull(exchange.getException());
+ assertEquals(MODEL,
sent(GetModelStatus.GetModelStatusRequest.class).getModelSpec().getName());
+ assertInstanceOf(GetModelStatus.GetModelStatusResponse.class,
exchange.getMessage().getBody());
+ }
+
+ static Stream<Object> emptyBodies() {
+ return Stream.of("", " \n", new byte[0]);
+ }
+
+ @ParameterizedTest
+ @MethodSource("emptyBodies")
+ void modelStatusWithAnEmptyBodyAsksForTheModelOfTheOptions(Object body) {
+ Exchange exchange = send("model-status", body);
+
+ assertNull(exchange.getException());
+ assertEquals(MODEL,
sent(GetModelStatus.GetModelStatusRequest.class).getModelSpec().getName());
+ }
+
+ @Test
+ void modelMetadataWithoutBodyAsksForSignatureDef() {
+ Exchange exchange = send("model-metadata", null);
+
+ assertNull(exchange.getException());
+ GetModelMetadata.GetModelMetadataRequest sent =
sent(GetModelMetadata.GetModelMetadataRequest.class);
+ assertEquals(MODEL, sent.getModelSpec().getName());
+ assertEquals(List.of("signature_def"), sent.getMetadataFieldList());
+ }
+
+ @Test
+ void modelMetadataAsksOnceForSignatureDefWhenTheRequestNamesIt() {
+ GetModelMetadata.GetModelMetadataRequest request =
GetModelMetadata.GetModelMetadataRequest.newBuilder()
+ .addMetadataField("signature_def")
+ .build();
+
+ Exchange exchange = send("model-metadata", request);
+
+ assertNull(exchange.getException());
+ GetModelMetadata.GetModelMetadataRequest sent =
sent(GetModelMetadata.GetModelMetadataRequest.class);
+ assertEquals(MODEL, sent.getModelSpec().getName());
+ assertEquals(List.of("signature_def"), sent.getMetadataFieldList());
+ }
+
+ @Test
+ void modelMetadataAsksForSignatureDefWhenTheRequestNamesNoField() {
+ GetModelMetadata.GetModelMetadataRequest request =
GetModelMetadata.GetModelMetadataRequest.newBuilder()
+ .setModelSpec(Model.ModelSpec.newBuilder().setName("other"))
+ .build();
+
+ Exchange exchange = send("model-metadata", request);
+
+ assertNull(exchange.getException());
+ GetModelMetadata.GetModelMetadataRequest sent =
sent(GetModelMetadata.GetModelMetadataRequest.class);
+ assertEquals("other", sent.getModelSpec().getName());
+ assertEquals(List.of("signature_def"), sent.getMetadataFieldList());
+ }
+
+ private Exchange send(String api, Object body) {
+ return template.send("tensorflow-serving:" + api + "?modelName=" +
MODEL, e -> e.getMessage().setBody(body));
+ }
+
+ private static <T> T sent(Class<T> type) {
+ return assertInstanceOf(type, REQUESTS.poll(), "The request the server
received");
+ }
+
+ private static Predict.PredictRequest predictRequest() {
+ return Predict.PredictRequest.newBuilder()
+ .putInputs("x", TensorProto.newBuilder()
+ .setDtype(DataType.DT_FLOAT)
+ .addFloatVal(1.0f)
+ .addFloatVal(2.0f)
+ .addFloatVal(5.0f)
+ .build())
+ .build();
+ }
+
+ private static InputOuterClass.Input input() {
+ return InputOuterClass.Input.newBuilder()
+ .setExampleList(InputOuterClass.ExampleList.newBuilder()
+ .addExamples(Example.newBuilder()
+ .setFeatures(Features.newBuilder()
+ .putFeature("x", Feature.newBuilder()
+
.setFloatList(FloatList.newBuilder().addValue(1.0f))
+ .build()))))
+ .build();
+ }
+
+ private static <T> void reply(Object request, T response,
StreamObserver<T> observer) {
+ REQUESTS.add(request);
+ observer.onNext(response);
+ observer.onCompleted();
+ }
+
+ private static final class RecordingModelService extends
ModelServiceGrpc.ModelServiceImplBase {
+
+ @Override
+ public void getModelStatus(
+ GetModelStatus.GetModelStatusRequest request,
+ StreamObserver<GetModelStatus.GetModelStatusResponse>
observer) {
+ reply(request,
GetModelStatus.GetModelStatusResponse.getDefaultInstance(), observer);
+ }
+ }
+
+ private static final class RecordingPredictionService extends
PredictionServiceGrpc.PredictionServiceImplBase {
+
+ @Override
+ public void classify(
+ Classification.ClassificationRequest request,
+ StreamObserver<Classification.ClassificationResponse>
observer) {
+ Classification.ClassificationResponse response =
Classification.ClassificationResponse.newBuilder()
+ .setModelSpec(request.getModelSpec())
+ .build();
+ reply(request, response, observer);
+ }
+
+ @Override
+ public void regress(
+ RegressionOuterClass.RegressionRequest request,
+ StreamObserver<RegressionOuterClass.RegressionResponse>
observer) {
+ RegressionOuterClass.RegressionResponse response =
RegressionOuterClass.RegressionResponse.newBuilder()
+ .setModelSpec(request.getModelSpec())
+ .build();
+ reply(request, response, observer);
+ }
+
+ @Override
+ public void predict(Predict.PredictRequest request,
StreamObserver<Predict.PredictResponse> observer) {
+ reply(request,
Predict.PredictResponse.newBuilder().setModelSpec(request.getModelSpec()).build(),
observer);
+ }
+
+ @Override
+ public void getModelMetadata(
+ GetModelMetadata.GetModelMetadataRequest request,
+ StreamObserver<GetModelMetadata.GetModelMetadataResponse>
observer) {
+ reply(request,
+
GetModelMetadata.GetModelMetadataResponse.newBuilder().setModelSpec(request.getModelSpec()).build(),
+ observer);
+ }
+ }
+}
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 092914b318d6..7350bf33ef9b 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
@@ -4359,6 +4359,22 @@ To keep the previous behaviour, filter the exchanges
with a bad status out right
`filter(simple("${body.statusCode.bad} == false"))`. This also drops the other
bad statuses, such as the refusal of a
node, and the current values delivered after an outage still arrive.
+=== camel-tensorflow-serving - a body that is not the request of the api fails
+
+The `tensorflow-serving` producer converted the body to the protobuf request
of the api, and dropped a body that
+did not convert, such as text, JSON, bytes or a `Map`, without an error. The
request was then sent without it, so
+`predict`, `classify` and `regress` failed on the server with an error about
the missing inputs, and `model-status`
+and `model-metadata` ignored the body. Such a body, when it is not empty, now
fails the exchange with an
+`InvalidPayloadException` that names the request type of the api, before
anything is sent. Build the protobuf request
+before the endpoint, or remove a body that an earlier step left before calling
`model-status` or `model-metadata`.
+
+No body, a blank `String` or an empty `byte[]` still sends a request built
from the options and headers, and a body
+that is the request or its `Builder`, or an `Input` for `classify` and
`regress`, is sent as before.
+
+`model-metadata` asked for `signature_def` in addition to the metadata fields
of a `GetModelMetadataRequest` body, so
+a request that named `signature_def` asked for it twice. It now asks for
`signature_def` only when the request names
+no metadata field.
+
=== camel-milo - an awaited write, call or read that fails now fails the
exchange
An awaited `milo-client` write, method call or read that fails at the service
or transport level, for example a
