解密kafka avro消息

d7v8vwbk  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(501)

我正在解析一条Kafka的信息,它是某种加密的avro格式。我有以下avroschema.avsc avro模式文件:

{
    "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" }
        ]
 }

现在,我编写了以下代码来获取json格式的数据:

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);
                        }

请帮我解密这封信。
加密字节消息的类型如下: 080-21-0001 :�Aw�@@��A�ǐ�U :�Aw�@@��A 我按照建议进行了更改,现在我有了以下代码:

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);
                }

但我仍然得到的错误是“不是一个数据文件”。

rjjhvcjd

rjjhvcjd1#

反序列化不需要解密
首先你得到你的模式行

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

那么

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);

相关问题