Kafka系列之消息重新消费

概述

需求来源，在review前人留下的屎山代码时发现如下截图所示的代码片段：

也就是说代码是空实现的。另外，从类名定义也知道需求未实现。

于是有此需求：已经消费过的消息重新消费。

调研

调研下来，主要有以下3种可能性方案

实现方案

1. 修改偏移量，即offset，可通过脚本快速实现
2. 新增Group，这将使 Kafka 认为您正在使用一个新的消费者组，并从起始偏移量开始重新消费消息。需通过代码实现
3. 消息产生者重新发送消息，实际业务中不太常见，也不现实

Kafka的偏移量的保存方式，根据不同版本号有3种方式：保存在zookeeper中、保存在kafka的自带_consumer_offset这个topic中、保存在自定义的存储系统中。

版本

首先需要知道Kafka什么版本，找到Kafka的安装目录。可以通过find命令：find / -name '*kafka*'。

根据命令输出得知安装目录为：/usr/local/kafka，进入到libs目录，发现很多kafka_2.11-2.2.0.*文件。其中，2.11为scala版本，2.2.0为kafka版本。

脚本

使用Kafka自带的bin目录下的kafka-consumer-groups.sh脚本设置消费者组（consumer group）的位移， 这是0.11.0.0版本提供的新功能且只适用于新版本consumer。在此版本之前，如果要为已有的consumer group调整位移必须要手动编写Java程序调用KafkaConsumer.seek()方法。

使用此脚本可修改consumer group的位移，有个前提：consumer group必须是inactive的，即不能是处于正在工作中的状态。

格式：

./kafka-consumer-groups.sh --bootstrap-server ip:9092 --group <> --topic --reset-offsets

如：./kafka-consumer-groups.sh --bootstrap-server 172.100.200.200:9092 --group collect_data_business_service --topic topic_event:0 --reset-offsets --to-offset 195725 --execute

topic：指定topic的作用域

• --all-topics：为consumer group下所有topic的所有分区调整位移
• --topic t1 --topic t2：为指定的若干个topic的所有分区调整位移
• --topic t1:0,1,2：为指定的topic的某个分区调整位移

  reset-offsets：确定位移重设策略

  • --to-earliest：把位移调整到分区当前最小位移
  • --to-latest：把位移调整到分区当前最新位移
  • --to-current：把位移调整到分区当前位移
  • --to-offset ：把位移调整到指定位移处
  • --shift-by N：把位移调整到当前位移 + N处，N为负数表示向前移动
  • --to-datetime ：把位移调整到大于给定时间的最早位移处，datetime格式yyyy-MM-ddTHH:mm:ss.xxx
  • --by-duration ：把位移调整到距离当前时间指定间隔的位移处，duration格式是PnDTnHnMnS，如PT0H5M0S
  • --from-file ：从CSV文件中读取调整策略

    常见报错：

    • Error: Executing consumer group command failed due to org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.

      原因：命令指定的IP有误，或端口不对，或未开放等原因导致连接超时

    • Error: Assignments can only be reset if the group ‘collect_data_business_service’ is inactive, but the current state is Stable.

      原因：在修改offset时，需要停止消费者应用程序，如杀掉Java进程，停止container容器等

    • WARN New offset (0) is lower than earliest offset for topic partition topic_event-0. Value will be set to 195551 (kafka.admin.ConsumerGroupCommand$)

      原因：

      代码

      也可以通过代码的方式，指定一个新的消费者组，即group.id。

      @Override
      public void afterPropertiesSet() throws Exception {
          Properties p = new Properties();
          p.setProperty("bootstrap.servers", "172.100.200.200:9092");
          p.setProperty("group.id", "collect_data_business_service");
          p.setProperty("auto.offset.reset", "earliest");
          p.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          p.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          KafkaConsumer consumer = new KafkaConsumer<>(p);
          consumer.subscribe(Collections.singleton("topic_event"));
          consumer.poll(Duration.ofSeconds(1));
          consumer.seekToBeginning(consumer.assignment());
          while (true) {
              ConsumerRecords message = consumer.poll(Duration.ofSeconds(1));
              log.info("topic_event---collect_data_business_service--消费消息:" + message);
              CollectDataV2Content content = JsonUtil.jsonToBean(message.toString(), CollectDataV2Content.class);
              service.saveContent(content);
          }
      }
      

      配置：

      • auto.offset.reset：只能配置以下3种情况：latest, earliest, none，Spring-Kafka应用启动时会检查此项配置，不正确的配置会报错，应用启动失败
      • key.deserializer：必须配置，否则报错：Caused by: org.apache.kafka.common.config.ConfigException: Missing required configuration "key.deserializer" which has no default value.，应用启动失败。
      • value.deserializer：同上

        参考