Entschlüsseln Kafka Avro Nachricht

Question

Ich versuche, eine Kafka-Nachricht zu analysieren, es ist in einem verschlüsselten AVRO-Format. Ich habe folgendes AvroSchema.avsc avro Schemadatei:

{ 
    "type": "record", 
    "namespace": "kafka.events", 
    "name": "AvroSchema", 
     "fields": [ 
      { "name": "product_id", "type": "string" }, 
      { "name": "available_to_promise_quantity", "type": "double" }, 
      { "name": "online_available_to_promise_quantity", "type": "double" }, 
      { "name": "stores_available_to_promise_quantity", "type": "double" }, 
      { "name": "is_infinite_inventory", "type": "boolean", "default" : false }, 
      { "name": "event_timestamp", "type": "long" }, 
      { "name": "previous_event", "type": "AvroSchema" } 
     ] 
} 

Jetzt habe ich den folgenden Code geschrieben, um die Daten im JSON-Format zu erhalten:

for (final KafkaStream<byte[], byte[]> stream : streams){ 
    ConsumerIterator<byte[], byte[]> consumerIterator = stream.iterator(); 
    byte[] consumedEncryptedMessage; 
    MessageAndMetadata<byte[], byte[]> consumedEntry; 
    while(consumerIterator.hasNext()){ 
     consumedEntry = consumerIterator.next(); 
      if(null != consumedEntry){ 
       consumedEncryptedMessage = consumedEntry.message(); 
        try { 
          Schema schema = null; 
          schema = new Schema.Parser().parse(new File("src/AvroSchema.avsc")); 
          DatumReader<GenericRecord> reader = new SpecificDatumReader<GenericRecord>(schema); 
          Decoder decoder = DecoderFactory.get().binaryDecoder(consumedEncryptedMessage , null); 
          GenericRecord decryptedmsg = null; 
          decryptedmsg = reader.read(null, decoder); 
          System.out.println(decryptedmsg); 
         } 
         catch(Exception e) { 
          e.printStackTrace(); 
          System.out.println(e); 
         } 

Bitte helfen Sie mir, wie Sie die Nachricht entschlüsseln.

Die verschlüsselte Byte-Nachricht ist von dieser Art: 080-21-0001 :�Aw�@@��A�ǐ�U :�Aw�@@��A

ich die Änderungen wie vorgeschlagen, und jetzt habe ich folgende Codestück:

while(consumerIterator.hasNext()){ 
    consumedEntry = consumerIterator.next(); 
     if(null != consumedEntry){ 
      consumedEncryptedMessage = consumedEntry.message(); 
       try { 
        Schema schema = new Schema.Parser().parse(new File("src/AVROSchema.avsc")); 
        File myfile = new File("/Users/z001ldc/Desktop/myfile.txt"); 
        DatumReader<GenericRecord> reader = new GenericDatumReader<GenericRecord>(schema); 
        FileUtils.writeByteArrayToFile(myfile, consumedEncryptedMessage); 
        @SuppressWarnings("resource") 
        DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(myfile, reader); 
        while (dataFileReader.hasNext()) { 
         decryptedMessage = dataFileReader.next(decryptedMessage); 
         System.out.println(decryptedMessage.get("product_id").toString()); 
        } 
       } 
       catch(Exception e) { 
        e.printStackTrace(); 
        System.out.println(e); 
       } 

Aber noch bin ich immer den Fehler als " Keine Datei ".

Answer 1

Deserialisieren erfordern wont Entschlüsselung

Zuerst erhalten Sie Ihr Schema Linie

schema = new Schema.Parser().parse(new File("src/AvroSchema.avsc")); 
DatumReader<GenericRecord> reader = new GenericDatumReader<GenericRecord>(schema); 

Dann

DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(consumedEncryptedMessage, reader); 
GenericRecord user = null; 
while (dataFileReader.hasNext()) { 
// Reuse user object by passing it to next(). This saves us from 
// allocating and garbage collecting many objects for files with 
// many items. 
user = dataFileReader.next(user); 
System.out.println(user);