Apache AVRO support

[cagalmarini]

Hi, in the company I work for we've started to work with Kafka, and we're using Bridge as an alternative for some legacy clients that are no able to interact with Kafka.
We've already implemented some use cases using JSON as the embebbed format, nowadays we're investigating further on apache avro and we wonder if there is any support to produce Avro encoded messages over the bridge and on the other hand consume records from topics that have avro serialized messages.

Thanks in advance

[scholzj]

Right now, the Bridge supports plaintext, binary, and JSON messages. There is no support for AVRO. But you should be able to send them as binary I assume, or @ppatierno?

[ppatierno]

Yeah, it's the only viable way we have today on the bridge for AVRO.

[cagalmarini]

Thank you, I'll try it out.

[cagalmarini]

Hi, just in case anyone else needs a solution for this I'm posting what I did to be able to deal with avro messages and the bridge.

1. Messages posted to topic have to be, as you said, application/vnd.kafka.binary.v2+json. Avro objects are serialized using base64. For example:
   curl --location '[your bridge host]/topics/test-topic-avro-bridge2' \ --header 'Content-Type: application/vnd.kafka.binary.v2+json' \ --data '{ "records": [ { "value": "T2JqAQIWYXZyby5zY2hlbWGmA3sidHlwZSI6InJlY29yZCIsIm5hbWUiOiJGdWxsTmFtZSIsIm5hbWVzcGFjZSI6ImNvbS5leGFtcGxlIiwiZmllbGRzIjpbeyJuYW1lIjoiZmlyc3QiLCJ0eXBlIjp7InR5cGUiOiJzdHJpbmciLCJhdnJvLmphdmEuc3RyaW5nIjoiU3RyaW5nIn19LHsibmFtZSI6Imxhc3QiLCJ0eXBlIjp7InR5cGUiOiJzdHJpbmciLCJhdnJvLmphdmEuc3RyaW5nIjoiU3RyaW5nIn19XX0AeDD1Wl7fRTEfTWpmvZxJ7QIeDENhcmxvcw5BbGJlcnRveDD1Wl7fRTEfTWpmvZxJ7Q==" } ] }'
2. Those messages when consumed from a bridge are retrived as base64, so is up to you process them to get the real object. Consumer must be created using "format": "binary", Accept header on GET /records request must be application/vnd.kafka.binary.v2+json
3. If you are using a Java client, you need to use a org.apache.kafka.common.serialization.ByteArrayDeserializer deserializer, the code I wrote to consume messages is:

KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<String, byte[]>(props);

consumer.subscribe(Arrays.asList("test-topic-avro-bridge2"));

try {
	while(true) {
		ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));

		for(ConsumerRecord<String, byte[]> record : records) {
			byte[] bytes = record.value();
			
			FullName name = new FullName(); //this is the avro object generated by maven plugin based on an avro schema
			
                        //maybe there is a better way for the next block of code, I'll check it.
			DatumReader<FullName> dr = new SpecificDatumReader<FullName>(name.getSchema());
			SeekableByteArrayInput bin = new SeekableByteArrayInput(bytes);
			DataFileReader<FullName> dfr = null;
			try {
				dfr = new DataFileReader<FullName>(bin, dr);
				dfr.next(name);	
			}
			catch (Exception e) {
				e.printStackTrace();
			}
			finally {
				dfr.close();
			}
			
                         //then we're able to successfully retrieve object members
			System.out.println(name.getFirst());
			System.out.println(name.getLast());	
		}
		
	}
}
catch(Exception e) {
	e.printStackTrace();
}
finally {
	consumer.close();
}

4. Finally, I was thinking that it might be possible to use an AvroDeserializer and avoid to deal with all the stuff related to object creation from bytes array

[ppatierno]

@cagalmarini that's great to hear! Are you planning to try the AvroDeserializer as well? I agree that it could simplify a lot the consuming part.

[cagalmarini]

Yes, i'm trying it, but as the first results I'm dealing with the fact that the deserializer I have available right now comes from apicurio registry, it expects an artifact id which is not available in the message (remember that I created the avro record from a "local schema"), so I think I have two choices:

• Working with the registry from the very beginning (get the schema from the registry, create the object from that schema, encode and send it)
• Find a "plain avro deserializer", if such thing exists