This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new f3e4f9b404 Fix #1543: deliver stream bodies of source records as
byte[] (#1852)
f3e4f9b404 is described below
commit f3e4f9b404aeaee8762770241971cbde1814b4a2
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 5 11:35:27 2026 +0100
Fix #1543: deliver stream bodies of source records as byte[] (#1852)
With Camel stream caching, which Camel Main enables by default, the
output of a marshal step is a StreamCache. Many source Kamelets marshal
their output (for example marshal: json), and CamelSourceTask put the
InputStreamCache object itself into the SourceRecord: StringConverter
wrote its object reference and ByteArrayConverter rejected it.
CamelSourceTask now reads StreamCache and InputStream bodies into a
byte[] with the Camel type converter before building the record, so the
value gets a BYTES schema. This is what the marshal step produces when
stream caching is disabled.
- Add tests for a marshalled body (syslog data format), an
InputStreamCache body and an InputStream body.
- Document source record values and the converters to use in
basic-configuration.adoc.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/kafkaconnector/CamelSourceTask.java | 36 ++++++++++---
.../camel/kafkaconnector/CamelSourceTaskTest.java | 59 ++++++++++++++++++++++
.../ROOT/pages/user-guide/basic-configuration.adoc | 21 ++++++++
3 files changed, 109 insertions(+), 7 deletions(-)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
index 876f91fe52..30597b3741 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.kafkaconnector;
+import java.io.InputStream;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.ArrayList;
@@ -28,8 +29,10 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
+import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.PollingConsumer;
import org.apache.camel.StreamCache;
+import org.apache.camel.TypeConversionException;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.kafkaconnector.utils.CamelKafkaConnectMain;
import org.apache.camel.kafkaconnector.utils.SchemaHelper;
@@ -228,19 +231,13 @@ public class CamelSourceTask extends SourceTask {
Map<String, String> sourceOffset =
Collections.singletonMap("position", exchange.getExchangeId());
final Object messageHeaderKey = camelMessageHeaderKey != null ?
exchange.getMessage().getHeader(camelMessageHeaderKey) : null;
- Object messageBodyValue = exchange.getMessage().getBody();
+ final Object messageBodyValue = toRecordValue(exchange);
final Schema messageKeySchema = messageHeaderKey != null ?
SchemaHelper.buildSchemaBuilderForType(messageHeaderKey) : null;
final Schema messageBodySchema = messageBodyValue != null ?
SchemaHelper.buildSchemaBuilderForType(messageBodyValue) : null;
final long timestamp = calculateTimestamp(exchange);
- // take in account Cached camel streams
- if (messageBodyValue instanceof StreamCache) {
- StreamCache sc = (StreamCache) messageBodyValue;
- // reset to be sure that the cache is ready to be used before
sending it in the record (could be useful for SMTs)
- sc.reset();
- }
// every record below is produced from this one exchange; it may
only be acknowledged towards the external
// system after the last of them has been committed
AtomicInteger pendingCommits = new AtomicInteger(topics.length);
@@ -356,6 +353,31 @@ public class CamelSourceTask extends SourceTask {
return System.currentTimeMillis();
}
+ /**
+ * Returns the record value for the body of the given exchange.
+ * <p>
+ * Kafka Connect converters cannot handle a stream: it would end up in the
record as an object reference. A
+ * stream body is common, since with stream caching (enabled by default)
the output of a {@code marshal} step,
+ * such as the one many Kamelets end with, is a {@link StreamCache}. Such
a body, like any other
+ * {@link InputStream}, is read into a {@code byte[]}, which is also what
the {@code marshal} step produces when
+ * stream caching is disabled.
+ */
+ private static Object toRecordValue(Exchange exchange) {
+ final Object body = exchange.getMessage().getBody();
+ if (body instanceof StreamCache) {
+ // the cache may have been read already, so make sure it is
converted from the beginning
+ ((StreamCache) body).reset();
+ } else if (!(body instanceof InputStream)) {
+ return body;
+ }
+
+ try {
+ return
exchange.getContext().getTypeConverter().mandatoryConvertTo(byte[].class,
exchange, body);
+ } catch (NoTypeConversionAvailableException | TypeConversionException
e) {
+ throw new ConnectException("Unable to read the stream body of
exchange " + exchange.getExchangeId(), e);
+ }
+ }
+
private void setAdditionalHeaders(SourceRecord record, Map<String, Object>
map, String prefix) {
for (Map.Entry<String, Object> entry : map.entrySet()) {
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
index c8cace47ba..8defdbb799 100644
---
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.kafkaconnector;
+import java.io.ByteArrayInputStream;
import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
@@ -31,6 +33,7 @@ import java.util.stream.IntStream;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.converter.stream.InputStreamCache;
import org.apache.camel.kafkaconnector.utils.StringJoinerAggregator;
import org.apache.camel.support.SynchronizationAdapter;
import org.apache.kafka.connect.data.Decimal;
@@ -41,6 +44,7 @@ import org.junit.jupiter.api.Test;
import static org.apache.camel.util.CollectionHelper.mapOf;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -228,6 +232,61 @@ public class CamelSourceTaskTest {
sourceTask.stop();
}
+ @Test
+ public void testSourcePollingWithMarshalledBody() {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+ // like the marshal step at the end of many Kamelets: with stream
caching (enabled by default) the
+ // data format output is a StreamCache
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_MARSHAL_CONF,
"syslog");
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ sourceTask.start(props);
+ try {
+ sourceTask.getCms().getProducerTemplate().sendBody(DIRECT_URI,
+ "<34>Oct 11 22:14:15 mymachine su: 'su root' failed for
lonvick on /dev/pts/8");
+
+ List<SourceRecord> poll = sourceTask.poll();
+ assertEquals(1, poll.size());
+ SourceRecord record = poll.get(0);
+ assertThat(record.value()).isInstanceOf(byte[].class);
+ assertEquals(Schema.Type.BYTES, record.valueSchema().type());
+ assertThat(new String((byte[]) record.value(),
StandardCharsets.UTF_8))
+ .contains("mymachine")
+ .contains("'su root' failed for lonvick on /dev/pts/8");
+ } finally {
+ sourceTask.stop();
+ }
+ }
+
+ @Test
+ public void testSourcePollingWithStreamBody() {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ sourceTask.start(props);
+ try {
+ final ProducerTemplate template =
sourceTask.getCms().getProducerTemplate();
+ final byte[] payload = "{\"id\": 1, \"name\":
\"camel\"}".getBytes(StandardCharsets.UTF_8);
+
+ template.sendBody(DIRECT_URI, new InputStreamCache(payload));
+ template.sendBody(DIRECT_URI, new ByteArrayInputStream(payload));
+
+ List<SourceRecord> poll = sourceTask.poll();
+ assertEquals(2, poll.size());
+ for (SourceRecord record : poll) {
+ assertThat(record.value()).isInstanceOf(byte[].class);
+ assertArrayEquals(payload, (byte[]) record.value());
+ assertEquals(Schema.Type.BYTES, record.valueSchema().type());
+ }
+ } finally {
+ sourceTask.stop();
+ }
+ }
+
@Test
public void testUrlPrecedenceOnComponentProperty() throws
InterruptedException {
Map<String, String> props = new HashMap<>();
diff --git a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
index 037f6279eb..6dbe7188a0 100644
--- a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
+++ b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
@@ -63,6 +63,27 @@ For a Source connector the basic options are:
For more options related to single connector you can have a look at
xref:reference/index.adoc[Connectors list].
+[[BasicConfiguration-SourceRecordValues]]
+== Source Record Values
+
+A source connector uses the body of each Camel message as the value of the
Kafka record. Text, numbers and booleans keep their type. Binary content and
stream bodies become a `byte[]` value with a `BYTES` schema. Stream bodies are
common: with stream caching, which Camel enables by default, the output of a
`marshal` step (many Kamelets end with one, for example to produce JSON) is a
stream.
+
+To write these bytes to the topic unchanged, configure the connector with:
+
+[source,properties]
+----
+value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
+----
+
+The `StringConverter` used in the generated example configurations writes a
`byte[]` value as a Java object reference (`[B@...`). If you need the value as
a string, for example for JSON produced by a Kamelet, convert it first with the
`CamelTypeConverterTransform` SMT:
+
+[source,properties]
+----
+transforms=toString
+transforms.toString.type=org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value
+transforms.toString.target.type=java.lang.String
+----
+
[[BasicConfiguration-MessageLevelHeaders]]
== Message Level Headers
