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

["Kartik (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/CAMEL-20864?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17854285#comment-17854285 ] 

Kartik commented on CAMEL-20864:
--------------------------------

[~davsclaus]  I see the same issue in 4.6.0 on checking there is a code that checks whether the input payload is of type Iterable.

!image-2024-06-12-12-51-58-966.png!

 

> camel-kafka - With confluent schema registry does not work properly.
> --------------------------------------------------------------------
>
>                 Key: CAMEL-20864
>                 URL: https://issues.apache.org/jira/browse/CAMEL-20864
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-kafka
>    Affects Versions: 3.22.2
>         Environment: Confluent Kafka
> Camel-3.22.x
> Java 17
>            Reporter: Kartik
>            Priority: Minor
>              Labels: camel-kafka
>         Attachments: image-2024-06-12-11-05-12-616.png, image-2024-06-12-11-06-33-915.png, image-2024-06-12-12-51-58-966.png
>
>
> 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.
> {code:java}
> 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); {code}
> 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.
> !image-2024-06-12-11-05-12-616.png|width=733,height=459!
>  
>  The code in "
> io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer expects whole object.
> !image-2024-06-12-11-06-33-915.png|width=906,height=438!
>  
> 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.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)