Hi all,
I am trying to work with flink to get avro data from kafka for which the
schemas are stored in kafka schema registry. Since, the producer for kafka is a
totally different service(an MQTT consumer sinked to kafka), I can’t have the
schema with me at the consumer end. I read around and diverged to the following
implementation of KeyedDeserializationSchema but I cannot understand why it’s
throwing a `com.esotericsoftware.kryo.KryoException:
java.lang.NullPointerException`
class AvroDeserializationSchema(schemaRegistryUrl: String) extends
KeyedDeserializationSchema[GenericRecord] {
// Flink needs the serializer to be serializable => this "@transient lazy
val" does the trick
@transient lazy val valueDeserializer = {
val deserializer = new KafkaAvroDeserializer(new
CachedSchemaRegistryClient(schemaRegistryUrl,
AbstractKafkaAvroSerDeConfig.MAX_SCHEMAS_PER_SUBJECT_DEFAULT))
deserializer.configure(
Map(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG ->
schemaRegistryUrl,
KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG ->
false).asJava,
false)
deserializer
}
override def isEndOfStream(nextElement: GenericRecord): Boolean = false
override def deserialize(messageKey: Array[Byte], message: Array[Byte],
topic: String, partition: Int, offset: Long): GenericRecord = {
// val key = keyDeserializer.deserialize(topic,
messageKey).asInstanceOf[String]
val value = valueDeserializer.deserialize(topic,
message).asInstanceOf[GenericRecord]
value
}
override def getProducedType: TypeInformation[GenericRecord] =
TypeExtractor.getForClass(classOf[GenericRecord])
}
I have no clue how to go about solving this. I saw a lot of people trying to
implement the same. If someone can guide me, it’d be really helpful.
Thanks!
Nitish
