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 efe92f7b1f Fix #269: pass null keys and values through
CamelTypeConverterTransform (#1855)
efe92f7b1f is described below
commit efe92f7b1f77e301e845d94988efed30ebf275b1
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 5 11:36:18 2026 +0100
Fix #269: pass null keys and values through CamelTypeConverterTransform
(#1855)
CamelTypeConverterTransform failed on a tombstone (a record with a null
value) with a DataException, because the Camel type converter returns
null for a null input. It also failed with a NullPointerException on any
schemaless record, because it always copied the original schema.
- Return the record unchanged when the key or value to convert is null,
as Kafka's Cast SMT does.
- Keep schemaless records schemaless instead of copying a null schema.
- Add a reference page for the SMT and list it in the nav and in the
transformers index.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../transforms/CamelTypeConverterTransform.java | 9 +++-
.../CamelTypeConverterTransformTest.java | 51 ++++++++++++++++++++++
docs/modules/ROOT/nav.adoc | 1 +
.../reference/transformers/camelTypeConverter.adoc | 44 +++++++++++++++++++
.../ROOT/pages/reference/transformers/index.adoc | 3 ++
5 files changed, 106 insertions(+), 2 deletions(-)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransform.java
b/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransform.java
index 2078ea21fa..6241835d70 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransform.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransform.java
@@ -41,11 +41,16 @@ public abstract class CamelTypeConverterTransform<R extends
ConnectRecord<R>> ex
@Override
public R apply(R record) {
- final Schema schema = operatingSchema(record);
final Object value = operatingValue(record);
+ if (value == null) {
+ // nothing to convert, for example a tombstone: pass it through
unchanged, like Kafka's Cast SMT does
+ return record;
+ }
+ final Schema schema = operatingSchema(record);
final Object convertedValue =
convertValueWithCamelTypeConverter(value);
- final Schema updatedSchema = getOrBuildRecordSchema(schema,
convertedValue);
+ // a schemaless record stays schemaless
+ final Schema updatedSchema = schema == null ? null :
getOrBuildRecordSchema(schema, convertedValue);
return newRecord(record, updatedSchema, convertedValue);
}
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransformTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransformTest.java
index 4cc93ff110..4e65262acd 100644
---
a/core/src/test/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransformTest.java
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/transforms/CamelTypeConverterTransformTest.java
@@ -41,6 +41,7 @@ import static
org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -72,6 +73,56 @@ public class CamelTypeConverterTransformTest {
assertEquals(Schema.BOOLEAN_SCHEMA,
transformedValueSourceRecord.valueSchema());
}
+ @Test
+ public void testIfItPassesNullValuesThrough() {
+ final Map<String, Object> props = new HashMap<>();
+ props.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG,
"java.lang.String");
+
+ final Transformation<SourceRecord> transformationKey = new
CamelTypeConverterTransform.Key<>();
+ final Transformation<SourceRecord> transformationValue = new
CamelTypeConverterTransform.Value<>();
+ transformationKey.configure(props);
+ transformationValue.configure(props);
+
+ // a tombstone, with and without a schema
+ final SourceRecord schemalessTombstone = new
SourceRecord(Collections.emptyMap(), Collections.emptyMap(), "topic",
Schema.STRING_SCHEMA, "1234", null, null);
+ final SourceRecord tombstoneWithSchema = new
SourceRecord(Collections.emptyMap(), Collections.emptyMap(), "topic",
Schema.STRING_SCHEMA, "1234", Schema.OPTIONAL_BYTES_SCHEMA, null);
+ assertSame(schemalessTombstone,
transformationValue.apply(schemalessTombstone));
+ assertSame(tombstoneWithSchema,
transformationValue.apply(tombstoneWithSchema));
+
+ // a record without a key
+ final SourceRecord noKey = new SourceRecord(Collections.emptyMap(),
Collections.emptyMap(), "topic", null, null, Schema.STRING_SCHEMA, "value");
+ assertSame(noKey, transformationKey.apply(noKey));
+
+ transformationKey.close();
+ transformationValue.close();
+ }
+
+ @Test
+ public void testIfItConvertsSchemalessRecords() {
+ final SourceRecord connectRecord = new
SourceRecord(Collections.emptyMap(), Collections.emptyMap(), "topic", null,
"1234", null, "TRUE");
+
+ final Map<String, Object> propsForKeySmt = new HashMap<>();
+
propsForKeySmt.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG,
Integer.class.getName());
+ final Map<String, Object> propsForValueSmt = new HashMap<>();
+
propsForValueSmt.put(CamelTypeConverterTransform.FIELD_TARGET_TYPE_CONFIG,
"java.lang.Boolean");
+
+ final Transformation<SourceRecord> transformationKey = new
CamelTypeConverterTransform.Key<>();
+ final Transformation<SourceRecord> transformationValue = new
CamelTypeConverterTransform.Value<>();
+ transformationKey.configure(propsForKeySmt);
+ transformationValue.configure(propsForValueSmt);
+
+ final SourceRecord transformedKeySourceRecord =
transformationKey.apply(connectRecord);
+ final SourceRecord transformedValueSourceRecord =
transformationValue.apply(connectRecord);
+
+ assertEquals(1234, transformedKeySourceRecord.key());
+ assertNull(transformedKeySourceRecord.keySchema());
+ assertEquals(true, transformedValueSourceRecord.value());
+ assertNull(transformedValueSourceRecord.valueSchema());
+
+ transformationKey.close();
+ transformationValue.close();
+ }
+
@Test
public void testIfItConvertsNettyCorrectly() {
final String testMessage = "testMessage";
diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc
index 0f82d198d3..659f8d37ba 100644
--- a/docs/modules/ROOT/nav.adoc
+++ b/docs/modules/ROOT/nav.adoc
@@ -21,6 +21,7 @@
** xref:contributor-guide/release-guide.adoc[Release Guide]
* xref:reference/index.adoc[Reference]
** xref:reference/transformers/index.adoc[Single Message Transformers (SMT)]
+*** xref:reference/transformers/camelTypeConverter.adoc[Camel Type Converter]
*** xref:reference/transformers/fieldsToHeaders.adoc[Fields To Headers]
** xref:reference/index.adoc[Connectors]
// connectors: START
diff --git
a/docs/modules/ROOT/pages/reference/transformers/camelTypeConverter.adoc
b/docs/modules/ROOT/pages/reference/transformers/camelTypeConverter.adoc
new file mode 100644
index 0000000000..a7fe09da59
--- /dev/null
+++ b/docs/modules/ROOT/pages/reference/transformers/camelTypeConverter.adoc
@@ -0,0 +1,44 @@
+[[CamelTypeConverter-CamelTypeConverter]]
+= CamelTypeConverter
+The following provides usage information for the SMT
`org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform`
+
+== Description
+
+CamelTypeConverter converts the key or the value of a record to another Java
type using the
+https://camel.apache.org/manual/type-converter.html[Camel type converters],
for example a `byte[]`
+or a Camel stream into a `String`, or a Kafka Connect `Struct` into a `Map`.
+
+The schema of the converted key or value is derived from the converted value.
Records without a
+schema (for example records read with `value.converter.schemas.enable=false`)
stay without a schema.
+
+A null key or value, for example the value of a tombstone record, is passed
through unchanged.
+
+Use the concrete transformation type designed for the record key
+(`org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Key`)
or value
+(`org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value`).
+
+If a key or value cannot be converted to the target type, the transformation
fails with a
+`DataException`.
+
+== Examples
+
+=== Given the following configuration:
+
+[source,java-properties]
+----
+transforms=toString
+transforms.toString.type=org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value
+transforms.toString.target.type=java.lang.String
+----
+and a record whose value is the `byte[]` of the JSON document `{"id": 1}`, the
value of the
+transformed record is the string `{"id": 1}` with a `STRING` schema.
+
+== Properties
+
+[cols="^,^ ", options="header"]
+|===
+|Name | Description
+
+|target.type
+|The fully qualified name of the Java class to convert the key or value to,
for example `java.lang.String` or `java.util.Map`.
+|===
diff --git a/docs/modules/ROOT/pages/reference/transformers/index.adoc
b/docs/modules/ROOT/pages/reference/transformers/index.adoc
index 0f96b24514..36dc07d72b 100644
--- a/docs/modules/ROOT/pages/reference/transformers/index.adoc
+++ b/docs/modules/ROOT/pages/reference/transformers/index.adoc
@@ -17,6 +17,9 @@ Camel Kafka Connector provides additional SMTs:
|===
|Transform | Description
+|xref:reference/transformers/camelTypeConverter.adoc[CamelTypeConverter]
+|Convert a kafka connect record's key or value to another Java type with the
Camel type converters
+
|xref:reference/transformers/fieldsToHeaders.adoc[FieldsToHeaders]
|Extract fields from a kafka connect record's value or key and copy the value
in configured headers (struct + schema, map or copy the whole value/key are the
supported options)
