Hi all, I have a few doubts regarding the usage of Schema Registry URL in the

ConfluentRegistryAvroDeserializationSchema.forGeneric(...)

method in the

flink-avro-confluent-registry

package. 1. What is the use of passing this as an argument? The way I understood it, its only used to set the writer schema and nothing else. 2. If say I have a topic with 2 records r1 with schema s1 and r2 with schema s2 (s1 and s2 are slightly different - s2 has one new column than s1). Both schemas are registered in the schema registry. When I run a

FlinkKafkaConsumer

, and configure the deserialization schema as

ConfluentRegistryAvroDeserializationSchema.forGeneric(s1, URL)

, I see that the new column in r2 is completely ignored. I was under the impression that if the latest schema s2 is compatible with the record that was read, and that it would be used to deserialize the record. But now, that is not happening. What is the point of passing the URL? How do I make my consumer in such a way that they don't need to know the schema beforehand, but use the writer schema from the registry? Additionally, I tried to follow this post to write my own deserialization logic, but it fails with a kryo exception. 😞 Can someone please elaborate on this?

Hi!

How do I make my consumer in such a way that they don’t need to know the schema beforehand, but use the writer schema from the registry?

• First, you get the relevant schema from the registry using

io.confluent.kafka.schemaregistry.client.SchemaRegistryClient

, e.g.

schemaRegistryClient.getLatestSchemaMetadata(schemaSubject).getSchema

• Then you convert it to a DataType using

AvroSchemaConverter

, e.g.

val dataType: DataType = AvroSchemaConverter.convertToDataType(avroSchemaString)

• Cast DataType to RowType:

val rowType = dataType.getLogicalType.asInstanceOf[RowType]

• Then use a custom

KafkaRecordDeserializationSchema

and create a converter using

AvroToRowDataConverters.createRowConverter(rowType)

• Then you can convert raw

ConsumerRecord[Array[Byte], Array[Byte]]

to RowData like this

val rowData = converter.convert(avroPayload).asInstanceOf[RowData]

Thanks @sap1ens for the snippet. Calling the schema registry using the client would work in a scenario where the schema is not expected to change frequently. Please correct me if I'm missing something here. Rest of them make sense but again, expecting a consumer of my data to know all of this would become harder for them to adopt the platform. I would like it to be an automatic update of consumer schema as and when the producer updates it. I would ideally not want clients to call the registry explicitly - we want to abstract this from them as it is too much implementation. My ideal

FlinkKafkaConsumer

/

KafkaSource

client should look like this:

FlinkKafkaConsumer<GenericRecord> consumer = new FlinkKafkaConsumer<>(
          properties.getProperty("source.topic"),
          ConfluentRegistryAvroDeserializationSchema.forGeneric("REGISTRY_URL"),
          kafkaProperties);
        consumer.setStartFromEarliest();
        DataStream<GenericRecord> consumerStream = env.addSource(consumer);

You can theoretically move some of the logic inside the consumer, read the first 4 bytes to get the schema id and then fetch it. However, you will only be able to work with GenericRecords in this case, not the Flink type system (since it needs to know all the types during the startup).

@sap1ens Yes, that is something that's doable I believe. But doing it this way leads me to Kryo exception which I don't understand why. Flink is forcing to use its kryo libraries even when I have set

env.getConfig().enableForceAvro();

I am ok with

GenericRecord

as the avro schema will usually have 1000s of columns and using a POJO or

SpecificRecord

format would be very cumbersome. My custom deserialization schema looks like this:

public class KafkaGenericAvroDeserializationSchema implements KafkaDeserializationSchema<GenericRecord> {

  private final String registryUrl;
  private transient KafkaAvroDeserializer avroDeserializer;

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

  @Override
  public void open(DeserializationSchema.InitializationContext context) throws Exception {
    checkInitialized();
    KafkaDeserializationSchema.super.open(context);
  }

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

  @Override
  public GenericRecord deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
    return (GenericRecord) avroDeserializer.deserialize("sod-sample-confluent", consumerRecord.value());
  }

  @Override
  public void deserialize(ConsumerRecord<byte[], byte[]> consumerRecord, Collector<GenericRecord> collector)
      throws IOException {
    collector.collect((GenericRecord) avroDeserializer.deserialize("avro-sample-confluent", consumerRecord.value()));
  }

  private void checkInitialized() {
    if (avroDeserializer == 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);
      avroDeserializer = new KafkaAvroDeserializer(client, props);
    }
  }

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

This is leading to the following exception:

