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

我有一个自定义的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 偏移量未提交,但消息未被重新使用。如何强制使用者使用未提交的偏移量重新使用消息?