[HUDI-8378] Fix Avro schema deserializer failing with schema evolution #12111
Conversation
github-actions
bot
added
size:M
| } | ||
| props.put(KAFKA_VALUE_DESERIALIZER_SCHEMA.key(), schemaProvider.getSourceSchema().toString()); | ||
| // assign consumer group id based on the schema, since if there's a change in the schema we ensure KafkaRDDIterator doesn't use cached Kafka Consumer | ||
| String groupId = props.getString(NATIVE_KAFKA_CONSUMER_GROUP_ID, ""); |
There was a problem hiding this comment.
Instead of setting up the group id, why not just fetch the latest schema and set up it again if the schema evolved.
There was a problem hiding this comment.
The KafkaRDD class from the Kafka client package is caching the KafkaConsumer for a given group id and partition. As a result, we are using KafkaAvroSchemaDeserializer with the previous schema to deserialize the records. This leads to issues during schema evolution.
This changes makes sure that new KafkaConsumer is created when there's a schema change present.
There was a problem hiding this comment.
Does Kafka client itself has an option to control the cache TTL?
There was a problem hiding this comment.
No, while creating the Kafka RDD, it's always hardcoded to true
https://github.com/apache/spark/blob/876450c4130062a80f4ecb7b7afd232b5481a99f/connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaUtils.scala#L70
Since useConsumerCache is set to true. We always get the existing consumer with outdated schema
https://github.com/apache/spark/blob/876450c4130062a80f4ecb7b7afd232b5481a99f/connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaDataConsumer.scala#L337
There was a problem hiding this comment.
Can we copy this utility into hudi and override the flag as false then?
There was a problem hiding this comment.
But if we override the flag to false, we might end up creating new KafkaConsumer every time.
There was a problem hiding this comment.
Can we create based on the schema change then? The delta-streamer is a micro-batch for each run, maybe we can add some smart strategy here.
There was a problem hiding this comment.
This change will do exactly the same. Caching the KafkaConsumer happens as part of the RDD creation, I'm not sure we can extract that functionality of the kafka utils libr.
There was a problem hiding this comment.
Caching the KafkaConsumer happens as part of the RDD creation
Does it make sense if we hard code the cache as false first then create/reuse the consumer based on the schema change?
There was a problem hiding this comment.
We can't control how the KafkaConsumer is created, It's done during the RDD creation. I don't think we can control that.
vamsikarnika
force-pushed
the
fix_avro_deserializer
branch
from
71c4609 to
5680f4c
Compare
Change Logs
Add consumer group id to kafka params based on the schema hash when KafkaAvroSchemaDeserializer is used to avoid kafka connector caching the KafkaConsumer causing issues with schema evolution.
Impact
Kafka connector will create a new Kafka consumer whenever there's a change in the schema instead of using cached kafka consumer.
Risk level (write none, low medium or high below)
NA
Documentation Update
Describe any necessary documentation update if there is any new feature, config, or user-facing change. If not, put "none".
ticket number here and follow the instruction to make
changes to the website.
Contributor's checklist