def METHOD_NAME(i):
    assert i not in consumers
    assert i not in stop
    stop[i] = threading.Event()
    consumers[i] = KafkaConsumer(topic,
                                 bootstrap_servers=connect_str,
                                 group_id=group_id,
                                 heartbeat_interval_ms=500)
    while not stop[i].is_set():
        for tp, records in six.itervalues(consumers[i].poll(100)):
            messages[i][tp].extend(records)
    consumers[i].close()
    consumers[i] = None
    stop[i] = None