more consumer than partition count in kafka. some older consumer will be replaced, and some new consumer cant consume from topic.

kafka-more-consumer-than-partition.py

# -*- coding:utf8 -*-
"""
Author : Myth
Date   : 2018/5/9
Email  : email4myth at gmail.com
"""

import json
import os
import signal
import time
from collections import OrderedDict
from multiprocessing import Process, Queue

from confluent_kafka import Consumer
from datetime import datetime

bro_group_id = 'myth.worker.count.test'
queue = Queue(100)

consumers = OrderedDict()

'''
{
    pid: {
        'create_time': TIME,
        'last_update': TIME
    }
}
'''


def consumer(queue):
    c = Consumer({'group.id': bro_group_id, 'enable.auto.commit': True, 'bootstrap.servers': '10.0.81.9:9091'})
    c.subscribe(['test-topic'])
    while True:
        msg = c.poll(3)
        if not msg:
            continue
        if not msg.error():
            v = msg.value()
            queue.put(json.dumps({'pid': str(os.getpid()), 'ts': time.time()}))


def handle_signal(s, f):
    if s == signal.SIGHUP:
        start_new_consumer()
    else:
        print 'stoping all sub processes ...'
        for pid in consumers:
            pid_info = consumers[pid]
            p = pid_info['p']
            p.terminate()


def check_consumers():
    while True:

        # update lost update
        while not queue.empty():
            last_info = json.loads(queue.get())
            pid = last_info['pid']
            ts = last_info['ts']
            consumers[pid]['last_update'] = datetime.fromtimestamp(ts)

        os.system('clear')
        pids = consumers.keys()
        print ' pid  |   ctime  |   utime  | duration       (%s consumers)' % len(pids)
        print '-----------------------------------------------------------'
        for pid in pids:
            consumer_info = consumers[pid]
            now = datetime.now()
            create_time = consumer_info['create_time']
            last_update = consumer_info['last_update']
            if not last_update:
                last_update_str = '<notime>'
                duration = 0
            else:
                duration = (now - last_update).total_seconds()
                last_update_str = last_update.strftime('%H:%M:%S')

            print '%s | %s | %s | %s %s' % (pid.zfill(5), create_time.strftime('%H:%M:%S'), last_update_str, duration,
                                   '<<<<<<<<<<<<<<<<<<<' if duration > 30 else '')

        time.sleep(3)


def start_new_consumer():
    p = Process(target=consumer, args=(queue,))
    p.start()
    consumers[str(p.pid)] = {
        'p': p,
        'create_time': datetime.now(),
        'last_update': None
    }


def main():
    for i in range(30):
        start_new_consumer()

    check_consumers()


if __name__ == '__main__':
    signal.signal(signal.SIGINT, handle_signal)
    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGHUP, handle_signal)
    main()

result.txt

 pid  |   ctime  |   utime  | duration       (60 consumers)
