def _partition(self):
<EOL>if not self._kazoo_client:<EOL><INDENT>self._kazoo_client = KazooClient(hosts=self._zookeeper_url)<EOL>self._kazoo_client.start()<EOL>state_change_event = threading.Event()<EOL>self._set_partitioner = (<EOL>SetPartitioner(self._kazoo_client,<EOL>path=self._zookeeper_path,<EOL>set=self._consumer.fetch_offsets.keys(),<EOL>state_change_event=state_change_event,<EOL>identifier=str(datetime.datetime.now())))<EOL><DEDENT>try:<EOL><INDENT>while <NUM_LIT:1>:<EOL><INDENT>if self._set_partitioner.failed:<EOL><INDENT>raise Exception("<STR_LIT>")<EOL><DEDENT>elif self._set_partitioner.release:<EOL><INDENT>log.info("<STR_LIT>"<EOL>"<STR_LIT>".format(self._partitions,<EOL>self._kafka_topic))<EOL>self._set_partitioner.release_set()<EOL>self._partitions = []<EOL><DEDENT>elif self._set_partitioner.acquired:<EOL><INDENT>if not self._partitions:<EOL><INDENT>self._partitions = [p for p in self._set_partitioner]<EOL>if not self._partitions:<EOL><INDENT>log.info("<STR_LIT>"<EOL>"<STR_LIT>"<EOL>.format(self._kafka_topic))<EOL>state_change_event.wait()<EOL>state_change_event.clear()<EOL>continue<EOL><DEDENT>log.info("<STR_LIT>"<EOL>"<STR_LIT>".format(self._partitions, self._kafka_topic))<EOL>self._consumer.stop()<EOL>self._consumer = self._create_kafka_consumer(self._partitions)<EOL>return<EOL><DEDENT><DEDENT>elif self._set_partitioner.allocating:<EOL><INDENT>log.info("<STR_LIT>")<EOL>self._set_partitioner.wait_for_acquire()<EOL><DEDENT><DEDENT><DEDENT>except Exception:<EOL><INDENT>log.exception('<STR_LIT>'<EOL>'<STR_LIT>')<EOL>raise<EOL><DEDENT>
Consume messages from kafka

           Consume messages from kafka using the Kazoo SetPartitioner to
           allow multiple consumer processes to negotiate access to the kafka
           partitions