重新使用未提交偏移量的消息

Question

我有一个自定义的Kafka Consumer，用于将一些请求发送到REST API。根据API的响应，我要么提交偏移量，要么跳过不提交的消息。

最小示例：

while (true) {

    ConsumerRecords<String, Object> records = consumer.poll(200);

    for (ConsumerRecord<String, Object> record : records) {

        // Sending a POST request and retrieving the answer

        // ...

        if (responseCode.startsWith("2")) {

            try { 

               consumer.commitSync();

            } catch(CommitFailedException ex) {

              ex.printStackTrace(); 

            }

        } else {

              // Do Nothing

        }

    }

}

现在，当来自REST API的响应未以开头时，2不会提交偏移量，但是不会重新使用该消息。如何强制使用者重新使用未提交的偏移量的邮件？

Answer 1

提交偏移量只是存储使用者当前偏移量（也称为位置）的一种方法。因此，如果它停止了，它（或新的使用者实例将接管）可以找到其先前的位置并从那里重新开始消费。因此，即使您不提交，一旦收到记录，使用者的位置也会被移动。如果要重新使用某些记录，则必须更改使用者的当前位置。在Java客户端中，您可以使用来设置位置seek()。在您的方案中，您可能想计算相对于当前位置的新位置。如果是这样，您可以使用查找当前位置position()。