docs: def _partition(self):
          """Consume messages from kafka
  
             Consume messages from kafka using the Kazoo SetPartitioner to
             allow multiple consumer processes to negotiate access to the kafka
             partitions
          """
  
          # KazooClient and SetPartitioner objects need to be instantiated after
          # the consumer process has forked.  Instantiating prior to forking
          # gives the appearance that things are working but after forking the
          # connection to zookeeper is lost and no state changes are visible
  
          if not self._kazoo_client:
              self._kazoo_client = KazooClient(hosts=self._zookeeper_url)
              self._kazoo_client.start()
  
              state_change_event = threading.Event()
  
              self._set_partitioner = (
                  SetPartitioner(self._kazoo_client,
                                 path=self._zookeeper_path,
                                 set=self._consumer.fetch_offsets.keys(),
                                 state_change_event=state_change_event,
                                 identifier=str(datetime.datetime.now())))
  
          try:
              while 1:
                  if self._set_partitioner.failed:
                      raise Exception("Failed to acquire partition")
  
                  elif self._set_partitioner.release:
                      log.info("Releasing locks on partition set {} "
                               "for topic {}".format(self._partitions,
                                                     self._kafka_topic))
                      self._set_partitioner.release_set()
  
                      self._partitions = []
  
                  elif self._set_partitioner.acquired:
                      if not self._partitions:
                          self._partitions = [p for p in self._set_partitioner]
  
                          if not self._partitions:
                              log.info("Not assigned any partitions on topic {},"
                                       " waiting for a Partitioner state change"
                                       .format(self._kafka_topic))
                              state_change_event.wait()
                              state_change_event.clear()
                              continue
  
                          log.info("Acquired locks on partition set {} "
                                   "for topic {}".format(self._partitions, self._kafka_topic))
  
                          # Reconstruct the kafka consumer object because the
                          # consumer has no API that allows the set of partitons
                          # to be updated outside of construction.
                          self._consumer.stop()
                          self._consumer = self._create_kafka_consumer(self._partitions)
                          return
  
                  elif self._set_partitioner.allocating:
                      log.info("Waiting to acquire locks on partition set")
                      self._set_partitioner.wait_for_acquire()
  
          except Exception:
              log.exception('KafkaConsumer encountered fatal exception '
                            'processing messages.')
              raise