Потребители Kafka не могут отправить биение, но получают сообщения - PullRequest
Новая
       33

Потребители Kafka не могут отправить биение, но получают сообщения

0 голосов
/ 27 мая 2019

Я использую кластер Kafka из Bitnami в Azure (https://bitnami.com/stack/kafka/cloud/azure). Это кластер из одного узла с брокером на одной виртуальной машине и Zookeeper на другой. Я включил порты 9090 и 2181.

У меня есть .Net Core WebApps, работающие в Azure с Confluent .NET Client (v1.0.0) для Kafka. В WebApp настроены производители и потребители.

Проблема, с которой я сталкиваюсь, заключается в том, что вскоре после запуска веб-приложения происходит сбой потребителя из-за истечения пульса . Но пока что он может получать некоторые сообщения от производителя. Даже если я просто запускаю только одного Потребителя, он автоматически завершается ошибкой после SessionTimeoutMs из-за истечения срока действия пульса.

Я не могу понять, почему сердцебиение прекращается (когда сообщения принимаются потребителем). Это происходит на отдельном порту? Если я установлю SessionTimeoutMs = 300000 (макс.), Он истекает ровно через 5 минут.

Журнал сервера (когда я добавляю потребителя в WebApp - обратите внимание, что пульс истекает в течение 16 секунд): 

[2019-05-22 17:57:42,302] INFO [GroupCoordinator 1001]: Preparing to rebalance group test-consumer-group-new2 in state PreparingRebalance with old generation 4 (__consumer_offsets-40) (reason: Adding new member rdkafka-03f8db78-4f74-4cb2-9389-bb949c53281c) (kafka.coordinator.group.GroupCoordinator)
[2019-05-22 17:57:42,303] INFO [GroupCoordinator 1001]: Stabilized group test-consumer-group-new2 generation 5 (__consumer_offsets-40) (kafka.coordinator.group.GroupCoordinator)
[2019-05-22 17:57:42,349] INFO [GroupCoordinator 1001]: Assignment received from leader for group test-consumer-group-new2 for generation 5 (kafka.coordinator.group.GroupCoordinator)
[2019-05-22 17:57:58,403] INFO [GroupCoordinator 1001]: Member rdkafka-03f8db78-4f74-4cb2-9389-bb949c53281c in group test-consumer-group-new2 has failed, removing it from the group (kafka.coordinator.group.GroupCoordinator)
[2019-05-22 17:57:58,403] INFO [GroupCoordinator 1001]: Preparing to rebalance group test-consumer-group-new2 in state PreparingRebalance with old generation 5 (__consumer_offsets-40) (reason: removing member rdkafka-03f8db78-4f74-4cb2-9389-bb949c53281c on heartbeat expiration) (kafka.coordinator.group.GroupCoordinator)
[2019-05-22 17:57:58,404] INFO [GroupCoordinator 1001]: Group test-consumer-group-new2 with generation 6 is now empty (__consumer_offsets-40) (kafka.coordinator.group.GroupCoordinator)

Мой server.properties:

listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://40.X.X.X:9092

Конфигурация потребителя: 

var conf = new ConsumerConfig
{
    GroupId = "test-consumer-group-new2",
    BootstrapServers = "40.X.X.X:9092",
    SecurityProtocol = SecurityProtocol.Plaintext,
    SaslMechanism = SaslMechanism.Plain,
    SaslUsername = "user",
    SaslPassword = "<passowrd>"
}

Я также не понимаю старое / новое Поколения в Кафке и не смог найти документацию по этому же. Кто-нибудь может пролить свет на это? Спасибо.

1 Ответ

0 голосов
/ 27 мая 2019

max.poll.records - возможно, потребуется уменьшить;

обратите внимание на heartbeat.interval.ms - значение должно быть установлено ниже session.timeout.ms, но обычно оно должно быть установлено не выше 1/3 от этого значения

...