Caused by: com.esotericsoftware.kryo.KryoException: java.lang.UnsupportedOperationException
Serialization trace:
reserved (org.apache.avro.Schema$Field)
fieldMap (org.apache.avro.Schema$RecordSchema)
schema (org.apache.avro.generic.GenericData$Record)
	at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
	at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:143)
	at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
	at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
	at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
	at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
	at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:354)
	at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:191)
	at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:46)
	at org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:53)
	at org.apache.flink.runtime.io.network.api.serialization.NonSpanningWrapper.readInto(NonSpanningWrapper.java:337)
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNonSpanningRecord(SpillingAdaptiveSpanningRecordDeserializer.java:128)
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:103)
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:93)
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:95)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:496)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:809)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:761)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
	at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.UnsupportedOperationException
	at java.util.Collections$UnmodifiableCollection.add(Collections.java:1057)
	at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:109)
	at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
	at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
	at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)

Any idea why this happens?

Yes, GenericRecords are not serializable with Kryo by default. You can fix it with this, but using Kryo will affect performance. Personally, I couldn’t make

env.getConfig().enableForceAvro();

work. Instead, I make sure that there are no steps in my topology that require serialization and I convert GenericRecord to a class as soon as possible.

Will try this out, thank you for the suggestions!

Nice, thanks!

When do you use

<GenericRecord>

,

forGeneric

with your

FlinkKafkaConsumer

?

@Reme Ajayi - if you want your consumer to have a schema beforehand, you can pass it via the

forGeneric(...)

method. If not, you can do what @sap1enssuggested. Below is the code snippet for the custom deserialization schema I had to implement:

import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

@Slf4j
public class KafkaGenericAvroDeserializationSchema implements KafkaDeserializationSchema<GenericRecord> {

  private final String registryUrl;
  private final String sourceTopic;

  private transient KafkaAvroDeserializer avroDeserializer;

  private static final Map<Integer, Schema> SCHEMA_CACHE = new HashMap<>();
  private CachedSchemaRegistryClient cachedSchemaRegistryClient;

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

  @Override
  public void open(DeserializationSchema.InitializationContext context) throws Exception {
    cachedSchemaRegistryClient = new CachedSchemaRegistryClient(
        registryUrl, AbstractKafkaAvroSerDeConfig.MAX_SCHEMAS_PER_SUBJECT_DEFAULT);
    KafkaDeserializationSchema.super.open(context);
  }

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

  @Override
  public GenericRecord deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
    long start = System.currentTimeMillis();
    final Schema schema = checkInitialized(consumerRecord.value());
    GenericRecord record =  (GenericRecord) avroDeserializer.deserialize("sod-sample-confluent", consumerRecord.value(),
        schema);
    long end = System.currentTimeMillis();
    <http://log.info|log.info>("deserialize time taken: {}ms", end-start);
    return record;
  }

  private Schema checkInitialized(byte[] value) throws RestClientException, IOException {
    if (avroDeserializer == null) {
      Map<String, Object> props = new HashMap<>();
      props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, registryUrl);
      props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, false);

      avroDeserializer = new KafkaAvroDeserializer(cachedSchemaRegistryClient, props);
    }

    <http://log.info|log.info>("schema caching");
    if (value.length < 4) {
      log.error("length < 4");
      return null;
    }
    // schema ID is bytes 1 to 4, byte 0 is magic byte.
    ByteBuffer schemaIdBuffer = ByteBuffer.wrap(value, 1, 4);
    Integer schemaId = schemaIdBuffer.getInt();
    <http://log.info|log.info>("schema id: {}", schemaId);
    long startTime = System.currentTimeMillis();
    Schema schema = cachedSchemaRegistryClient.getBySubjectAndID(sourceTopic+"-value", schemaId);
    long endTime = System.currentTimeMillis();
    <http://log.info|log.info>("got the schema: {} with id: {}, time taken: {}ms", schema, schemaId, endTime-startTime);
    return schema;
  }

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

For your

FlinkKafkaConsumer

client instance:

FlinkKafkaConsumer<GenericRecord> consumer = new FlinkKafkaConsumer<>(
          sourceTopic,
          new KafkaGenericAvroDeserializationSchema(REGISTRY_URL, sourceTopic),
          kafkaProperties);
        consumer.setStartFromEarliest();
        DataStream<GenericRecord> consumerStream = env.addSource(consumer);

To avoid Kryo deserialization exceptions, add the following :

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    ...
    AvroKryoSerializerUtils avroKryoSerializerUtil = new AvroKryoSerializerUtils();
    avroKryoSerializerUtil.addAvroSerializersIfRequired(env.getConfig(), GenericData.Record.class);
    env.getConfig().enableForceAvro();

thank you for your response. I am new to Java and Flink. My use case is also a bit simpler. I only have one schema.

Oh, that’s pretty straightforward. You need to import org.Apache.avro maven package. To parse the schema, you can refer this: https://avro.apache.org/docs/1.7.4/api/java/org/apache/avro/Schema.Parser.html Schema schema = new Schema.Parser().parse(new File("/path/to/emp.avsc"));

Thanks a lot!!! I got it to work!!!!