#java #apache-kafka #apache-flink
#java #apache-kafka #apache-flink
Вопрос:
У меня есть тема kafka с 15 разделами [0-14], и я запускаю flink с 5 параллелизмом. Поэтому в идеале каждый параллельный потребитель flink должен использовать по 3 раздела каждый. Но даже после нескольких перезапусков несколько разделов kafka не подписаны никакими работниками flink.
org.apache.kafka.clients.consumer.KafkaConsumer assign Subscribed to partition(s): topic_name-13, topic_name-8, topic_name-9
org.apache.kafka.clients.consumer.KafkaConsumer assign Subscribed to partition(s): topic_name-11, topic_name-12, topic_name-13
org.apache.kafka.clients.consumer.KafkaConsumer assign Subscribed to partition(s): topic_name-14, topic_name-0, topic_name-10
org.apache.kafka.clients.consumer.KafkaConsumer assign Subscribed to partition(s): topic_name-5, topic_name-6, topic_name-10
org.apache.kafka.clients.consumer.KafkaConsumer assign Subscribed to partition(s): topic_name-2, topic_name-3, topic_name-7
Из приведенных выше журналов видно, что разделы 10 и 13 были подписаны 2 потребителями, а разделы 1 и 4 вообще не подписаны.
Примечание: Если я запускаю задание с 1 параллелизмом, задание работает отлично.
Версия Flink: 1.3.3
Комментарии:
1. Вы уверены, что для всех задач установлен один и тот же идентификатор группы?
2. да, идентификатор группы тот же. Я даже пытался изменить его на новый идентификатор группы, но это тоже не помогло
Ответ №1:
Это звучит как https://issues.apache.org/jira/browse/FLINK-7143.
Ознакомление с деталями в билете Jira и в запросе на извлечение (https://github.com/apache/flink/pull/4301 ), похоже, что если вы используете Flink 1.3.x, вы можете извлечь выгоду из этого исправления ошибки только при повторном перезапуске. Перезапуска с точки сохранения недостаточно, чтобы извлечь выгоду из исправления.