test-kafka-consumer.py

import logging, logging.config

TOPIC_NAME = 'TOPIC-NAME'
GROUP_NAME = 'GROUP-NAME'
BOOTSTRAP_SERVERS = ['BROKER-HOST']


def setup_logging():
    logging.config.dictConfig({
        'version': 1,
        'formatters': {
            'verbose': {
                'format': '[%(asctime)s] %(levelname)-8s %(module)s %(process)d %(thread)d %(message)s'
            },
            'console': {
                'format': '[%(asctime)s] %(levelname)-8s %(message)s',
            },
        },
        'handlers': {
            'console': {
                'level': 'DEBUG',
                'class': 'logging.StreamHandler',
                'formatter': 'verbose'
            },
        },
        'root': {
            'level': 'DEBUG',
            'handlers': ['console'],
        },
    })


def main():
    from kafka import KafkaConsumer
    log = logging.getLogger(__name__)

    consumer = KafkaConsumer(
        TOPIC_NAME,
        bootstrap_servers=BOOTSTRAP_SERVERS,
        group_id=GROUP_NAME,
        auto_commit_interval_ms=1000,
    )
    while True:
        try:
            _, messages = consumer.poll(1000).popitem()
        except KeyError:
            log.info('empty result set')
            continue

        for message in messages:
            log.info('received message with offset %d' % message.offset)


if __name__ == '__main__':
    setup_logging()
    main()