Spring boot потребительский параллелизм Kafka не работает должным образом ⇐ JAVA

[Anonymous]

В теме 8 разделов(я проверял)
Topic: local_test TopicId: MYU9FfMBQvuVL4DdPOl_9g PartitionCount: 8 ReplicationFactor: 1 Configs:
Topic: local_test Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: local_test Partition: 1 Leader: 0 Replicas: 0 Isr: 0
Topic: local_test Partition: 2 Leader: 0 Replicas: 0 Isr: 0
Topic: local_test Partition: 3 Leader: 0 Replicas: 0 Isr: 0
Topic: local_test Partition: 4 Leader: 0 Replicas: 0 Isr: 0
Topic: local_test Partition: 5 Leader: 0 Replicas: 0 Isr: 0
Topic: local_test Partition: 6 Leader: 0 Replicas: 0 Isr: 0
Topic: local_test Partition: 7 Leader: 0 Replicas: 0 Isr: 0

Конфигурация потребителя (частично представлена ​​здесь)
@Bean("containerbean")
public ConcurrentKafkaListenerContainerFactory MyListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory factory =
new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(MyConsumerFactory());
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
factory.setConcurrency(3);
return factory;
}


У моего потребителя есть журнал, в котором он записывает сообщение и переходит в режим сна на 10 секунд. Я ожидаю, что журнал появится как минимум 3 раза подряд. Но это происходит каждые 10 секунд. Конечно, существует теоретическая вероятность того, что потоки так и не были запланированы, но я пробовал несколько раз, и каждый раз получалась только 10-секундная метка времени.
Кроме того, перед запуском службы я отправляю несколько сообщений до запуска потребителя.
Код прослушивателя
@KafkaListener(topics = "local_test", groupId = "localgroup", containerFactory = "containerbean")
public void handler(@Payload final MyMessage message, Acknowledgment acknowledgment) throws InterruptedException {
log.info("Received message: {}", message);
Thread.sleep(10_000);
acknowledgment.acknowledge();
}
---


Подробнее здесь: https://stackoverflow.com/questions/791 ... s-expected