#python #apache-kafka
#python #apache-kafka
Вопрос:
У меня есть тема Kafka с 40 разделами. В кластере Kubernetes. У меня также есть микросервис, который использует эту тему.
Иногда в рамках пакетного процесса случается, что в какой-то момент некоторые разделы остаются с необработанными данными, в то время как большинство разделов завершено. Использование kafka-consumer-groups.sh this выглядит следующим образом:
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
- - - - - kafka-python-2.0.1-f1259971-c8ed-4d98-ba37-40f263b14a78/10.44.2.119 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-328f6a97-22ea-4f59-b702-4173feb9f025/10.44.0.29 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-9a2ea04e-3bf1-40f4-9262-6c14d0791dfc/10.44.7.35 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-81f5be15-535c-436c-996e-f8098d0613a1/10.44.4.26 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-ffcf76e2-f0ed-4894-bc70-ee73220881db/10.44.14.2 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-fc5709a0-a0b5-4324-92ff-02b6ee0f1232/10.44.2.123 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-c058418c-51ec-43e2-b666-21971480665b/10.44.15.2 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-0c14afab-af2a-4668-bb3c-015932fbfd13/10.44.14.5 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-1cb308f0-203f-43ae-9252-e0fc98eb87b8/10.44.14.4 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-42753a7f-80d0-481e-93a6-67445cb1bb5e/10.44.14.6 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-63e97395-e1ec-4cab-8edc-c5dd251932af/10.44.2.122 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-7116fdc2-809f-4f99-b5bd-60fbf2aba935/10.44.1.37 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-f5ef8ff1-f09c-498e-9b27-1bcac94b895b/10.44.2.125 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-8feec117-aa3a-42c0-91e8-0ccefac5f134/10.44.2.121 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-45cc5605-d3c8-4c77-8ca8-88afbde81a69/10.44.14.3 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-9a575ac4-1531-4b2a-b516-12ffa2496615/10.44.5.32 kafka-python-2.0.1
- - - - - kafka-python-2.0.1-d33e112b-a1f4-4699-8989-daee03a5021c/10.44.14.7 kafka-python-2.0.1
my-topic 20 890 890 0 - - -
my-topic 38 857 857 0 - - -
my-topic 28 918 918 0 - - -
my-topic 23 66 909 843 - - -
my-topic 10 888 888 0 - - -
my-topic 2 885 885 0 - - -
my-topic 7 853 853 0 - - -
my-topic 16 878 878 0 - - -
my-topic 15 47 901 854 - - -
my-topic 26 934 934 0 - - -
my-topic 32 898 898 0 - - -
my-topic 21 921 921 0 - - -
my-topic 13 933 933 0 - - -
my-topic 5 879 879 0 - - -
my-topic 12 945 945 0 - - -
my-topic 4 918 918 0 - - -
my-topic 29 924 924 0 - - -
my-topic 39 895 895 0 - - -
my-topic 25 30 926 896 - - -
my-topic 9 915 915 0 - - -
my-topic 35 31 890 859 - - -
my-topic 3 69 897 828 - - -
my-topic 1 911 911 0 - - -
my-topic 6 22 901 879 - - -
my-topic 14 41 881 840 - - -
my-topic 30 900 900 0 - - -
my-topic 22 847 847 0 - - -
my-topic 8 919 919 0 - - -
my-topic 0 902 902 0 - - -
my-topic 18 924 924 0 - - -
my-topic 36 864 864 0 - - -
my-topic 34 929 929 0 - - -
my-topic 24 864 864 0 - - -
my-topic 19 937 937 0 - - -
my-topic 27 859 859 0 - - -
my-topic 11 838 838 0 - - -
my-topic 31 49 922 873 - - -
my-topic 37 882 882 0 - - -
my-topic 17 942 942 0 - - -
my-topic 33 928 928 0 - - -
Далее в нем указано, что группа потребителей является rebalancing .
Здесь следует отметить, что в разделе CONSUMER-ID указано меньше потребителей, как и должно быть. Должно быть 20 потребителей, но в этом выводе показано только 17, хотя все модули запущены. Это число варьируется, и я не уверен, является ли это проблемой вывода или их действительно нет. Это также сбивает меня с толку, потому что при первоначальном запуске (все новые развертывания Kafka и consumer) этого не происходит. Так что, похоже, это действительно связано с масштабированием или иным уничтожением пользовательских развертываний.
Затем на короткое время происходит назначение потребителей, и примерно через полминуты та же картина, что и выше, снова показывает, где группа потребителей перебалансируется.
Это происходит и при уменьшении масштаба. Например, когда у меня всего 4 потребителя. Я не уверен, что здесь происходит. Все модули запускаются, и я использую тот же базовый код и шаблон в других микросервисах, где он, кажется, работает нормально.
Я подозреваю, что это как-то связано с уничтожением модуля-потребителя, потому что, как я уже сказал, при новом развертывании он работает изначально. Этот пакет также работает немного дольше, чем другие, которые у меня есть, поэтому во время его выполнения более вероятно уничтожение модуля. Я также не уверен, имеет ли это какое-то отношение к тому, что большинство разделов уже завершено, это также может быть просто причудой моего варианта использования.
Я узнал это, потому что обработка, казалось, длилась вечно, но новые данные все еще обрабатывались. Итак, я думаю, что происходит то, что на короткий момент, когда назначаются потребители, они обрабатывают данные, но они никогда не фиксируют смещение перед перебалансировкой, оставляя их в бесконечном цикле. Единственной слегка связанной вещью, которую я обнаружил, была эта проблема, но она была из нескольких предыдущих версий и не полностью описывает мою ситуацию.
Я использую kafka-python клиент, и я использую изображение kafka confluentinc/cp-kafka:5.0.1 .
Я создаю тему с помощью клиента администратора NewTopic(name='my-topic', num_partitions=40, replication_factor=1) и создаю клиент следующим образом:
consumer = KafkaConsumer(consume_topic,
bootstrap_servers=bootstrap_servers,
group_id=consume_group_id,
value_deserializer=lambda m: json.loads(m))
for message in consumer:
process(message)
Что здесь не так?
У меня какая-то ошибка конфигурации?
Любая помощь приветствуется.
Ответ №1:
Проблема была в конфигурации сердцебиения. Оказывается, что, хотя для обработки большинства сообщений требуется всего несколько секунд, обработка немногих сообщений занимает очень много времени. В этих особых случаях обновление сердцебиения заняло слишком много времени для некоторых потребителей, в результате чего брокер предположил, что потребитель отключен, и начал перебалансировку.
Я предполагаю, что дальше произошло то, что потребители были переназначены на одно и то же сообщение, слишком долго обрабатывая его снова и вызывая еще одну перебалансировку. Что приводит к бесконечному циклу.
Я, наконец, решил это, увеличив оба session_timeout_ms и heartbeat_interval_ms в потребителе (задокументировано здесь ). Я также уменьшил размер пакета, чтобы сердцебиение обновлялось более регулярно.