KAFKA提交问题

在代码中，可以使用以下方法解决Kafka提交问题：

1.首先，确保Kafka生产者已经成功发送消息到Kafka Broker。

2.使用带有回调函数的commitAsync()方法来提交偏移量，以确保在Kafka Broker上正确提交偏移量。

例子:

// 创建Kafka消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 订阅消息主题
consumer.subscribe(Collections.singletonList(topic));

// 循环接收消息
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record : records) {
        // 处理消息
        System.out.printf("topic = %s, partition = %s, offset = %d, value = %s%n",
                record.topic(), record.partition(), record.offset(), record.value());
    }
    // 提交偏移量
    consumer.commitAsync(new OffsetCommitCallback() {
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
            if (exception != null) {
                System.err.printf("Commit failed for offsets %s%n", offsets);
                exception.printStackTrace();
            }
        }
    });
}

注意：使用commitAsync()方法时，一定要记得指定回调函数。如果提交失败，回调函数会打印异常信息。