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 来检索消息 。希望他们会在下一个版本中为本机客户端添加从给定时间戳检索消息的能力。

Answer 2

你在找这个吗？

readOffset = messageAndOffset.nextOffset();
ByteBuffer payload = messageAndOffset.message().payload();

    if(payload == null) {
        System.err.println("Message is null : " + readOffset);
        continue;
    }

final byte[] realData = new byte[payload.limit()];
payload.get(realData);
System.out.println("Read " + new String(realData, "UTF-8"));

Answer 3

您可以使用消息的时间戳定期记录分区您提交的偏移量（可能不是每次提交），然后您可以在将来有一些措施来设置您的消费者偏移量。我认为这是用于生产调试。

我怀疑他们会添加这样的功能，考虑到 Kafka 的工作原理，这似乎是不可行的，尽管我可能错了，总会有天才的事情发生。我会做日志记录的事情。

Kafka Java SimpleConsumer 奇怪的编码

Kafka Java SimpleConsumer strange encoding

java

apache-kafka

kafka-consumer-api