java - 在消费者中反序列化加密的kafka消息

Question

我有一个 kafka 主题，其中包含没有 avro 模式的消息。我们最近想使用 avro 模式将消息推送到该主题。

现在该主题包含带有/不带有架构的消息。

我有一个消费者从这个主题消费。

-> 如果我设置 value.deserializer在消费者配置中为“KafkaAvroDeserializer.class ”，我没有看到任何消息被消费。

-> 如果我设置 value.deserializer在消费者配置为“StringDeserializer.class ”中，我能够使用这些消息，但是，具有 avro 模式的消息现在看起来是加密的。

例如:ConsumerRecord(topic = sample-events, partition = 2, offset = 1089595, CreateTime = 1544116093932, checksum = 2421249481, serialized key size = -1, serialized value size = 159, key = null, value = ���test_impressLbhpb_extranet_opportunity_cleaning_fecron�����YH00756f54-ba55-11e7-8df0-fdb86cefa6ed$abcde) .

我已经为 avro 模式生成了 java 类，并且我想将来自消费者的带/不带模式的消息转换为生成的 avro java 类。我能够使用 objectMapper 将没有模式的消息映射到 avro java 类。

但是对于来自消费者的带有 avro 模式的消息(看起来像是示例中提到的加密的)，我正在尝试下面的代码片段:

SpecificDatumReader<SampleEvents> reader = new SpecificDatumReader<SampleEvents>(SampleEvents.getClassSchema());
Decoder decoder = DecoderFactory.get().binaryDecoder(new ByteArrayInputStream(record), null);
SampleEvents event = reader.read(null, decoder);

但这不起作用。我收到“错误:

java.lang.ArrayIndexOutOfBoundsException: -1".

如何反序列化此消息？

Answer 1

If I set value.deserializer in consumer config as "KafkaAvroDeserializer.class", I don't see any messages being consumed.

嗯，您至少应该收到 HTTP 或反序列化器错误......

<小时/>

首先，您应该使用BytesDeserializer或其变体

然后，你需要熟悉ByteBuffer的方法，并将byte[]转换为一个......

如果您有架构注册表编码的 Avro 消息，那么这些消息的编号为 well-defined wire format

因此，您可能会得到类似以下内容的内容，但最终，它需要对主题中可能包含哪些数据进行一些推断。

// consumerConfig.put("value.deserializer", ByteBufferDeserializer.class)

ByteBuffer buf = record.value();
Deserializer d;

if (buf == null) {
    System.err.println("Tombstoned record");
} else if (buf.get() == 0x0) { // Check for Avro
    int schemaId = buf.getInt();  // If you wanted it

    d = new KafkaAvroDeserializer();        
    Map<String, String> config = new HashMap<>();
    config.put("schema.registry.url", "http://..."); // address to registry
    boolean isKey = false;
    d.configure(config, isKey);

    AvroValue v = d.deserialize(value);
    // TODO: Handle record
} else {
    try {
        d = new StringDeserializer();
        String s = d.deserialize(value);
        // TODO: Handle record
    } catch (Exception e) {
        e.printStackTrace();
    }

}

要点:不要在主题中生成 Avro 和非 Avro 数据类型。否则，您只需消耗字节并自行处理自定义逻辑。