Multiprocess queue blocked?

[Diego-Hernandez-Moodys]

I'm trying to continuously consume messages and send them in batch to a client, but the batch upload may take longer than it takes to receive messages and we're expecting a throughput of 600K+ messages per day. I'm looking to use a multiprocessing queue to avoid losing information as opposed to avoiding duplicates, which is why I don't auto-commit. If that's bad practice, please let me know.

It looks like I'm always getting None values even though I receive messages just fine if I have this all in a single for-loop. This also works well if I use the kafka-python module, but that's because the consumer works like an iterable, but I'd rather use this module for speed.

"""
Continuously consume and process while consuming
"""
import confluent_kafka
import json
import logging
import logging.config
import multiprocessing
import queue
import sys
import time

# in-house
import config
import _deserialize

logging.basicConfig(level = logging.DEBUG)

def get_messages(instance_queue):
    logging.info("Calling consumer")
    while True:
        message = instance_consumer.poll(1.0)

        if message is None:
            continue
        if message.value() is None:
            continue
        if message.error():
            logging.exception(message.error(), exc_info=True)
            continue
        logging.debug(message)

        message_decoded = _deserialize.deserialize_json(message.value())
        if message_decoded is None:
            continue
        logging.debug(f"Topic Name={message.topic()}, Message={message_decoded['id']}")

        instance_queue.put(message_decoded)

def process_messages(instance_queue):
    """ Send to outbound client """
    while True:
        try:
            message = instance_queue.get(timeout=1)
        except queue.Empty:
            pass
        else:
           # do stuff with messages

# Run Kakfa =======================================================================================

def callback_error(data):
    logging.error(data)

time.sleep(3) # server has time to run
instance_consumer = confluent_kafka.Consumer({
    'bootstrap.servers': ",".join(config.dict_setup_aws['servers']),
    'auto.offset.reset': 'earliest',
    'group.id' : 'Client-Consumer',
    'error_cb': callback_error,
    'enable.auto.commit': False
    # 'auto.commit.interval.ms' : 30 * 1000, # adjust later
})
instance_consumer.subscribe(config.dict_setup_aws['topics'])

q = multiprocessing.Queue()

process_consume = multiprocessing.Process(target=get_messages, args=[q])
process_consume.start()

process_upload = multiprocessing.Process(target=process_messages, args=[q])
process_upload.start()

process_consume.join()
process_upload.join()

instance_consumer.close()
sys.exit()

The logs just hang at "Calling consumer"

[edenhill]

Client instances can't be used across forks, and from a superficial glance it looks like multiprocessing will call get_messages from multiple forked sub-processes.
I would recommend polling messages in the main process and passing their key,payload,topic,partition,offset to the child processes.

[Diego-Hernandez-Moodys]

That makes sense. I put the process_messages in it's own thread and populate a regular queue.Queue() instance in the main thread -- this works now. I'm pretty new to multi-threading, so thanks for your time.