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

Question

我有一个自定义的Kafka消费者，我用它向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

如果计划使用seek（），请确保您的数据是幂等的。由于您是有选择地提交偏移量，所以被遗漏的记录可能会在提交（成功处理）记录之前。如果您执行seek（）--它将GroupID的指针移动到未提交的偏移量并开始重播，您也将得到那些成功处理的消息。它也有可能成为一个无限循环。

或者，您可以将不成功记录的元数据保存在内存或db中，并从“poll（retention.ms）”开始重播主题，以便重播所有记录，但添加一个筛选器，只处理那些通过API进行的元数据与您先前保存的数据匹配的记录。每小时或几小时进行一次批处理。