多个分区上的Kafka使用者

Question

我的消费者并不是每次都能收到信息。我有3个代理（3个服务器）的Kafka集群，有3个主题和复制因子3的分区。

Topic: my-topic       Partition: 0    Leader: 2       Replicas: 2,1,3 Isr: 3,1,2
Topic: my-topic       Partition: 1    Leader: 3       Replicas: 3,2,1 Isr: 3,2,1
Topic: my-topic       Partition: 2    Leader: 1       Replicas: 1,3,2 Isr: 1,3,2

我有Java中的消费者，我将最大轮询记录设置在50000获取字节上，配置在50MB上。应用程序每分钟都进行轮询。当我向主题“my-topic”发送10条消息时，consumer不会给我所有的消息，而是只给我其中的一部分，其余的将在下一次运行中给我。消息是在applicatin睡眠期间由脚本生成的。你认为轮询方法只给我来自服务器/分区的消息是由分区造成的吗？哪个消息首先响应，其余的消息在下一次运行时才响应？

Map<String, Object> configurations = new HashMap<>();
configurations.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
configurations.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "true");
configurations.put(ConsumerConfig.GROUP_ID_CONFIG, groupId ;
configurations.put(ConsumerConfig.CLIENT_ID_CONFIG, groupId);
configurations.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
configurations.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
configurations.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
configurations.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "52428800");
configurations.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "52428800");
configurations.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "3600000");
configurations.put(JsonDeserializer.TRUSTED_PACKAGES, "my.package.model");
configurations.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "50000");

consumer = new KafkaConsumer<Object, Object>(configurations);
consumer.subscribe(Collections.singletonList("my-topic"));

while(true) {
    ConsumerRecords<Object, Object> records = consumer.poll(Duration.ofMillis(10000));

    if(records.count() > 0) {
        LOGGER.debug("records count: {}", records.count());
        handleMessages(records);
        consumer.commitSync();
    }
    sleep(60000);
}

2021-11-23 08:27:14.851 [DEBUG] --- records count: 3
2021-11-23 08:27:14.853 [DEBUG] --- offset=1175, client=test-27, ts=1637652419417
2021-11-23 08:27:14.857 [DEBUG] --- offset=1176, client=test-28, ts=1637652419418
2021-11-23 08:27:14.860 [DEBUG] --- offset=1177, client=test-29, ts=1637652419418

2021-11-23 08:28:14.924 [DEBUG] --- records count: 7
2021-11-23 08:28:14.925 [DEBUG] --- offset=232304, client=test-20, ts=1637652419406
2021-11-23 08:28:14.929 [DEBUG] --- offset=232305, client=test-21, ts=1637652419407
2021-11-23 08:28:14.933 [DEBUG] --- offset=232306, client=test-24, ts=1637652419411
2021-11-23 08:28:14.937 [DEBUG] --- offset=1141, client=test-22, ts=1637652419408
2021-11-23 08:28:14.941 [DEBUG] --- offset=1142, client=test-23, ts=1637652419410
2021-11-23 08:28:14.944 [DEBUG] --- offset=1143, client=test-25, ts=1637652419414
2021-11-23 08:28:14.949 [DEBUG] --- offset=1144, client=test-26, ts=1637652419415

有谁知道我在配置中做错了什么或遗漏了一些参数，以及如何修复它吗？

谢谢

Answer 1

这是正常的，Kafka不能保证在一个批处理中获得所有消息。
在轮询之间通常不会Hibernate，因为这会导致客户端超时。相反，您依赖于轮询持续时间来防止繁忙的旋转。