Подтвердить что ты не робот

Почему потребителю Кафки требуется много времени, чтобы начать потреблять?

Мы начинаем потребителя Kafka, слушая тему, которая еще не может быть создана (хотя автоматическое создание темы включено).

Вскоре после этого продюсер публикует сообщения на эту тему.

Однако потребителю требуется некоторое время, чтобы заметить это: 5 минут, если быть точным. В этот момент потребитель отменяет свои разделы и снова присоединяется к группе потребителей. Кафка вновь стабилизирует группу. Если посмотреть на временные метки журналов "потребитель против кафки", то этот процесс инициируется на стороне потребителя.

Я предполагаю, что это ожидаемое поведение, но я хотел бы понять это. Это на самом деле происходит перебалансировка (от 0 до 1 раздела)? Если бы мы создавали темы заранее, этого бы не случилось?

2017-02-01 08:36:45.692  INFO 7 --- [afka-consumer-1] o.a.k.c.c.internals.ConsumerCoordinator  : Revoking previously assigned partitions [] for group tps-kafka-partitioning
2017-02-01 08:36:45.692  INFO 7 --- [afka-consumer-1] o.s.k.l.KafkaMessageListenerContainer    : partitions revoked:[]
2017-02-01 08:36:45.693  INFO 7 --- [afka-consumer-1] o.a.k.c.c.internals.AbstractCoordinator  : (Re-)joining group tps-kafka-partitioning
2017-02-01 08:36:45.738  INFO 7 --- [afka-consumer-1] o.a.k.c.c.internals.AbstractCoordinator  : Successfully joined group tps-kafka-partitioning with generation 1
2017-02-01 08:36:45.747  INFO 7 --- [afka-consumer-1] o.a.k.c.c.internals.ConsumerCoordinator  : Setting newly assigned partitions [] for group tps-kafka-partitioning
2017-02-01 08:36:45.749  INFO 7 --- [afka-consumer-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned:[]
2017-02-01 08:41:45.540  INFO 7 --- [afka-consumer-1] o.a.k.c.c.internals.ConsumerCoordinator  : Revoking previously assigned partitions [] for group tps-kafka-partitioning
2017-02-01 08:41:45.544  INFO 7 --- [afka-consumer-1] o.s.k.l.KafkaMessageListenerContainer    : partitions revoked:[]
2017-02-01 08:41:45.544  INFO 7 --- [afka-consumer-1] o.a.k.c.c.internals.AbstractCoordinator  : (Re-)joining group tps-kafka-partitioning

бревна кафки

[2017-02-01 08:41:45,546] INFO [GroupCoordinator 1001]: Preparing to restabilize group tps-kafka-partitioning with old generation 1 (kafka.coordinator.GroupCoordinator)
[2017-02-01 08:41:45,546] INFO [GroupCoordinator 1001]: Stabilized group tps-kafka-partitioning generation 2 (kafka.coordinator.GroupCoordinator)
[2017-02-01 08:41:45,551] INFO [GroupCoordinator 1001]: Assignment received from leader for group tps-kafka-partitioning for generation 2 (kafka.coordinator.GroupCoordinator)
[2017-02-01 08:42:14,636] INFO [GroupCoordinator 1001]: Preparing to restabilize group tps-kafka-group-id with old generation 1 (kafka.coordinator.GroupCoordinator)
[2017-02-01 08:42:14,636] INFO [GroupCoordinator 1001]: Stabilized group tps-kafka-group-id generation 2 (kafka.coordinator.GroupCoordinator)
4b9b3361

Ответ 1

Это, вероятно, связано с значением по умолчанию параметра metadata.max.age.ms, который контролирует, как часто пользователь заставляет обновлять метаданные для темы.

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

Если вы запустите своего потребителя, например, 1000 мс, вы должны увидеть гораздо более короткую задержку до тех пор, пока не будут использованы сообщения.

Кроме того, если вы создаете темы вперед или запускаете производителя перед потребителем, это поведение не должно происходить вообще.