0
点赞
收藏
分享

微信扫一扫

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

莞尔小迷糊 2022-09-17 阅读 87
kafkajavaapacheJava编程语言

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 + "_" + LocalDateTime.now();
            System.out.println("Send value: " + value);
            producer.send(new ProducerRecord<>(topic, value), (metadata, exception) -> {
                if (exception == null) {
                    String str = MessageFormat.format("Send success! topic: {0}, partition: {1}, offset: {2}", metadata.topic(), metadata.partition(), metadata.offset());
                    System.out.println(str);
                }
            });
            Thread.sleep(500);
        }

        producer.close();
    }
}

发送消息成功：

Send value: message_0_2022-09-16T18:26:15.943749600
Send success! topic: TestOffsetResetTopic, partition: 0, offset: 0
Send value: message_1_2022-09-16T18:26:17.066608900
Send success! topic: TestOffsetResetTopic, partition: 0, offset: 1
Send value: message_2_2022-09-16T18:26:17.568667200
Send success! topic: TestOffsetResetTopic, partition: 0, offset: 2
Send value: message_3_2022-09-16T18:26:18.069093600
Send success! topic: TestOffsetResetTopic, partition: 0, offset: 3
Send value: message_4_2022-09-16T18:26:18.583288100
Send success! topic: TestOffsetResetTopic, partition: 0, offset: 4

现在​​TestOffsetResetTopic​​这个topic有5条消息，且还没有任何消费者组对其进行消费过，也就是没有任何​​offset​​

4\. 测试latest​​#​​

已知topic已经存在5条历史消息，此时启动一个消费者

public class TestConsumerLatest {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.123.124:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 指定消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
        // 设置 auto.offset.reset
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        String topic = "TestOffsetResetTopic";
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singletonList(topic));

        // 消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }

    }
}

发现如上面所述，历史已存在的5条消息不会消费到，消费者没有任何动静，现在保持消费者在线

启动​​TestProducer​​再发5条消息，会发现这后面新发的，​​offset​​从​​5​​开始的消息就被消费了

ConsumerRecord(topic = TestOffsetResetTopic, partition = 0, leaderEpoch = 0, offset = 5, CreateTime = 1663329725731, serialized key size = -1, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = message_0_2022-09-16T20:02:05.523581500)
ConsumerRecord(topic = TestOffsetResetTopic, partition = 0, leaderEpoch = 0, offset = 6, CreateTime = 1663329726251, serialized key size = -1, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = message_1_2022-09-16T20:02:06.251399400)
ConsumerRecord(topic = TestOffsetResetTopic, partition = 0, leaderEpoch = 0, offset = 7, CreateTime = 1663329726764, serialized key size = -1, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = message_2_2022-09-16T20:02:06.764186200)
ConsumerRecord(topic = TestOffsetResetTopic, partition = 0, leaderEpoch = 0, offset = 8, CreateTime = 1663329727264, serialized key size = -1, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = message_3_2022-09-16T20:02:07.264268500)
ConsumerRecord(topic = TestOffsetResetTopic, partition = 0, leaderEpoch = 0, offset = 9, CreateTime = 1663329727778, serialized key size = -1, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = message_4_2022-09-16T20:02:07.778469700)

此时该消费者组对于这个topic的​​offset​​已经为​​9​​了，现在停掉这个消费者（下线），再启动​​TestProducer​​发5条消息，接着再启动​​TestConsumerLatest​​，会发现紧接上一次的offset之后开始，即从​​10​​继续消费

如果测试发现没动静，请多等一会，估计机器性能太差...

5\. 测试earliest​​#​​

新建一个测试消费者，设置​​auto.offset.reset​​为​​earliest​​，注意​​groupid​​为新的​​group2​​，表示对于topic来说是全新的消费者组

public class TestConsumerEarliest {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.123.124:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 指定消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");
        // 设置 auto.offset.reset
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        String topic = "TestOffsetResetTopic";
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singletonList(topic));

        // 消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }

    }
}

一运行发现已有的10条消息（最开始5条加上面一次测试又发了5条，一共10条）是可以被消费到的，且消费完后，对于这个topic就已经有了​​group2​​这个组的​​offset​​了，无论之后启停，只要​​groupid​​不变，都会从最新的​​offset​​往后开始消费

6\. 测试none​​#​​

新建一个测试消费者，设置​​auto.offset.reset​​为​​none​​，注意​​groupid​​为新的​​group3​​，表示对于topic来说是全新的消费者组

public class TestConsumerNone {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.123.124:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 指定消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group3");
        // 设置 auto.offset.reset
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");

        String topic = "TestOffsetResetTopic";
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singletonList(topic));

        // 消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }

    }
}

一运行，程序报错，因为对于topic来说是全新的消费者组，且又指定了​​auto.offset.reset​​为​​none​​，直接抛异常，程序退出

Exception in thread "main" org.apache.kafka.clients.consumer.NoOffsetForPartitionException: Undefined offset with no reset policy for partitions: [TestOffsetResetTopic-0]
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.resetInitializingPositions(SubscriptionState.java:706)
    at org.apache.kafka.clients.consumer.KafkaConsumer.updateFetchPositions(KafkaConsumer.java:2434)
    at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1266)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1231)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211)
    at kakfa.TestConsumerNone.main(TestConsumerNone.java:31)

7\. 总结​​#​​

  • 如果​​topic​​已经有历史消息了，又需要消费这些历史消息，则必须要指定一个从未消费过的​​消费者组​​，同时指定​​auto.offset.reset​​为​​earliest​​，才可以消费到历史数据，之后就有提交​​offset​​。有了​​offset​​，无论是​​earliest​​还是​​latest​​，效果都是一样的了。
  • 如果​​topic​​没有历史消息，或者不需要处理历史消息，那按照默认​​latest​​即可。
举报

相关推荐

kafka之consumer参数auto.offset.reset

vue.jsjavascript前端elementui

kafka复习：（11）auto.offset.reset的默认值

docker容器运维

kafka enable.auto.commit和auto.offset.reset使用说明

cocoamacosobjective-c

Kafka offset管理

kafkakafka和z...

Kafka Offset Storage

大数据javakafkasqlzookeeperSpark

大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset

大数据sparkkafka分布式javaHtml/CSS前端开发

Kafka中 topic、partition、offset、replica、ISR等概念详解

kafka分布式java

kafka-consumer-offset位移

kafka

Kafka：Consumer手动提交offset

kafkazookeeperjava

3、Kafka图形工具-Offset Explorer (Kafka Tool)

Kafka图形工具KafkaTool运维yyds干货盘点
0 条评论