Deserialize Avro-Nachrichten in bestimmte Daten mit KafkaAvroDecoder

Question

Ich lese von einem Kafka-Thema, das Avro-Nachrichten enthält, die unter Verwendung der KafkaAvroEncoder (die automatisch die Schemas mit den Themen registriert). Ich benutze das maven-avro-plugin, um einfache Java-Klassen zu erzeugen, die ich gerne beim Lesen benutzen würde.

Die KafkaAvroDecoder unterstützt nur die Deserialisierung in GenericData.Record Typen, die (meiner Meinung nach) den ganzen Punkt einer statisch getippten Sprache verfehlt. Mein Deserialisierung Code sieht derzeit wie folgt aus:

SpecificDatumReader<event> reader = new SpecificDatumReader<>(
     event.getClassSchema() // event is my class generated from the schema 
    ); 
    byte[] in = ...; // my input bytes; 
    ByteBuffer stuff = ByteBuffer.wrap(in); 
    // the KafkaAvroEncoder puts a magic byte and the ID of the schema (as stored 
    // in the schema-registry) before the serialized message 
    if (stuff.get() != 0x0) { 
     return; 
    } 
    int id = stuff.getInt(); 

    // lets just ignore those special bytes 
    int length = stuff.limit() - 4 - 1; 
    int start = stuff.position() + stuff.arrayOffset(); 

    Decoder decoder = DecoderFactory.get().binaryDecoder(
     stuff.array(), start, length, null 
    ); 
    try { 
     event ev = reader.read(null, decoder); 
    } catch (IOException e) { 
     e.printStackTrace(); 
    } 

ich meine Lösung umständlich gefunden, also würde ich gerne wissen, ob es eine einfachere Lösung ist, dies zu tun.

Answer 1

Dank dem Kommentar konnte ich die Antwort finden. Das Geheimnis mit einem Properties zu instanziiert KafkaAvroDecoder war die Verwendung des spezifischen Avro Lesers spezifiziert, das heißt:

Properties props = new Properties(); 
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "..."); 
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
      io.confluent.kafka.serializers.KafkaAvroSerializer.class); 
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
      io.confluent.kafka.serializers.KafkaAvroSerializer.class); 
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_C‌ONFIG, "..."); 
    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true); 
    VerifiableProp vProps = new VerifiableProperties(props); 

    KafkaAvroDecoder decoder = new KafkaAvroDecoder(vProps); 
    MyLittleData data = (MyLittleData) decoder.fromBytes(input); 

Die gleiche Konfiguration für den Fall gilt direkt für die Verwendung der KafkaConsumer<K, V> Klasse (ich raubend von Kafka in Sturm mit dem KafkaSpout aus dem storm-kafka-projekt, das die SimpleConsumer verwendet, muss ich also die messages manuell deserialisieren.Für die mutigen gibt es das storm-kafka-client-projekt, das macht das automatisch mit dem neuen style consumer).