def _mp_consume(client, group, topic, queue, size, events, **consumer_options):
<EOL>interval = <NUM_LIT:1><EOL>while not events.exit.is_set():<EOL><INDENT>try:<EOL><INDENT>client.reinit()<EOL>consumer = SimpleConsumer(client, group, topic,<EOL>auto_commit=False,<EOL>auto_commit_every_n=None,<EOL>auto_commit_every_t=None,<EOL>**consumer_options)<EOL>consumer.provide_partition_info()<EOL>while True:<EOL><INDENT>events.start.wait()<EOL>if events.exit.is_set():<EOL><INDENT>break<EOL><DEDENT>count = <NUM_LIT:0><EOL>message = consumer.get_message()<EOL>if message:<EOL><INDENT>while True:<EOL><INDENT>try:<EOL><INDENT>queue.put(message, timeout=FULL_QUEUE_WAIT_TIME_SECONDS)<EOL>break<EOL><DEDENT>except queue.Full:<EOL><INDENT>if events.exit.is_set():<EOL><INDENT>break<EOL><DEDENT><DEDENT><DEDENT>count += <NUM_LIT:1><EOL>if count == size.value:<EOL><INDENT>events.pause.wait()<EOL><DEDENT><DEDENT>else:<EOL><INDENT>time.sleep(NO_MESSAGES_WAIT_TIME_SECONDS)<EOL><DEDENT><DEDENT>consumer.stop()<EOL><DEDENT>except KafkaError as e:<EOL><INDENT>log.error(<EOL>"<STR_LIT>" % (e, interval))<EOL>time.sleep(interval)<EOL>interval = interval * <NUM_LIT:2> if interval * <NUM_LIT:2> < MAX_BACKOFF_SECONDS else MAX_BACKOFF_SECONDS<EOL><DEDENT><DEDENT>
A child process worker which consumes messages based on the
notifications given by the controller process

NOTE: Ideally, this should have been a method inside the Consumer
class. However, multiprocessing module has issues in windows. The
functionality breaks unless this function is kept outside of a class