This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26174-stitch-map-types in repository https://gitbox.apache.org/repos/asf/camel.git
commit b2e30ccbc09c758d61ffe705f2d441ab8ec0905a Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 23:30:31 2026 +0100 CAMEL-26174: camel-stitch - accept key_names as a String and any Number as sequence in a Map body, and set the response headers for an empty response body A Map body with key_names as a String failed with a ClassCastException, and a message Map with an Integer sequence failed the cast to Long. key_names now takes one key name or comma separated names, and sequence any Number. A response with an empty body produced no StitchResponse, so the exchange kept the request body without the CamelStitchCode, CamelStitchStatus and CamelStitchHeaders headers, and an error response with an empty body was ignored. An empty body is now handled as an empty JSON object. A message without a sequence got the time the StitchMessage class was loaded, the same for every message; it now gets the time it is built. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/catalog/components/stitch.json | 2 +- .../camel/catalog/docs/stitch-component.adoc | 3 +- components/camel-stitch/pom.xml | 6 + .../org/apache/camel/component/stitch/stitch.json | 2 +- .../src/main/docs/stitch-component.adoc | 3 +- .../camel/component/stitch/StitchEndpoint.java | 3 +- .../component/stitch/client/StitchClientImpl.java | 2 + .../stitch/client/models/StitchMessage.java | 16 ++- .../stitch/client/models/StitchRequestBody.java | 14 ++- .../stitch/client/StitchClientImplTest.java | 124 +++++++++++++++++++++ .../stitch/client/models/StitchMessageTest.java | 35 ++++++ .../client/models/StitchRequestBodyTest.java | 35 ++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 ++ 13 files changed, 243 insertions(+), 12 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/stitch.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/stitch.json index c69d4ab09d98..48aa12dc8159 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/stitch.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/stitch.json @@ -44,7 +44,7 @@ "CamelStitchStatus": { "index": 5, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The status message that Stitch returns after sending the data through Stitch Import API.", "constantName": "org.apache.camel.component.stitch.StitchConstants#STATUS" } }, "bodyTypes": { - "producerIn": { "carriers": [ "pojo", "map", "list" ], "description": "A StitchRequestBody, a StitchMessage, a Map of the Stitch batch JSON (table_name, schema, messages, key_names as a list), or a List of StitchMessage, Map of a message, StitchRequestBody, Message or Exchange. Text, bytes, a stream or no body fails. The tableName, keyNames and schema options or headers set the rest of the batch" }, + "producerIn": { "carriers": [ "pojo", "map", "list" ], "description": "A StitchRequestBody, a StitchMessage, a Map of the Stitch batch JSON (table_name, schema, messages, key_names as a list or a comma separated String), or a List of StitchMessage, Map of a message, StitchRequestBody, Message or Exchange. Text, bytes, a stream or no body fails. The tableName, keyNames and schema options or headers set the rest of the batch" }, "producerOut": { "carriers": [ "text" ], "javaType": "java.lang.String", "description": "The message of the response of Stitch, empty when it has none; the HTTP status code, the status and the response headers are in the CamelStitchCode, CamelStitchStatus and CamelStitchHeaders headers" } }, "properties": { diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/stitch-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/stitch-component.adoc index c33945f9cb63..ebec20288fa2 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/stitch-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/stitch-component.adoc @@ -172,7 +172,8 @@ from("direct:sendStitch") ---- In YAML, a Groovy map has the same structure. The keys are those of the Stitch JSON message: `table_name`, `schema`, -`messages` and `key_names`: +`messages` and `key_names`. The `key_names` can be a list, or a String with one key name or comma separated key names. +The `sequence` of a message can be any number, and when it is not set, it is the time the message is built (in millis): [source,yaml] ---- diff --git a/components/camel-stitch/pom.xml b/components/camel-stitch/pom.xml index d4da1a108d18..9f4a3f669089 100644 --- a/components/camel-stitch/pom.xml +++ b/components/camel-stitch/pom.xml @@ -65,5 +65,11 @@ <version>${mockito-version}</version> <scope>test</scope> </dependency> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <version>${awaitility-version}</version> + <scope>test</scope> + </dependency> </dependencies> </project> diff --git a/components/camel-stitch/src/generated/resources/META-INF/org/apache/camel/component/stitch/stitch.json b/components/camel-stitch/src/generated/resources/META-INF/org/apache/camel/component/stitch/stitch.json index c69d4ab09d98..48aa12dc8159 100644 --- a/components/camel-stitch/src/generated/resources/META-INF/org/apache/camel/component/stitch/stitch.json +++ b/components/camel-stitch/src/generated/resources/META-INF/org/apache/camel/component/stitch/stitch.json @@ -44,7 +44,7 @@ "CamelStitchStatus": { "index": 5, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The status message that Stitch returns after sending the data through Stitch Import API.", "constantName": "org.apache.camel.component.stitch.StitchConstants#STATUS" } }, "bodyTypes": { - "producerIn": { "carriers": [ "pojo", "map", "list" ], "description": "A StitchRequestBody, a StitchMessage, a Map of the Stitch batch JSON (table_name, schema, messages, key_names as a list), or a List of StitchMessage, Map of a message, StitchRequestBody, Message or Exchange. Text, bytes, a stream or no body fails. The tableName, keyNames and schema options or headers set the rest of the batch" }, + "producerIn": { "carriers": [ "pojo", "map", "list" ], "description": "A StitchRequestBody, a StitchMessage, a Map of the Stitch batch JSON (table_name, schema, messages, key_names as a list or a comma separated String), or a List of StitchMessage, Map of a message, StitchRequestBody, Message or Exchange. Text, bytes, a stream or no body fails. The tableName, keyNames and schema options or headers set the rest of the batch" }, "producerOut": { "carriers": [ "text" ], "javaType": "java.lang.String", "description": "The message of the response of Stitch, empty when it has none; the HTTP status code, the status and the response headers are in the CamelStitchCode, CamelStitchStatus and CamelStitchHeaders headers" } }, "properties": { diff --git a/components/camel-stitch/src/main/docs/stitch-component.adoc b/components/camel-stitch/src/main/docs/stitch-component.adoc index c33945f9cb63..ebec20288fa2 100644 --- a/components/camel-stitch/src/main/docs/stitch-component.adoc +++ b/components/camel-stitch/src/main/docs/stitch-component.adoc @@ -172,7 +172,8 @@ from("direct:sendStitch") ---- In YAML, a Groovy map has the same structure. The keys are those of the Stitch JSON message: `table_name`, `schema`, -`messages` and `key_names`: +`messages` and `key_names`. The `key_names` can be a list, or a String with one key name or comma separated key names. +The `sequence` of a message can be any number, and when it is not set, it is the time the message is built (in millis): [source,yaml] ---- diff --git a/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/StitchEndpoint.java b/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/StitchEndpoint.java index 675e693feb25..fc1094cb593d 100644 --- a/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/StitchEndpoint.java +++ b/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/StitchEndpoint.java @@ -41,7 +41,8 @@ import org.apache.camel.util.ObjectHelper; headersClass = StitchConstants.class) @BodyTypes(producerIn = @BodyType(value = { Carrier.POJO, Carrier.MAP, Carrier.LIST }, description = "A StitchRequestBody, a StitchMessage, a Map of the Stitch batch JSON" - + " (table_name, schema, messages, key_names as a list), or a List of" + + " (table_name, schema, messages, key_names as a list or a comma separated" + + " String), or a List of" + " StitchMessage, Map of a message, StitchRequestBody, Message or Exchange." + " Text, bytes, a stream or no body fails. The tableName, keyNames and schema" + " options or headers set the rest of the batch"), diff --git a/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/StitchClientImpl.java b/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/StitchClientImpl.java index 194b24cf482d..152e98ce0a88 100644 --- a/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/StitchClientImpl.java +++ b/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/StitchClientImpl.java @@ -88,6 +88,8 @@ public class StitchClientImpl implements StitchClient { .map(JsonUtils::convertJsonToMap) // if we get a problem parsing the json, we just return empty map for graceful handling .onErrorReturn(Collections.emptyMap()) + // an empty body emits nothing, so we use an empty map to still produce the response (or the error) + .defaultIfEmpty(Collections.emptyMap()) .handle((bodyMap, sink) -> { final StitchResponse stitchResponse = getStitchResponse(code, headers, bodyMap); diff --git a/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/models/StitchMessage.java b/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/models/StitchMessage.java index 71085bf644d3..235528f06fe8 100644 --- a/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/models/StitchMessage.java +++ b/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/models/StitchMessage.java @@ -32,7 +32,6 @@ public final class StitchMessage implements StitchModel { public static final String DATA = "data"; private static final Action DEFAULT_ACTION = Action.UPSERT; - private static final long DEFAULT_SEQUENCE = System.currentTimeMillis(); public enum Action { UPSERT @@ -55,13 +54,18 @@ public final class StitchMessage implements StitchModel { @SuppressWarnings("unchecked") public static Builder fromMap(final Map<String, Object> data) { final Action action = Action.valueOf(data.getOrDefault(ACTION, DEFAULT_ACTION.name()).toString().toUpperCase()); - final long sequence = ObjectHelper.cast(Long.class, data.getOrDefault(SEQUENCE, DEFAULT_SEQUENCE)); + final Object sequence = data.get(SEQUENCE); final Map<String, Object> inputData = ObjectHelper.cast(Map.class, data.getOrDefault(DATA, Collections.emptyMap())); - return new Builder() + final Builder builder = new Builder() .withAction(action) - .withData(inputData) - .withSequence(sequence); + .withData(inputData); + if (sequence != null) { + // a Map read from JSON or YAML has a small number as an Integer + builder.withSequence(sequence instanceof Number number + ? number.longValue() : ObjectHelper.cast(Long.class, sequence)); + } + return builder; } public Action getAction() { @@ -156,7 +160,7 @@ public final class StitchMessage implements StitchModel { } if (ObjectHelper.isEmpty(sequence)) { - sequence = DEFAULT_SEQUENCE; + sequence = System.currentTimeMillis(); } return new StitchMessage(action, sequence, data); diff --git a/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/models/StitchRequestBody.java b/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/models/StitchRequestBody.java index 5d0dcb7f1150..452f1ba8939f 100644 --- a/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/models/StitchRequestBody.java +++ b/components/camel-stitch/src/main/java/org/apache/camel/component/stitch/client/models/StitchRequestBody.java @@ -77,7 +77,7 @@ public final class StitchRequestBody implements StitchModel { .fromMap(ObjectHelper.cast(Map.class, message)) .build()) .collect(Collectors.toList()); - final Collection<String> keyNames = ObjectHelper.cast(Collection.class, data.get(KEY_NAMES)); + final Collection<String> keyNames = getKeyNames(data.get(KEY_NAMES)); return new Builder() .withSchema(schema) @@ -87,6 +87,18 @@ public final class StitchRequestBody implements StitchModel { .addMessages(messages); } + @SuppressWarnings("unchecked") + private static Collection<String> getKeyNames(final Object keyNames) { + if (keyNames instanceof String names) { + // one key name, or comma separated key names + return Arrays.stream(names.split(",")) + .map(String::trim) + .filter(ObjectHelper::isNotEmpty) + .collect(Collectors.toCollection(LinkedHashSet::new)); + } + return ObjectHelper.cast(Collection.class, keyNames); + } + public String getTableName() { return tableName; } diff --git a/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/StitchClientImplTest.java b/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/StitchClientImplTest.java new file mode 100644 index 000000000000..565e123431cc --- /dev/null +++ b/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/StitchClientImplTest.java @@ -0,0 +1,124 @@ +/* + * 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.stitch.client; + +import java.io.IOException; +import java.io.InputStream; +import java.net.InetSocketAddress; +import java.util.Map; + +import com.sun.net.httpserver.HttpServer; +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.stitch.StitchConstants; +import org.apache.camel.component.stitch.client.models.StitchException; +import org.apache.camel.component.stitch.client.models.StitchMessage; +import org.apache.camel.component.stitch.client.models.StitchSchema; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import reactor.netty.http.client.HttpClient; +import reactor.netty.resources.ConnectionProvider; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * Sends the batch to a local HTTP server that answers with an empty body. + */ +class StitchClientImplTest extends CamelTestSupport { + + private static HttpServer server; + private static ConnectionProvider connectionProvider; + private static volatile int statusCode; + + @BeforeAll + static void startServer() throws IOException { + server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + server.createContext("/v2/import/batch", httpExchange -> { + try (InputStream is = httpExchange.getRequestBody()) { + is.readAllBytes(); + } + httpExchange.getResponseHeaders().add("Test", "empty"); + // -1 is no response body + httpExchange.sendResponseHeaders(statusCode, -1); + httpExchange.close(); + }); + server.start(); + connectionProvider = ConnectionProvider.create("stitch-test"); + } + + @AfterAll + static void stopServer() { + server.stop(0); + connectionProvider.disposeLater().block(); + } + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext context = super.createCamelContext(); + context.getRegistry().bind("client", new StitchClientImpl( + HttpClient.create(connectionProvider), "http://localhost:"; + server.getAddress().getPort(), "dummy")); + return context; + } + + @Test + void testOkResponseWithEmptyBody() { + statusCode = 200; + + final Exchange exchange = template.send("direct:sendStitch", this::createMessage); + + assertNull(exchange.getException()); + assertEquals("", exchange.getMessage().getBody()); + assertEquals(200, exchange.getMessage().getHeader(StitchConstants.CODE)); + assertEquals("", exchange.getMessage().getHeader(StitchConstants.STATUS)); + assertEquals("empty", exchange.getMessage().getHeader(StitchConstants.HEADERS, Map.class).get("Test")); + } + + @Test + void testErrorResponseWithEmptyBody() { + statusCode = 500; + + final Exchange exchange = template.send("direct:sendStitch", this::createMessage); + + final StitchException exception = assertInstanceOf(StitchException.class, exchange.getException()); + assertEquals(500, exception.getResponse().getHttpStatusCode()); + } + + private void createMessage(Exchange exchange) { + exchange.getMessage().setHeader(StitchConstants.SCHEMA, + StitchSchema.builder().addKeyword("field_1", "string").build()); + exchange.getMessage().setHeader(StitchConstants.KEY_NAMES, "field_1"); + exchange.getMessage().setBody(StitchMessage.builder() + .withData("field_1", "data") + .build()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:sendStitch") + .to("stitch:table_1?stitchClient=#client&token=dummy"); + } + }; + } +} diff --git a/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/models/StitchMessageTest.java b/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/models/StitchMessageTest.java index 9ba12216bd41..96b0264450b8 100644 --- a/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/models/StitchMessageTest.java +++ b/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/models/StitchMessageTest.java @@ -18,12 +18,15 @@ package org.apache.camel.component.stitch.client.models; import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.TimeUnit; import org.apache.camel.component.stitch.client.JsonUtils; import org.junit.jupiter.api.Test; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; class StitchMessageTest { @@ -75,4 +78,36 @@ class StitchMessageTest { assertEquals(data, stitchMessage.getData()); assertEquals(123456L, stitchMessage.getSequence()); } + + @Test + void testIfCreateFromMapWithIntegerSequence() { + final Map<String, Object> message = new LinkedHashMap<>(); + // a Map read from JSON or YAML has a small number as an Integer + message.put(StitchMessage.SEQUENCE, 123456); + message.put(StitchMessage.DATA, Map.of("id", 2)); + + final StitchMessage stitchMessage = StitchMessage + .fromMap(message) + .build(); + + assertEquals(123456L, stitchMessage.getSequence()); + } + + @Test + void testIfDefaultSequenceIsTheTimeTheMessageIsBuilt() { + // make sure the StitchMessage class is loaded, and let the clock move on + StitchMessage.builder(); + final long loaded = System.currentTimeMillis(); + await().atMost(5, TimeUnit.SECONDS).until(() -> System.currentTimeMillis() > loaded); + + final long before = System.currentTimeMillis(); + final StitchMessage fromBuilder = StitchMessage.builder() + .withData("id", 2) + .build(); + final StitchMessage fromMap = StitchMessage.fromMap(Map.of(StitchMessage.DATA, Map.of("id", 3))) + .build(); + + assertTrue(fromBuilder.getSequence() >= before, "The default sequence should be the time the message is built"); + assertTrue(fromMap.getSequence() >= before, "The default sequence should be the time the message is built"); + } } diff --git a/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/models/StitchRequestBodyTest.java b/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/models/StitchRequestBodyTest.java index 39a6e2db2ebe..1b909177139c 100644 --- a/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/models/StitchRequestBodyTest.java +++ b/components/camel-stitch/src/test/java/org/apache/camel/component/stitch/client/models/StitchRequestBodyTest.java @@ -18,6 +18,7 @@ package org.apache.camel.component.stitch.client.models; import java.util.Collections; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import com.fasterxml.jackson.core.JsonProcessingException; @@ -125,4 +126,38 @@ class StitchRequestBodyTest { assertEquals("test_key", requestBody.getKeyNames().stream().findFirst().get()); } + @Test + void testIfCreateRequestBodyFromMapWithKeyNamesAsString() { + final Map<String, Object> data = createRequestMap(); + data.put(StitchRequestBody.KEY_NAMES, "id"); + + final StitchRequestBody requestBody = StitchRequestBody + .fromMap(data) + .build(); + + assertEquals(List.of("id"), List.copyOf(requestBody.getKeyNames())); + } + + @Test + void testIfCreateRequestBodyFromMapWithCommaSeparatedKeyNames() { + final Map<String, Object> data = createRequestMap(); + data.put(StitchRequestBody.KEY_NAMES, "id, name"); + + final StitchRequestBody requestBody = StitchRequestBody + .fromMap(data) + .build(); + + assertEquals(List.of("id", "name"), List.copyOf(requestBody.getKeyNames())); + } + + private static Map<String, Object> createRequestMap() { + final Map<String, Object> data = new LinkedHashMap<>(); + data.put(StitchRequestBody.TABLE_NAME, "my_table"); + data.put(StitchRequestBody.SCHEMA, + Collections.singletonMap("properties", + Collections.singletonMap("id", Collections.singletonMap("type", "integer")))); + data.put(StitchRequestBody.MESSAGES, + Collections.singletonList(Collections.singletonMap("data", Collections.singletonMap("id", 2)))); + return data; + } } 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..e590d70b5af5 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 @@ -632,6 +632,16 @@ closed that client for every other endpoint. Camel now closes only the clients it created. A client set with `amazonKinesisClient` or `amazonKinesisAsyncClient`, on the component or on an endpoint, is no longer closed when Camel stops: close it yourself when you no longer need it. +=== camel-stitch - a response with an empty body, and the default sequence of a message + +A response of Stitch with an empty body now sets the `CamelStitchCode`, `CamelStitchStatus` and `CamelStitchHeaders` +headers, and the body is an empty message. Before, the exchange kept the request body and got none of these headers. +An error response (HTTP status 300 or above) with an empty body now fails the exchange with a `StitchException`, as +any other error response does. Before, it was ignored. + +A `StitchMessage` without a `sequence` now gets the time it is built (in millis) as its sequence. Before, every message +got the same sequence: the time the `StitchMessage` class was loaded. + === camel-stripe - the CamelStripeOperation header chooses the operation The `CamelStripeOperation` header now overrides the operation of the endpoint, as documented. Before, the header was
