java - Kafka Java SimpleConsumer奇怪的编码

Question

我正在尝试使用 Kafka 9 中的 SimpleConsumer 来允许用户从一个时间偏移量重播事件 - 但我从 Kafka 收到的消息采用一种非常奇怪的编码:

7icf-test-testEvent.ebebf1a4.2911.431d.a138.f5d6db4647d7\�W>8������{"namespace":"test","type":"testEvent.ebebf1a4.2911.431d.a138.f5d6db4647d7","received":1464819330373,"context":{"userid":0,"username":"testUser"}}�!}�a�����{"namespace":"test","type":"testEvent.ebebf1a4.2911.431d.a138.f5d6db4647d7","received":1464819331637,"context":{"userid":1,"username":"testUser"}}���r�����{"namespace":"test","type":"testEvent.ebebf1a4.2911.431d.a138.f5d6db4647d7","received":1464819332754,"context":{"userid":2,"username":"testUser"}}��������{"namespace":"test","type":"testEvent.ebebf1a4.2911.431d.a138.f5d6db4647d7","received":1464819333868,"context":{"userid":3,"username":"testUser"}}�p=
                            ������{"namespace":"test","type":"testEvent.ebebf1a4.2911.431d.a138.f5d6db4647d7","received":1464819334997,"context":{"userid":4,"username"

使用 KafkaConsumer 可以很好地解析此消息。这是我使用 SimpleConsumer 检索消息的代码:

    for (MessageAndOffset messageAndOffset : fetchResponse.messageSet(topic, partition)) {
        long currentOffset = messageAndOffset.offset();
        if (currentOffset < readOffset) {
            log.debug("Found an old offset - skip");
            continue;
        }

        readOffset = messageAndOffset.nextOffset();

        int payloadOffset = 14 + messageAndOffset.message().keySize(); // remove first x bytes, schema Id
        byte[] data = messageAndOffset.message().payload().array();
        byte[] realData = Arrays.copyOfRange(data, payloadOffset, data.length - payloadOffset);
        log.debug("Read " + new String(realData, "UTF-8"));
}

在我不断收到有关字节过高的 UTF-32 错误后，我添加了代码以跳过前 x 个字节，我认为这是因为 Kafka 将消息大小等信息添加到负载中。这是 Avro 工件吗？

Answer 1

我从来没有找到一个很好的答案 - 但我转而使用 SimpleConsumer 查询 Kafka 的偏移量 我需要的(每个分区......虽然实现很差)，然后使用使用 seek(TopicPartition, offset) 或 seekToBeginning(TopicPartition) 的原生 KafkaConsumer 检索消息。希望他们会在下一个版本中为 native 客户端添加从给定时间戳检索消息的能力。