Schema-registry: Reflection-based avro serialization is not supported

[phaas]

The AbstractKafkaAvroSerializer only handles SpecificRecord and GenericRecords. There is no way to send a properly annotated POJO.

I've tried to work around this by using a NonRecordContainer with raw avro-byte[]'s, but that is not a supported option either (The serializer will re-encode the byte[]s but send it using the schema of the NonRecordContainer.)

[phaas]

A workaround is to extend the serializer and perform the required encoding:

/**
 * Extension of KafkaAvroSerializer that supports reflection-based serialization of values when
 * objects are wrapped in a ReflectContainer.
 */
public class KafkaAvroReflectSerializer extends KafkaAvroSerializer {
    public static class ReflectContainer {
        private final Object object;

        public ReflectContainer(Object object) {
            this.object = object;
        }

        public Object getObject() {
            return object;
        }
    }

    private static final EncoderFactory ENCODER_FACTORY = EncoderFactory.get();
    private static final ReflectData REFLECT_DATA = ReflectData.get();

    @Override
    protected byte[] serializeImpl(String subject, Object object) throws SerializationException {
        if (object instanceof ReflectContainer) {
            Object value = ((ReflectContainer) object).getObject();
            if (value == null) {
                return null;
            }
            Schema schema = REFLECT_DATA.getSchema(value.getClass());
            try {
                int registeredSchemaId = this.schemaRegistry.register(subject, schema);

                ByteArrayOutputStream out = new ByteArrayOutputStream();
                out.write(0);
                out.write(ByteBuffer.allocate(4).putInt(registeredSchemaId).array());

                DatumWriter<Object> dw = new ReflectDatumWriter<>(schema);
                Encoder encoder = ENCODER_FACTORY.directBinaryEncoder(out, null);
                dw.write(value, encoder);
                encoder.flush();
                return out.toByteArray();
            } catch (RuntimeException | IOException e) {
                throw new SerializationException("Error serializing Avro message", e);
            } catch (RestClientException e) {
                throw new SerializationException("Error registering Avro schema: " + schema, e);
            }
        }
        return super.serializeImpl(subject, object);
    }
}

[ewencp]

@phaas Yeah, this is absolutely something we'd want to support in the serializer. The ReflectContainer approach might not even be necessary if we just have a config that makes it use ReflectDatumWriter -- I'd imagine most apps want to use only one of (SpecificRecord, GenericRecord, POJOs).

[miguno]

See also the related discussion at http://stackoverflow.com/questions/39606026/kafkaavrodeserializer-does-not-return-specificrecord-but-returns-genericrecord, including some code snippets on how to make it work. /cc @ewencp

[bobby]

I've added PR #1111 in an attempt to address this issue.