Kafka commitAsync 使用提交顺序重试

Question

我正在阅读 Kafka 权威指南，在消费者章节中有一个关于“重试异步提交”的简介：

为异步重试正确获取提交顺序的一个简单模式是使用单调递增的序列号。每次提交时增加序列号，并将提交时的序列号添加到 commitAsync 回调。当您准备发送重试时，检查回调获得的提交序列号是否等于实例变量；如果是，则没有更新的提交，可以安全地重试。如果实例序列号更高，请不要重试，因为已经发送了更新的提交。

作者的一个简单示例在这里对于像我这样的密集人群来说会很棒。我对上面加粗的部分特别不清楚。

谁能阐明这意味着什么，或者更好地提供一个玩具示例来证明这一点？

Answer 1

这就是我的想法，但谦虚我可能是错的

      try {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(5);
            for (ConsumerRecord<String, String> record : records) {
                System.out.format("offset: %d\n", record.offset());
                System.out.format("partition: %d\n", record.partition());
                System.out.format("timestamp: %d\n", record.timestamp());
                System.out.format("timeStampType: %s\n", record.timestampType());
                System.out.format("topic: %s\n", record.topic());
                System.out.format("key: %s\n", record.key());
                System.out.format("value: %s\n", record.value());
            }

            consumer.commitAsync(new OffsetCommitCallback() {
                private int marker = atomicInteger.incrementAndGet();
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
                                       Exception exception) {
                    if (exception != null) {
                        if (marker == atomicInteger.get()) consumer.commitAsync(this);
                    } else {
                        //Cant' try anymore
                    }
                }
            });
        }
    } catch (WakeupException e) {
        // ignore for shutdown
    } finally {
        consumer.commitSync(); //Block
        consumer.close();
        System.out.println("Closed consumer and we are done");
    }