kafka的auto.offset.reset详解与测试

1. 取值及定义#
   auto.offset.reset有以下三个可选值：

latest (默认)
earliest
none
三者均有共同定义：
对于同一个消费者组，若已有提交的offset，则从提交的offset开始接着消费

意思就是，只要这个消费者组消费过了，不管auto.offset.reset指定成什么值，效果都一样，每次启动都是已有的最新的offset开始接着往后消费

不同的点为：

latest（默认）：对于同一个消费者组，若没有提交过offset，则只消费消费者连接topic后，新产生的数据
就是说如果这个topic有历史消息，现在新启动了一个消费者组，且auto.offset.reset=latest，此时已存在的历史消息无法消费到，那保持消费者组运行，如果此时topic有新消息进来了，这时新消息才会被消费到。而一旦有消费，则必然会提交offset
这时候如果该消费者组意外下线了，topic仍然有消息进来，接着该消费者组在后面恢复上线了，它仍然可以从下线时的offset处开始接着消费，此时走的就是共同定义

earliest：对于同一个消费者组，若没有提交过offset，则从头开始消费
就是说如果这个topic有历史消息存在，现在新启动了一个消费者组，且auto.offset.reset=earliest，那将会从头开始消费，这就是与latest不同之处。
一旦该消费者组消费过topic后，此时就有该消费者组的offset了，这种情况下即使指定了auto.offset.reset=earliest，再重新启动该消费者组，效果是与latest一样的，也就是此时走的是共同的定义

none：对于同一个消费者组，若没有提交过offset，会抛异常
一般生产环境基本用不到该参数

2. 新建全新topic#
   ./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --topic TestOffsetResetTopic --partitions 1 --replication-factor 1 --create

3. 往新建的topic发送消息#
   便于测试，用Java代码发送5条消息

public class TestProducer {

public static void main(String[] args) throws InterruptedException {
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.123.124:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

    String topic = "TestOffsetResetTopic";

    for (int i = 0; i < 5; i++) {
        String value = "message_" + i + "_"
1
2
3
4
5
6
7
8
9
10