为什么 commitAsync 无法提交前 2 个偏移量答案

【问题标题】：Why commitAsync fails to commit the first 2 offsets为什么 commitAsync 无法提交前 2 个偏移量
【发布时间】：2016-06-15 15:38:15
【问题描述】：

我遇到了一个奇怪的问题，即消费者无法将comitAsync 设为日志的前两个偏移量，我不知道原因。这很奇怪，因为消费者收到并成功提交了生产者的同一异步发送的其他消息。有人能找到这个问题的根源吗。我在下面引用我的代码和一个输出示例

package com.panos.example;

import kafka.utils.ShutdownableThread;
import org.apache.kafka.clients.consumer.*;

import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;

public class Consumer extends ShutdownableThread {
    private final KafkaConsumer<Integer, String> consumer;
    private final String topic;

    public Consumer(String topic) {
        super("KafkaConsumerExample", false);
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.75:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        consumer = new KafkaConsumer<Integer, String>(props);
        this.topic = topic;
    }

    @Override
    public void doWork() {

        consumer.subscribe(Collections.singletonList(this.topic));
        try {
            ConsumerRecords<Integer, String> records = consumer.poll(1000);
            long startTime = System.currentTimeMillis();
            if (!records.isEmpty()) {
                System.out.println("C : {} Total No. of records received : {}" + records.count());

                for (ConsumerRecord<Integer, String> record : records) {
                    System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
                    consumer.commitAsync(new ConsumerCallBack(startTime,record.value(), record.offset()));
                }

            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public String name() {
        return null;
    }

    @Override
    public boolean isInterruptible() {
        return false;
    }


    class ConsumerCallBack implements OffsetCommitCallback {

        private final long startTime;
        private String message;
        private final String NewLine = System.getProperty("line.separator");
        private long offset;

        public ConsumerCallBack(long startTime) {
            this.startTime = startTime;
        }

        public ConsumerCallBack(long startTime, String message, long offset) {
            this.startTime = startTime;
            this.message=message;
            this.offset = offset;
        }

        public void onComplete(Map<TopicPartition, OffsetAndMetadata> CurrentOffset,
                               Exception exception) {
            long elapsedTime = System.currentTimeMillis() - startTime;
            if (exception != null) {
                System.out.println("Message : {" + message + "}, committed successfully at offset " + offset +
                        CurrentOffset + "elapsed time :" + elapsedTime);
            } else {
                System.out.println(exception.toString());
               /* JOptionPane.showMessageDialog(new Frame(),
                        "Something Goes Wrong with the Server Please Try again Later.",
                        "Inane error",
                        JOptionPane.ERROR_MESSAGE);*/
            }
        }
    }
}

您可以看到所有消息都成功提交，除了前 2 个，没有任何例外。为什么会这样？

Received message: (1, Message_1) at offset 160
Received message: (2, Message_2) at offset 161
Received message: (3, Message_3) at offset 162
Received message: (4, Message_4) at offset 163
Message : {Message_3}, committed successfully at offset 162{test-0=OffsetAndMetadata{offset=164, metadata=''}}elapsed time :6
Message : {Message_4}, committed successfully at offset 163{test-0=OffsetAndMetadata{offset=164, metadata=''}}elapsed time :6

【问题讨论】：

标签： apache-kafka kafka-consumer-api

【解决方案1】：

如果您使用commitAsync，则可能会发生多个提交被压缩到单个提交消息中的情况。由于偏移量是按递增顺序提交的，因此偏移量 X 的提交是所有小于 X 的偏移量的隐式提交。在您的情况下，似乎提交或前三个偏移量是通过偏移量 3 的单个提交完成的.

【讨论】：