Kafka python Время ожидания группового сеанса для потребителя - PullRequest
Новая
       56

Kafka python Время ожидания группового сеанса для потребителя

0 голосов
/ 27 апреля 2020

Я использую confluent-kafka v1.3.0 и у меня возникла следующая проблема с тайм-аутом сеанса в группе потребителей. Мой конфиг выглядит следующим образом: 

c['KAFKA'] = {
    'bootstrap.servers': 'host.docker.internal:9104',
    'consumer': {
        'group.id': 'consumer',
        'enable.auto.commit': True,
        'default.topic.config': {
            'auto.offset.reset': 'earliest
        },
        'heartbeat.interval.ms': 100000,
        'max.poll.interval.ms': 300000,
        'session.timeout.ms': 100000
    },
}

и logi c в коде, подобном: 

consumer.subscribe('database_changes')

with ThreadPoolExecutor(max_workers=500) as executor:
    while True:
        msg = consumer.poll(100)
        if msg is not None:
            executor.submit(process_message, msg)

Код в сообщении процесса обработки ждет несколько мс, потому что это действительно простая логика c. Все работает хорошо, но каждый раз, когда я получаю эту ошибку: 

{"asctime":"2020-04-27 08:42:25,759","levelname":"WARNING","name":"services.kafka","message":"SESSTMOUT [rdkafka#consumer-2] [thrd:main]: Consumer group session timed out (in join-state started) after 30131 ms without a successful response from the group coordinator (broker 0, last error was Success): revoking assignment and rejoining group"}

, эта перебалансировка сильно затрудняет весь процесс.

Кто-нибудь имеет представление о том, что может быть неправильно установлено? Я подозреваю, что сердцебиение не работает, но я не знаю, как это проверить или лучше исправить.

Спасибо

...