我正在解析一条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);
}
但我仍然得到的错误是“不是一个数据文件”。
1条答案
按热度按时间rjjhvcjd1#
反序列化不需要解密
首先你得到你的模式行
那么