-----------------------------------------------------------
25817 | 16:36:08 | 16:56:19 | 0.003047
25818 | 16:36:08 | 16:56:19 | 0.002946
25819 | 16:36:08 | 16:40:11 | 968.385806 <<<<<<<<<<<<<<<<<<<
25820 | 16:36:08 | 16:56:19 | 0.633428
25821 | 16:36:08 | 16:52:10 | 249.464819 <<<<<<<<<<<<<<<<<<<
25826 | 16:36:08 | 16:52:06 | 253.638164 <<<<<<<<<<<<<<<<<<<
25831 | 16:36:08 | 16:52:10 | 249.465542 <<<<<<<<<<<<<<<<<<<
25833 | 16:36:08 | 16:44:11 | 728.586204 <<<<<<<<<<<<<<<<<<<
25837 | 16:36:08 | 16:52:09 | 250.651227 <<<<<<<<<<<<<<<<<<<
25841 | 16:36:08 | 16:37:27 | 1132.652902 <<<<<<<<<<<<<<<<<<<
25845 | 16:36:08 | 16:56:18 | 1.649614
25852 | 16:36:08 | 16:44:11 | 728.644092 <<<<<<<<<<<<<<<<<<<
25854 | 16:36:08 | 16:44:11 | 728.587277 <<<<<<<<<<<<<<<<<<<
25857 | 16:36:08 | 16:37:26 | 1133.111523 <<<<<<<<<<<<<<<<<<<
25862 | 16:36:08 | 16:56:19 | 0.00315
25866 | 16:36:08 | 16:36:40 | 1179.650479 <<<<<<<<<<<<<<<<<<<
25872 | 16:36:08 | 16:56:19 | 0.003286
25876 | 16:36:08 | 16:56:19 | 0.635288
25879 | 16:36:08 | 16:56:19 | 0.002431
25884 | 16:36:08 | 16:37:27 | 1132.645133 <<<<<<<<<<<<<<<<<<<
25888 | 16:36:08 | 16:44:11 | 728.586583 <<<<<<<<<<<<<<<<<<<
25893 | 16:36:08 | 16:56:19 | 0.002418
25896 | 16:36:08 | 16:43:45 | 754.642605 <<<<<<<<<<<<<<<<<<<
25902 | 16:36:08 | 16:52:05 | 254.276382 <<<<<<<<<<<<<<<<<<<
25907 | 16:36:08 | 16:56:19 | 0.001862
25909 | 16:36:08 | 16:56:19 | 0.003379
25912 | 16:36:08 | 16:56:18 | 1.111968
25916 | 16:36:08 | 16:44:10 | 729.430122 <<<<<<<<<<<<<<<<<<<
25920 | 16:36:08 | 16:56:19 | 0.003328
25924 | 16:36:08 | 16:37:28 | 1131.64276 <<<<<<<<<<<<<<<<<<<
26151 | 16:36:38 | 16:56:19 | 0.002647
26418 | 16:37:26 | 16:56:19 | 0.002425
26423 | 16:37:27 | 16:52:06 | 253.608338 <<<<<<<<<<<<<<<<<<<
26429 | 16:37:27 | 16:56:19 | 0.002384
26434 | 16:37:28 | 16:52:10 | 249.467808 <<<<<<<<<<<<<<<<<<<
26439 | 16:37:28 | <notime> | 0
27350 | 16:40:11 | 16:56:19 | 0.774119
28476 | 16:43:43 | 16:56:19 | 0.003328
28635 | 16:44:10 | 16:56:19 | 0.003103
28641 | 16:44:10 | <notime> | 0
28647 | 16:44:11 | 16:52:09 | 250.715207 <<<<<<<<<<<<<<<<<<<
28652 | 16:44:11 | 16:52:09 | 250.717509 <<<<<<<<<<<<<<<<<<<
28659 | 16:44:12 | 16:56:19 | 0.003899
28665 | 16:44:12 | 16:52:06 | 253.472814 <<<<<<<<<<<<<<<<<<<
28670 | 16:44:12 | <notime> | 0
31199 | 16:52:05 | <notime> | 0
31206 | 16:52:05 | 16:56:19 | 0.003503
31213 | 16:52:06 | 16:56:19 | 0.002838
31218 | 16:52:06 | 16:56:19 | 0.004075
31225 | 16:52:07 | 16:56:18 | 1.734051
31233 | 16:52:07 | 16:56:19 | 0.002824
31238 | 16:52:08 | 16:56:19 | 0.663096
31249 | 16:52:08 | 16:56:19 | 0.003649
31254 | 16:52:08 | <notime> | 0
31259 | 16:52:09 | <notime> | 0
31264 | 16:52:09 | 16:52:10 | 249.463612 <<<<<<<<<<<<<<<<<<<
31269 | 16:52:09 | 16:56:19 | 0.00364
31275 | 16:52:09 | 16:56:19 | 0.003229
31307 | 16:52:10 | 16:56:19 | 0.004104
31349 | 16:52:10 | <notime> | 0