Open malhomaid opened 3 years ago
I will preface my comment with - "I AM NOT A JAVA DEVELOPER, AND I DON'T KNOW JAVA." I will also preface my comment with - "I AM NOT A DEVELOPER AT ALL"
I am a DBA that is dabbling in Data Engineering, but coding is my weakness.
I ran into an issue with a Python producer that was out of my jurisdiction that was producing to a topic using Avro, but was not using Schema Registry.
The serialization of the message value looked something like this:
def serialize(myschema, myobject):
writer = DatumWriter(myschema)
bytes_writer = io.BytesIO()
encoder = BinaryEncoder(bytes_writer)
writer.write(myobject, encoder)
raw_bytes = bytes_writer.getvalue()
return (raw_bytes)
I was running into an issue with the RegistrylessAvroConverter - it was complaining that the value of the messages was not a valid Avro file. So, I went looking and found this issue from @mhomaid1 and got some feedback from the folks at r/ApacheKafka.
His solution worked, for the most part. Here is what mine ended up looking like:
public SchemaAndValue toConnectData(String topic, byte[] value) {
DatumReader<GenericRecord> datumReader;
if (avroSchema != null) {
datumReader = new GenericDatumReader<>(avroSchema);
} else {
datumReader = new GenericDatumReader<>();
}
GenericRecord instance;
Decoder binaryDecoder = DecoderFactory.get().binaryDecoder(value, null);
try {
instance = datumReader.read(null, binaryDecoder);
} catch (IOException ioe) {
throw new DataException("Failed to deserialize Avro data from topic %s :".format(topic), ioe);
}
if (instance == null) {
logger.warn("Instance was null");
}
if (avroSchema != null) {
return avroDataHelper.toConnectData(avroSchema, instance);
} else {
return avroDataHelper.toConnectData(instance.getSchema(), instance);
}
}
I also had to have the correct imports:
import org.apache.avro.io.*;
I then had to compile while ignoring the test. I am guessing this was because I changed the .java file and the tests were no longer applicable to how the class was coded.
./gradlew build -x test
After that, my connector worked and was able to move data downstream. I am not happy about it, but I do appreciate the work of both @farmdawgnation and @mhomaid1
Hi Matt, Thank you for this project.
I had to use
BinaryDecoder
to decode my messages but I didn't know how to tellDataFileReader
to use theBinaryDecoder
to decode my messages so I usedGenericDatumReader
directly, I'm assuming you are usingDataFileReader
in case the avro file contained more than one message(which isn't my case at least for now), is this something from the Connect framework ?Attached my code: