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 8a1fa14951b4 CAMEL-26172: camel-splunk-hec - send bytes, stream and
file bodies as text, and write the JSON before sending (#28998)
8a1fa14951b4 is described below
commit 8a1fa14951b488a69b6a1597f6970863217bfb35
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 07:37:28 2026 +0100
CAMEL-26172: camel-splunk-hec - send bytes, stream and file bodies as text,
and write the JSON before sending (#28998)
… text, and write the JSON before sending
The producer wrote the payload with Jackson inside an EntityTemplate, while
the
request was already being sent: a payload Jackson could not write (such as a
stream body) failed half way after a truncated, invalid JSON request had
reached
Splunk. The JSON is now written to a byte array first.
The body was put in the event as is, so Jackson wrote a byte array as
base64,
a File as its path and failed on a stream or the GenericFile of the file
consumer. These bodies are now converted to text.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../camel/catalog/components/splunk-hec.json | 2 +-
.../camel/catalog/docs/splunk-hec-component.adoc | 4 +
.../camel/component/splunkhec/splunk-hec.json | 2 +-
.../src/main/docs/splunk-hec-component.adoc | 4 +
.../component/splunkhec/SplunkHECEndpoint.java | 7 +-
.../component/splunkhec/SplunkHECProducer.java | 36 ++++++--
.../splunkhec/SplunkHECProducerHttpTest.java | 99 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 ++
8 files changed, 149 insertions(+), 14 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/splunk-hec.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/splunk-hec.json
index e1c35db33902..be2855f4497c 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/splunk-hec.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/splunk-hec.json
@@ -33,7 +33,7 @@
"CamelSplunkHECIndexTime": { "index": 0, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Epoch-formatted time. Specify with the time query string
parameter. Sets a default for all events in the request. The default time can
be overridden.", "constantName":
"org.apache.camel.component.splunkhec.SplunkHECConstants#INDEX_TIME" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text", "value", "map", "list", "pojo",
"null" ], "byOption": [ { "option": "headersOnly", "value": "true", "carriers":
[ "any" ] } ], "description": "The event, written as JSON by Jackson with the
headers beside it, or alone with bodyOnly. A stream fails, bytes become base64
and a File its path: convert them to text first. With headersOnly the body is
not sent" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "value",
"map", "list", "pojo", "null" ], "byOption": [ { "option": "headersOnly",
"value": "true", "carriers": [ "any" ] } ], "description": "The event, written
as JSON by Jackson with the headers beside it, or alone with bodyOnly. Bytes, a
stream or a file are sent as text. With headersOnly the body is not sent" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/splunk-hec-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/splunk-hec-component.adoc
index 1baf659cc83b..b9847eb189c7 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/splunk-hec-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/splunk-hec-component.adoc
@@ -48,6 +48,10 @@ The body must be serializable to JSON, so it may be sent to
Splunk.
A `String` or a `java.util.Map` object can be serialized easily.
+A byte array, an `InputStream`, a `Reader` or a file (such as the body of the
file consumer) is converted to text
+and sent as a JSON string. The payload is written as JSON before the request
is sent, so a body that cannot be
+serialized fails the exchange without sending anything to Splunk.
+
== Use Cases
The Splunk HEC endpoint may be used to stream data to Splunk for ingestion.
diff --git
a/components/camel-splunk-hec/src/generated/resources/META-INF/org/apache/camel/component/splunkhec/splunk-hec.json
b/components/camel-splunk-hec/src/generated/resources/META-INF/org/apache/camel/component/splunkhec/splunk-hec.json
index e1c35db33902..be2855f4497c 100644
---
a/components/camel-splunk-hec/src/generated/resources/META-INF/org/apache/camel/component/splunkhec/splunk-hec.json
+++
b/components/camel-splunk-hec/src/generated/resources/META-INF/org/apache/camel/component/splunkhec/splunk-hec.json
@@ -33,7 +33,7 @@
"CamelSplunkHECIndexTime": { "index": 0, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Epoch-formatted time. Specify with the time query string
parameter. Sets a default for all events in the request. The default time can
be overridden.", "constantName":
"org.apache.camel.component.splunkhec.SplunkHECConstants#INDEX_TIME" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text", "value", "map", "list", "pojo",
"null" ], "byOption": [ { "option": "headersOnly", "value": "true", "carriers":
[ "any" ] } ], "description": "The event, written as JSON by Jackson with the
headers beside it, or alone with bodyOnly. A stream fails, bytes become base64
and a File its path: convert them to text first. With headersOnly the body is
not sent" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "value",
"map", "list", "pojo", "null" ], "byOption": [ { "option": "headersOnly",
"value": "true", "carriers": [ "any" ] } ], "description": "The event, written
as JSON by Jackson with the headers beside it, or alone with bodyOnly. Bytes, a
stream or a file are sent as text. With headersOnly the body is not sent" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/components/camel-splunk-hec/src/main/docs/splunk-hec-component.adoc
b/components/camel-splunk-hec/src/main/docs/splunk-hec-component.adoc
index 1baf659cc83b..b9847eb189c7 100644
--- a/components/camel-splunk-hec/src/main/docs/splunk-hec-component.adoc
+++ b/components/camel-splunk-hec/src/main/docs/splunk-hec-component.adoc
@@ -48,6 +48,10 @@ The body must be serializable to JSON, so it may be sent to
Splunk.
A `String` or a `java.util.Map` object can be serialized easily.
+A byte array, an `InputStream`, a `Reader` or a file (such as the body of the
file consumer) is converted to text
+and sent as a JSON string. The payload is written as JSON before the request
is sent, so a body that cannot be
+serialized fails the exchange without sending anything to Splunk.
+
== Use Cases
The Splunk HEC endpoint may be used to stream data to Splunk for ingestion.
diff --git
a/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECEndpoint.java
b/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECEndpoint.java
index 122744454079..bd17dd6961ae 100644
---
a/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECEndpoint.java
+++
b/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECEndpoint.java
@@ -47,11 +47,12 @@ import
org.apache.commons.validator.routines.InetAddressValidator;
syntax = "splunk-hec:splunkURL", category = { Category.MONITORING
},
headersClass = SplunkHECConstants.class)
@BodyTypes(producerIn = @BodyType(value = {
- Carrier.TEXT, Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO,
Carrier.NULL },
+ Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
Carrier.VALUE, Carrier.MAP, Carrier.LIST, Carrier.POJO,
+ Carrier.NULL },
byOption = @BodyType.When(option =
"headersOnly", value = "true", carriers = Carrier.ANY),
description = "The event, written as JSON by
Jackson with the headers beside it, or alone with"
- + " bodyOnly. A stream fails,
bytes become base64 and a File its path: convert"
- + " them to text first. With
headersOnly the body is not sent"),
+ + " bodyOnly. Bytes, a stream
or a file are sent as text. With headersOnly the body"
+ + " is not sent"),
producerOut = @BodyType(Carrier.UNCHANGED))
public class SplunkHECEndpoint extends DefaultEndpoint implements
EndpointServiceLocation {
diff --git
a/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECProducer.java
b/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECProducer.java
index 390a06d3e6c2..e1ea84014123 100644
---
a/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECProducer.java
+++
b/components/camel-splunk-hec/src/main/java/org/apache/camel/component/splunkhec/SplunkHECProducer.java
@@ -17,7 +17,12 @@
package org.apache.camel.component.splunkhec;
import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.InputStream;
+import java.io.Reader;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
import java.util.HashMap;
import java.util.Map;
@@ -25,8 +30,10 @@ import javax.net.ssl.SSLContext;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.WrappedFile;
import org.apache.camel.support.DefaultProducer;
import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
@@ -38,7 +45,7 @@ import org.apache.hc.client5.http.ssl.NoopHostnameVerifier;
import org.apache.hc.client5.http.ssl.SSLConnectionSocketFactory;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.config.RegistryBuilder;
-import org.apache.hc.core5.http.io.entity.EntityTemplate;
+import org.apache.hc.core5.http.io.entity.ByteArrayEntity;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import org.apache.hc.core5.http.message.StatusLine;
import org.apache.hc.core5.ssl.SSLContextBuilder;
@@ -102,10 +109,9 @@ public class SplunkHECProducer extends DefaultProducer {
+ endpoint.getSplunkURL() +
endpoint.getConfiguration().getSplunkEndpoint());
httppost.addHeader("Authorization", " Splunk " +
endpoint.getConfiguration().getToken());
- EntityTemplate entityTemplate = new EntityTemplate(
- -1, ContentType.APPLICATION_JSON, null, outputStream ->
MAPPER.writer().writeValue(outputStream, payload));
-
- httppost.setEntity(entityTemplate);
+ // write the JSON before sending, so a payload that cannot be written
fails before any request is made
+ byte[] json = MAPPER.writeValueAsBytes(payload);
+ httppost.setEntity(new ByteArrayEntity(json,
ContentType.APPLICATION_JSON));
httpClient.execute(
httppost,
response -> {
@@ -130,18 +136,17 @@ public class SplunkHECProducer extends DefaultProducer {
}
}
- Map<String, Object> createPayload(Message message) {
- Object body = message.getBody();
+ Map<String, Object> createPayload(Message message) throws
InvalidPayloadException {
Map<String, Object> payload = new HashMap<>();
buildPayload(message, payload);
if (endpoint.getConfiguration().isBodyOnly()) {
- payload.put("event", body);
+ payload.put("event", eventBody(message));
} else if (endpoint.getConfiguration().isHeadersOnly()) {
payload.put("event", message.getHeaders());
} else {
Map<String, Object> eventPayload = new HashMap<>();
- eventPayload.put("body", body);
+ eventPayload.put("body", eventBody(message));
eventPayload.put("headers", message.getHeaders());
payload.put("event", eventPayload);
}
@@ -149,6 +154,19 @@ public class SplunkHECProducer extends DefaultProducer {
return payload;
}
+ /**
+ * The body to write as JSON in the event. Jackson writes a byte array as
base64 and a File as its path, and fails
+ * half way on a stream, so these bodies (and the file of the file
consumer) are converted to text.
+ */
+ private static Object eventBody(Message message) throws
InvalidPayloadException {
+ Object body = message.getBody();
+ if (body instanceof byte[] || body instanceof InputStream || body
instanceof Reader || body instanceof File
+ || body instanceof Path || body instanceof WrappedFile || body
instanceof ByteBuffer) {
+ return message.getMandatoryBody(String.class);
+ }
+ return body;
+ }
+
private void buildPayload(Message message, Map<String, Object> payload) {
if (endpoint.getConfiguration().getSourceType() != null) {
payload.put("sourcetype",
endpoint.getConfiguration().getSourceType());
diff --git
a/components/camel-splunk-hec/src/test/java/org/apache/camel/component/splunkhec/SplunkHECProducerHttpTest.java
b/components/camel-splunk-hec/src/test/java/org/apache/camel/component/splunkhec/SplunkHECProducerHttpTest.java
index d8aba5146a8b..bbef42f40da0 100644
---
a/components/camel-splunk-hec/src/test/java/org/apache/camel/component/splunkhec/SplunkHECProducerHttpTest.java
+++
b/components/camel-splunk-hec/src/test/java/org/apache/camel/component/splunkhec/SplunkHECProducerHttpTest.java
@@ -16,22 +16,33 @@
*/
package org.apache.camel.component.splunkhec;
+import java.io.ByteArrayInputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.net.httpserver.HttpServer;
import org.apache.camel.Exchange;
import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.file.GenericFile;
import org.apache.camel.support.jsse.SSLContextParameters;
import org.apache.camel.test.junit6.CamelTestSupport;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -42,14 +53,23 @@ public class SplunkHECProducerHttpTest extends
CamelTestSupport {
private static final String TOKEN = "11111111-1111-1111-1111-111111111111";
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
private final List<String> requests = new CopyOnWriteArrayList<>();
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private final AtomicInteger arrived = new AtomicInteger();
private HttpServer server;
+ @TempDir
+ Path tempDir;
+
@BeforeEach
void startServer() throws Exception {
server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
server.createContext("/", httpExchange -> {
+ arrived.incrementAndGet();
String body = new
String(httpExchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
+ bodies.add(body);
requests.add(httpExchange.getRequestMethod() + " " +
httpExchange.getRequestURI().getPath() + " "
+
httpExchange.getRequestHeaders().getFirst("Authorization") + " " + body);
byte[] answer =
"{\"text\":\"Success\",\"code\":0}".getBytes(StandardCharsets.UTF_8);
@@ -89,6 +109,85 @@ public class SplunkHECProducerHttpTest extends
CamelTestSupport {
assertTrue(requests.get(0).contains("\"event\":\"hello\""),
requests.get(0));
}
+ @Test
+ void testStreamBody() throws Exception {
+ template.sendBody(uri("&bodyOnly=true"), new
ByteArrayInputStream("hello stream".getBytes(StandardCharsets.UTF_8)));
+
+ assertEquals("hello stream", event().asText());
+ }
+
+ @Test
+ void testStreamBodyWithHeaders() throws Exception {
+ template.sendBodyAndHeader(uri(""), new ByteArrayInputStream("hello
stream".getBytes(StandardCharsets.UTF_8)),
+ "foo", "bar");
+
+ JsonNode event = event();
+ assertEquals("hello stream", event.get("body").asText());
+ assertEquals("bar", event.get("headers").get("foo").asText());
+ }
+
+ @Test
+ void testBytesBody() throws Exception {
+ template.sendBody(uri("&bodyOnly=true"), "hello
bytes".getBytes(StandardCharsets.UTF_8));
+
+ assertEquals("hello bytes", event().asText());
+ }
+
+ @Test
+ void testFileBody() throws Exception {
+ Path file = Files.writeString(tempDir.resolve("event.txt"), "hello
file");
+
+ template.sendBody(uri("&bodyOnly=true"), file.toFile());
+ template.sendBody(uri("&bodyOnly=true"), file);
+
+ assertEquals(2, bodies.size());
+ assertEquals("hello file",
MAPPER.readTree(bodies.get(0)).get("event").asText());
+ assertEquals("hello file",
MAPPER.readTree(bodies.get(1)).get("event").asText());
+ }
+
+ @Test
+ void testFileConsumerBody() throws Exception {
+ Files.writeString(tempDir.resolve("event.txt"), "hello file consumer");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:file")
+ .pollEnrich("file:" + tempDir.toAbsolutePath() +
"?noop=true", 5000)
+ .to(uri("&bodyOnly=true"));
+ }
+ });
+
+ Exchange exchange = template.send("direct:file", e -> {
+ });
+
+ assertNull(exchange.getException());
+ assertInstanceOf(GenericFile.class, exchange.getMessage().getBody());
+ assertEquals("hello file consumer", event().asText());
+ }
+
+ @Test
+ void testBodyNotWrittenAsJsonFailsBeforeSending() throws Exception {
+ // Jackson cannot write a plain Object: it fails after it has written
the long text before it,
+ // and this must fail before any request reaches Splunk
+ List<Object> body = List.of("x".repeat(20000), new Object());
+ Exchange exchange = template.send(uri("&bodyOnly=true"), e ->
e.getIn().setBody(body));
+ assertNotNull(exchange.getException());
+
+ // the server handles the requests one by one, so once this one is
answered the failed one would have arrived
+ template.sendBody(uri("&bodyOnly=true"), "hello");
+ assertEquals(1, arrived.get());
+ assertEquals("hello", event().asText());
+ }
+
+ private String uri(String options) {
+ return "splunk-hec:localhost:" + server.getAddress().getPort() +
"?token=" + TOKEN + "&https=false" + options;
+ }
+
+ private JsonNode event() throws Exception {
+ assertEquals(1, bodies.size(), bodies::toString);
+ return MAPPER.readTree(bodies.get(0)).get("event");
+ }
+
@Test
void testStartWithoutSslContextParameters() throws Exception {
// https=true (the default) without sslContextParameters uses the
default SSL context of the JVM
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 f31bdcef0c23..6e5da218ad82 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
@@ -804,6 +804,15 @@ CamelContext of that provider. It now closes each of them
only when no other `JC
it returned the closed cache, whose operations throw `IllegalStateException`.
This matters for code that calls a
`JCacheManager` (for example one created by a custom `JCacheManagerFactory`)
after it was closed.
+=== camel-splunk-hec - bytes, stream and file bodies are sent as text
+
+The producer now sends a byte array, an `InputStream`, a `Reader` or a file
body (`File`, `Path` or the file of the
+file consumer) as text in the event. Before, a byte array was sent as a base64
string, a `File` as its path, and a
+stream failed after part of the request had already been sent.
+
+The JSON of the event is now written before the request is sent, with a
`Content-Length` instead of a chunked request
+body. A body or a header that cannot be written as JSON fails the exchange
without sending anything to Splunk.
+
=== camel-spring-batch - the CamelSpringBatchJobName header needs
jobFromHeader=true
The producer now takes the job name from the `CamelSpringBatchJobName` header
only when the endpoint has
