首页 文章

重新 balancer 后，kafka停止使用来自新分配的分区的消息

提问于
浏览 次
4

我对kafka（也是英语......）很新，我面对这个问题，不能谷歌任何解决方案 .

我使用spring-boot，spring-kafka支持，我在本地机器上安装了kafka_2.11-0.10.1.1（只有一个代理0）

s1.然后我创建主题

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

我的消费者配置：applitions.properties：

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

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

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

S3 . 启动pruducer向主题发送20条消息，每条消息开始消耗特定分区的消息

S4 . 我关闭了消耗1，kafka自动进行重新 balancer ，新分区分配：

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

S5 . 我发现分区'tracking-3'上的消息没有消耗!!

问题可以每次都重现，在新分配的分区中丢失一些消息，你能不能提出任何建议？请帮帮我，谢谢

spring apache-kafka spring-kafka

1 回答

  • 4

    我复制了它;它看起来像kafka本身（使用 auto.comit.enabled=true ）在重新 balancer 上的问题，kafka报告未读分区的"position"（ the offset of the <i>next record</i> that will be fetched (if a record with that offset exists) ）作为分区的结尾 .

    事实上，当我使用kafka-consumer-groups工具时，未读分区的偏移量已经处于“结束”状态 . 当我只用一个消费者运行它时，当它正在读取第一个分区时，我看到......

    $ kafka-consumer-groups --bootstrap-server localhost:9092 --describe -group so43405009

TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
tracking                       0          37              40              3          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1
tracking                       1          40              40              0          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1
tracking                       2          40              40              0          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1
tracking                       3          40              40              0          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1
tracking                       4          40              40              0          client1-8129bb3d-3a83-4c83-9128-3a2762ede758      /10.0.0.6                      client1

    请注意CURRENT_OFFSET列 .

    在下一次运行中，我运行了两次，一次是在处理第一个分区时，稍后再运行一次......

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

    和

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

    看看分区2的当前偏移量从44下降到41 .

    禁用自动提交为我解决了...

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

    ...

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

    这是我的测试程序：

    @SpringBootApplication
public class So43405009Application implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(So43405009Application.class, args);
    }

    @Autowired
    private KafkaTemplate<String, String> template;

    @Value("${spring.kafka.consumer.client-id}")
    private String clientId;

    @Override
    public void run(String... args) throws Exception {
        if (this.clientId.endsWith("1")) {
            for (int i = 0; i < 20; i++) {
                this.template.sendDefault("foo" + i);
            }
        }
    }

    @Bean
    public KafkaMessageListenerContainer<String, String> container(ConsumerFactory<String, String> cf) {
        ContainerProperties containerProperties = new ContainerProperties("tracking");
        containerProperties.setMessageListener((MessageListener<?, ?>) d -> {
            System.out.println(d);
            try {
                Thread.sleep(5_000);
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(cf,
                containerProperties);
        return container;
    }

}

    有 property

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

    我也看到了与0.10.2.0相同的结果 .

    EDIT

    事实证明这是一个 Spring 天的 Kafka 虫;它适用于启用自动提交，但您必须明确启用它

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

    否则容器假设它是 false 并导致上述奇怪的行为 - 如果启用了自动提交，则看起来像客户端没有_3028747的提交方法 . #288 .

    我通常建议设置为false，然后选择容器的 AckMode 之一;例如 RECORD 在记录收到之后提交， BATCH 在轮询收到的每个批次之后（默认） .

    回复于

相关问题