def rebalance(self):
        """Figure out which brokers and partitions we should be consuming from,
        based on the latest information about the other consumers and brokers
        that are present.

        We registered for notifications from ZooKeeper whenever a broker or 
        consumer enters or leaves the pool. But we usually only rebalance right
        before we're about to take an action like fetching.

        The rebalancing algorithm is slightly different from that described in
        the design doc (mostly in the sense that the design doc algorithm will
        leave partitions unassigned if there's an uneven distributions). The 
        idea is that we split the partitions as evently as possible, and if
        some consumers need to have more partitions than others, the extra 
        partitions always go to the earlier consumers in the list. So you could
        have a distribution like 4-4-4-4 or 5-5-4-4, but never 4-4-4-5.

        Rebalancing has special consequences if the Consumer is doing manual 
        commits (autocommit=False):

        1. This Consumer will keep using the in memory offset state for all 
           BrokerPartitions that it was already following before the rebalance.
        2. The offset state for any new BrokerPartitions that this Consumer is
           responsible for after the rebalance will be read from ZooKeeper.
        3. For those BrokerPartitions that this Consumer was reading but is no
           longer responsible for after the rebalance, the offset state is 
           simply discarded. It is not persisted to ZooKeeper.
        
        So there is no guarantee of single delivery in this circumstance. If 
        BrokerPartition 1-0 shifts ownership from Consumer A to Consumer B in 
        the rebalance, Consumer B will pick up from the last manual commit of 
        Consumer A -- *not* the offset that Consumer A was at when the rebalance
        was triggered.
        """
        log.info(("Rebalance triggered for Consumer {0}, broker partitions " + \
                  "before rebalance: {1}").format(self.id, unicode(self)))

        if not self._rebalance_enabled:
            log.info("Rebalancing disabled -- ignoring rebalance request")
            return 

        # Get all the consumer_ids in our consumer_group who are listening to 
        # this topic (this includes us).
        all_topic_consumers = self._zk_util.consumer_ids_for(self.topic, 
                                                             self.consumer_group)
        # Where do I rank in the consumer_group list?
        all_broker_partitions = self._zk_util.broker_partitions_for(self.topic)
        try:
            my_index = all_topic_consumers.index(self.id)
        except ValueError:
            msg_tmpl = "This consumer ({0}) not found list of consumers " +\
                       "for this topic {1}: {2}"
            raise ConsumerEntryNotFoundError(
                msg_tmpl.format(self.id, self.topic, all_topic_consumers))

        bp_per_consumer = len(all_broker_partitions) / len(all_topic_consumers)
        consumers_with_extra = range(len(all_broker_partitions) % len(all_topic_consumers))

        # If the current consumer is among those that have an extra partition...
        num_parts = bp_per_consumer + (1 if my_index in consumers_with_extra else 0)

        # We're the first index,
        if my_index == 0:
            start = 0
        # All the indexes before us where ones with extra consumers
        elif 1 - my_index in consumers_with_extra:
            start = my_index * (bp_per_consumer + 1)
        # There is 0 or more consumers with extra partitions followed by 
        # 1 or more partitions with no extra partitions
        else:
            start = (len(consumers_with_extra) * (bp_per_consumer + 1)) + \
                    ((my_index - len(consumers_with_extra)) * bp_per_consumer)

        ############## Set our state info... ##############
        self._broker_partitions = all_broker_partitions[start:start+num_parts]

        # We keep a mapping of BrokerPartitions to their offsets. We ditch those
        # BrokerPartitions we are no longer responsible for...
        for bp in self._bps_to_next_offsets.keys():
            if bp not in self._broker_partitions:
                del self._bps_to_next_offsets[bp]

        # We likewise add new BrokerPartitions we're responsible for to our 
        # BP->offset mapping, and set their offsets to None, to indicate that
        # we don't know, and we have to check ZK for them.
        for bp in self._broker_partitions:
            self._bps_to_next_offsets.setdefault(bp, None)

        # This will collapse duplicates so we only have one conn per host/port
        broker_conn_info = frozenset((bp.broker_id, bp.host, bp.port)
                                     for bp in self._broker_partitions)
        self._connections = dict((broker_id, Kafka(host, port))
                                 for broker_id, host, port in broker_conn_info)

        # Register all our callbacks so we know when to do this next
        self._register_callbacks()
        if self._all_callbacks_registered():
            self._needs_rebalance = False
        
        # Report our progress
        log.info("Rebalance finished for Consumer {0}: {1}".format(self.id, unicode(self)))
dataset/ETHPy150Open DataDog/brod/brod/zk.py/ZKConsumer.rebalance/swapped operands of `-`