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 27fd4a2e06 Fix #1861: give Map and List record values a schema
converters can serialize (#1863)
27fd4a2e06 is described below
commit 27fd4a2e06eb9b12a6fa8c85fd36afa24e0c0125
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 5 12:02:03 2026 +0100
Fix #1861: give Map and List record values a schema converters can
serialize (#1863)
SchemaHelper gives a Map or List value a MAP or ARRAY schema without key
or value schema: its fallback builds one, and copySchemaBasics drops
them from an inferred schema. The JsonConverter then fails with a
NullPointerException on any source record whose body is a Map or List,
for example form fields from a webhook or an unmarshalled document.
CamelSourceTask now gives such a key or value the schema Kafka Connect
infers for it, which is complete, or no schema when the types cannot be
inferred, which converters handle as schemaless data.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/kafkaconnector/CamelSourceTask.java | 21 +++++++++-
.../camel/kafkaconnector/CamelSourceTaskTest.java | 46 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 2 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 30597b3741..27dcbc8f30 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
@@ -40,6 +40,7 @@ import org.apache.camel.kafkaconnector.utils.TaskHelper;
import org.apache.camel.support.UnitOfWorkHelper;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Values;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
@@ -233,8 +234,8 @@ public class CamelSourceTask extends SourceTask {
final Object messageHeaderKey = camelMessageHeaderKey != null ?
exchange.getMessage().getHeader(camelMessageHeaderKey) : null;
final Object messageBodyValue = toRecordValue(exchange);
- final Schema messageKeySchema = messageHeaderKey != null ?
SchemaHelper.buildSchemaBuilderForType(messageHeaderKey) : null;
- final Schema messageBodySchema = messageBodyValue != null ?
SchemaHelper.buildSchemaBuilderForType(messageBodyValue) : null;
+ final Schema messageKeySchema = recordSchema(messageHeaderKey);
+ final Schema messageBodySchema = recordSchema(messageBodyValue);
final long timestamp = calculateTimestamp(exchange);
@@ -418,6 +419,22 @@ public class CamelSourceTask extends SourceTask {
}
}
+ /**
+ * Returns the schema of a record key or value. {@link SchemaHelper} gives
a {@link Map} or a {@link List} a MAP or
+ * ARRAY schema without key or value schema, which converters such as the
JsonConverter cannot serialize. Such a
+ * value gets the schema Kafka Connect infers for it, which is complete,
or no schema when its key and value types
+ * cannot be inferred, for example for a JSON object with values of
different types.
+ */
+ private static Schema recordSchema(Object value) {
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof Map || value instanceof List) {
+ return Values.inferSchema(value);
+ }
+ return SchemaHelper.buildSchemaBuilderForType(value);
+ }
+
private String getLocalUrlWithPollingOptions(long
pollingConsumerQueueSize, long pollingConsumerBlockTimeout, boolean
pollingConsumerBlockWhenFull) {
return LOCAL_URL + "?pollingConsumerQueueSize=" +
pollingConsumerQueueSize + "&pollingConsumerBlockTimeout=" +
pollingConsumerBlockTimeout
+ "&pollingConsumerBlockWhenFull=" +
pollingConsumerBlockWhenFull;
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 8defdbb799..e7a771d990 100644
---
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
@@ -20,8 +20,11 @@ import java.io.ByteArrayInputStream;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
@@ -39,6 +42,8 @@ import org.apache.camel.support.SynchronizationAdapter;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.header.Header;
+import org.apache.kafka.connect.json.JsonConverter;
+import org.apache.kafka.connect.json.JsonConverterConfig;
import org.apache.kafka.connect.source.SourceRecord;
import org.junit.jupiter.api.Test;
@@ -142,6 +147,47 @@ public class CamelSourceTaskTest {
sourceTask.stop();
}
+ @Test
+ public void testSourcePollingWithMapAndListBody() {
+ 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 (JsonConverter converter = new JsonConverter()) {
+
converter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG,
false), false);
+ final ProducerTemplate template =
sourceTask.getCms().getProducerTemplate();
+
+ // a JSON object with values of different types, such as form
fields or an unmarshalled document
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("token", "abc");
+ map.put("count", 3);
+ template.sendBody(DIRECT_URI, map);
+ template.sendBody(DIRECT_URI, Arrays.asList("camel", 1, true));
+ // values whose types can be inferred keep a complete schema
+ template.sendBody(DIRECT_URI, Collections.singletonMap("name",
"camel"));
+
+ List<SourceRecord> poll = sourceTask.poll();
+ assertEquals(3, poll.size());
+
+ assertNull(poll.get(0).valueSchema());
+ assertEquals("{\"token\":\"abc\",\"count\":3}", toJson(converter,
poll.get(0)));
+ assertNull(poll.get(1).valueSchema());
+ assertEquals("[\"camel\",1,true]", toJson(converter, poll.get(1)));
+ assertEquals(Schema.Type.MAP, poll.get(2).valueSchema().type());
+ assertEquals(Schema.Type.STRING,
poll.get(2).valueSchema().keySchema().type());
+ assertEquals(Schema.Type.STRING,
poll.get(2).valueSchema().valueSchema().type());
+ assertEquals("{\"name\":\"camel\"}", toJson(converter,
poll.get(2)));
+ } finally {
+ sourceTask.stop();
+ }
+ }
+
+ private static String toJson(JsonConverter converter, SourceRecord record)
{
+ return new String(converter.fromConnectData(record.topic(),
record.valueSchema(), record.value()), StandardCharsets.UTF_8);
+ }
+
@Test
public void testSourcePollingWithKey() {
Map<String, String> props = new HashMap<>();
