kafka在重新平衡后停止使用来自新分配分区的消息

cu6pst1q  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(345)

我对Kafka（也对英语……）相当陌生，我面对这个问题，无法谷歌任何解决方案。
我使用spring-boot，spring-kafka支持，我已经在本地机器上安装了kafka\u2.11-0.10.1.1（只有一个代理0）
s1.然后我创建主题

  1. bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 5 --topic tracking

我的使用者配置：applications.properties：

  1. kafka.servers.bootstrap=localhost:9092 
  2. kafka.topic.tracking=tracking
  3. kafka.group.id=trackingGroup
  4. kafka.client.id=client-1

s2。然后我通过更改“kafka.client.id”并运行spring boot主类来启动3个消费者。在eclipse控制台上，我可以检查分区分配：

  1. client-1: partitions assigned:[tracking-4, tracking-3]
  2. client-2: partitions assigned:[tracking-2, tracking-1]
  3. client-3: partitions assigned:[tracking-0]

第三。启动pruducer向主题发送20条消息，每条消息开始消耗特定分区的消息
s4。我关闭消耗1，Kafka自动重新平衡，新分区分配：

  1. client-1: partitions assigned:[]
  2. client-2: partitions assigned:[tracking-2,tracking-1, tracking-0]
  3. client-3: partitions assigned:[tracking-4,tracking-3]

s5。我发现分区'tracking-3'上的消息没有被消耗！！
问题每次都可以被复制，在新分配的分区中丢失一些消息，你能有什么建议吗？请帮帮我，谢谢

springapache-kafkaspring-kafka

1条答案

按热度按时间
lyfkaqu1

lyfkaqu11#

我复制了它；这看起来像是Kafka本身的问题 auto.comit.enabled=true )关于重新平衡，Kafka正在报告未读分区的“位置”( the offset of the <i>next record</i> that will be fetched (if a record with that offset exists) )作为分区的结尾。
事实上，当我使用kafka消费者组工具时，未读分区的偏移量已经在“末尾”了。当我用一个消费者运行它时，当它读取第一个分区时，我看到。。。

  1. $ kafka-consumer-groups --bootstrap-server localhost:9092 --describe -group so43405009
  3. TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
  4. tracking                       0          37              40              3          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1
  5. tracking                       1          40              40              0          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1
  6. tracking                       2          40              40              0          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1
  7. tracking                       3          40              40              0          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1
  8. tracking                       4          40              40              0          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1

请注意当前的\u offset列。
在下一次运行中，我运行了两次，一次是在处理第一个分区时，一次是在稍后运行。。。

  1. TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
  2. tracking                       0          41              44              3          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1
  3. tracking                       1          44              44              0          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1
  4. tracking                       2          44              44              0          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1
  5. tracking                       3          44              44              0          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1
  6. tracking                       4          44              44              0          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1

和

  1. TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
  2. tracking                       0          44              44              0          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1
  3. tracking                       1          44              44              0          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1
  4. tracking                       2          41              44              3          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1
  5. tracking                       3          44              44              0          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1
  6. tracking                       4          44              44              0          client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8      /10.0.0.6                      client1

看看分区2的当前偏移量是如何从44降到41的。
禁用自动提交为我解决了它。。。

  1. spring.kafka.consumer.enable-auto-commit=false

...

  1. TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
  2. tracking                       0          52              52              0          client1-59413599-81e8-49dd-bbd7-8a62152f11e5      /10.0.0.6                      client1
  3. tracking                       1          49              52              3          client1-59413599-81e8-49dd-bbd7-8a62152f11e5      /10.0.0.6                      client1
  4. tracking                       2          49              52              3          client2-edfe34f9-08d5-4825-80d0-4a6cf9526e42      /10.0.0.6                      client2
  5. tracking                       3          48              52              4          client2-edfe34f9-08d5-4825-80d0-4a6cf9526e42      /10.0.0.6                      client2
  6. tracking                       4          51              52              1          client3-20da8742-af38-403e-b125-5d0c7c771319      /10.0.0.6                      client3

这是我的测试程序：

  1. @SpringBootApplication
  2. public class So43405009Application implements CommandLineRunner {
  4.     public static void main(String[] args) {
  5.         SpringApplication.run(So43405009Application.class, args);
  6.     }
  8.     @Autowired
  9.     private KafkaTemplate<String, String> template;
  11.     @Value("${spring.kafka.consumer.client-id}")
  12.     private String clientId;
  14.     @Override
  15.     public void run(String... args) throws Exception {
  16.         if (this.clientId.endsWith("1")) {
  17.             for (int i = 0; i < 20; i++) {
  18.                 this.template.sendDefault("foo" + i);
  19.             }
  20.         }
  21.     }
  23.     @Bean
  24.     public KafkaMessageListenerContainer<String, String> container(ConsumerFactory<String, String> cf) {
  25.         ContainerProperties containerProperties = new ContainerProperties("tracking");
  26.         containerProperties.setMessageListener((MessageListener<?, ?>) d -> {
  27.             System.out.println(d);
  28.             try {
  29.                 Thread.sleep(5_000);
  30.             }
  31.             catch (InterruptedException e) {
  32.                 Thread.currentThread().interrupt();
  33.             }
  34.         });
  35.         KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(cf,
  36.                 containerProperties);
  37.         return container;
  38.     }
  40. }

具有属性

  1. spring.kafka.listener.ack-mode=record
  2. spring.kafka.consumer.enable-auto-commit=false
  3. spring.kafka.consumer.auto-offset-reset=earliest
  4. spring.kafka.consumer.group-id=so43405009
  5. spring.kafka.consumer.client-id=client1
  6. spring.kafka.template.default-topic=tracking

我在0.10.2.0中也看到了相同的结果。
编辑
原来是SpringKafka虫；它在启用自动提交的情况下工作，但是必须显式地启用它

  1. spring.kafka.consumer.enable-auto-commit=true

否则容器会假定 false 并导致上述奇怪的行为-看起来客户端不喜欢在启用自动提交的情况下调用使用者的提交方法#288
我通常会建议设置为false，并选择容器的 AckMode 而是s；例如。 RECORD 记录下来之后， BATCH 在轮询收到的每个批之后（默认）。

展开查看全部
赞(0）分享  回复(0）举报  2021-06-07
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com