4

I am using Kafka Connector in Apache Flink for access to streams served by Confluent Kafka.

Apart from schema registry url ConfluentRegistryAvroDeserializationSchema.forGeneric(...) expecting 'reader' schema. Instead of providing read schema I want to use same writer's schema(lookup in registry) for reading the message too because Consumer will not have latest schema.

FlinkKafkaConsumer010<GenericRecord> myConsumer =
        new FlinkKafkaConsumer010<>("topic-name", ConfluentRegistryAvroDeserializationSchema.forGeneric(<reader schema goes here>, "http://host:port"), properties);
myConsumer.setStartFromLatest();

https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html "Using these deserialization schema record will be read with the schema that was retrieved from Schema Registry and transformed to a statically provided"

Since I do not want to keep schema definition at consumer side how do I deserialize Avro message from Kafka using writer's schema?

Appreciate your help!

3
  • The default behavior is to parse incoming messages using the writer schemas and then converting to the reader schema. Isn't it what you want?
    – 0x26res
    Nov 14, 2019 at 9:51
  • I do not want it to convert, just use writer schema for deserializing the message.
    – Sparkle8
    Nov 14, 2019 at 15:37
  • I think you want to wrap/adapt a KafkaAvroDeserializer into a KeyedDeserializationSchema and make sure you set specific.avro.reader to false, and this should give you the raw GenericRecord as they were emitted by the producer.
    – 0x26res
    Nov 14, 2019 at 17:02

1 Answer 1

4

I don't think it is possible to use directly ConfluentRegistryAvroDeserializationSchema.forGeneric. It is intended to be used with a reader schema and they have preconditions checking for this.

You have to implement your own. Two import things:

  • Set specific.avro.reader to false (other wise you'll get specific records)
  • The KafkaAvroDeserializer has to be lazily initialized (because it isn't serializable it self, as it holds a reference to the schema registry client)
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
import java.util.HashMap;
import java.util.Map;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;

public class KafkaGenericAvroDeserializationSchema
    implements KeyedDeserializationSchema<GenericRecord> {

  private final String registryUrl;
  private transient KafkaAvroDeserializer inner;

  public KafkaGenericAvroDeserializationSchema(String registryUrl) {
    this.registryUrl = registryUrl;
  }

  @Override
  public GenericRecord deserialize(
      byte[] messageKey, byte[] message, String topic, int partition, long offset) {
    checkInitialized();
    return (GenericRecord) inner.deserialize(topic, message);
  }

  @Override
  public boolean isEndOfStream(GenericRecord nextElement) {
    return false;
  }

  @Override
  public TypeInformation<GenericRecord> getProducedType() {
    return TypeExtractor.getForClass(GenericRecord.class);
  }

  private void checkInitialized() {
    if (inner == null) {
      Map<String, Object> props = new HashMap<>();
      props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, registryUrl);
      props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, false);
      SchemaRegistryClient client =
          new CachedSchemaRegistryClient(
              registryUrl, AbstractKafkaAvroSerDeConfig.MAX_SCHEMAS_PER_SUBJECT_DEFAULT);
      inner = new KafkaAvroDeserializer(client, props);
    }
  }
}

env.addSource(
  new FlinkKafkaConsumer<>(
    topic, 
    new KafkaGenericAvroDeserializationSchema(schemaReigstryUrl), 
    kafkaProperties));
3
  • this doesn't work unless we give type information by manually connecting to schemaregistry flink.apache.org/news/2020/04/15/… Jun 11, 2022 at 6:33
  • 1
    It works but falls back to kryo.. do you know why ? Oct 10, 2022 at 18:47
  • I'm facing the same issue @SuchethShivakumar. Did you resolve this? If so, could you please share a code snippet or something?
    – guru
    Jan 18 at 17:20

Your Answer

Reminder: Answers generated by Artificial Intelligence tools are not allowed on Stack Overflow. Learn more

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.