[CAMEL-20864] camel-kafka - With confluent schema registry does not work properly.

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 3.22.2
• Fix Version/s: 3.21.5, 3.22.3, 4.0.6, 4.4.3, 4.7.0
• Component/s: camel-kafka
• Labels: camel-kafka

Environment

Confluent Kafka

Camel-3.22.x

Java 17

• Estimated Complexity: Unknown
• Assignee: Claus Ibsen
• Reporter: Kartik
• Created: 12/Jun/24 05:38
• Updated: 17/Jun/24 07:39
• Resolved: 12/Jun/24 17:43

Description

In confluent kafka, we can register the topic against schema validation from the schema registry. When configured the confluent document says either we should have a pojo object defined in the code that is used for serialization/deserialization OR a custom "ObjectNode" can be created from their schema utils. Attaching the document below

https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/serdes-json.html#sending-a-jsonnode-payload 

 

For our case, we have a different schema registered and can't have all the POJO defined as schema registered at run time, so we are using the below code to generate an object from the schema.

Map<String, Object> config = new HashMap<>();
config.put("basic.auth.credentials.source", "USER_INFO");
config.put("basic.auth.user.info", "<secret-key>");
config.put("auto.register.schemas", false);
config.put("use.latest.version", true);

CachedSchemaRegistryClient registryClient = new CachedSchemaRegistryClient("<registry-url>", 10, config);
String schemaDoc = registryClient.getLatestSchemaMetadata("topicTest-value").getSchema();
JsonSchema schema = new JsonSchema(schemaDoc);

ObjectMapper mapper = new ObjectMapper();
JsonNode jsonNode = mapper.readTree("{\"myField1\":123,\"myField2\":123.32,\"myField3\":\"pqr\"}");
ObjectNode envelope = JsonSchemaUtils.envelope(schema, jsonNode);

from("timer://foo?fixedRate=true&period=60000")
        .setBody(ExpressionBuilder.constantExpression(envelope))
        .log("Sent new message")// Message to send
        .to(kafkaEndpoint); 

If the "ObjectNode" payload is directly written using kafka-client library it works. But when written using camel component, The "KafkaProducer" in camel component does "isIterable" check and if true sends each value and this doesn't work for confluent kafka as the custom "io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer" expects a whole object.

 

 The code in "

io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer expects whole object.

 

Basically, in simple words, The "envelope" object created is no longer the same object but is iterated and values are iterated and sent independently resulting in schema validation error.

Issue Links

• links to: GitHub Pull Request #14492