Select first over loaded broker having victim_partition.

        Note: The broker with maximum siblings of victim-partitions (same topic)
        is selected to reduce topic-partition imbalance.
339735
def _elect_source_broker(self, victim_partition, broker_subset=None):
        
        broker_subset = broker_subset or self._brokers
        over_loaded_brokers = sorted(
            [
                broker
                for broker in broker_subset
                if victim_partition in broker.partitions and not broker.inactive
            ],
            key=lambda b: len(b.partitions),
            reverse=True,
        )
        if not over_loaded_brokers:
            return None

        broker_topic_partition_cnt = [
            (broker, broker.count_partitions(victim_partition.topic))
            for broker in over_loaded_brokers
        ]
        max_count_pair = max(
            broker_topic_partition_cnt,
            key=lambda ele: ele[1],
        )
        return max_count_pair[0]
536917
def index_partition(self, partition, force=False):
        
        self.backend.partition_index.index_one(partition, force=force)
383213
def _partition_species(composition, max_components=2):
        

        def _partition(collection):
            # thanks https://stackoverflow.com/a/30134039

            if len(collection) == 1:
                yield [collection]
                return

            first = collection[0]
            for smaller in _partition(collection[1:]):
                # insert `first` in each of the subpartition's subsets
                for n, subset in enumerate(smaller):
                    yield smaller[:n] + [[first] + subset] + smaller[n + 1:]
                # put `first` in its own subset
                yield [[first]] + smaller

        def _sort_partitions(partitions_to_sort):
            """
            Sort partitions by those we want to check first
            (typically, merging two sites into one is the
            one to try first).
            """

            partition_indices = [(idx, [len(p) for p in partition])
                                 for idx, partition in enumerate(partitions_to_sort)]

            # sort by maximum length of partition first (try smallest maximums first)
            # and secondarily by number of partitions (most partitions first, i.e.
            # create the 'least disordered' structures first)
            partition_indices = sorted(partition_indices, key=lambda x: (max(x[1]), -len(x[1])))

            # merge at most max_component sites,
            # e.g. merge at most 2 species into 1 disordered site
            partition_indices = [x for x in partition_indices if max(x[1]) <= max_components]

            partition_indices.pop(0)  # this is just the input structure

            sorted_partitions = [partitions_to_sort[x[0]] for x in partition_indices]

            return sorted_partitions

        collection = list(composition.keys())
        partitions = list(_partition(collection))
        partitions = _sort_partitions(partitions)

        return partitions
375573
void onConsumedSubpartition(int subpartitionIndex) {

		if (isReleased.get()) {
			return;
		}

		int refCnt = pendingReferences.decrementAndGet();

		if (refCnt == 0) {
			partitionManager.onConsumedPartition(this);
		}
		else if (refCnt < 0) {
			throw new IllegalStateException("All references released.");
		}

		LOG.debug("{}: Received release notification for subpartition {} (reference count now at: {}).",
				this, subpartitionIndex, pendingReferences);
	}
336959
def get_number_of_messages_in_topicpartition(self, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            topic_partition = [topic_partition]

        number_of_messages = 0
        assignment = self.consumer.assignment()

        self.consumer.unsubscribe()
        for Partition in topic_partition:
            if not isinstance(Partition, TopicPartition):
                raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")

            self.assign_to_topic_partition(Partition)

            self.consumer.seek_to_end(Partition)
            end = self.consumer.position(Partition)
            self.consumer.seek_to_beginning(Partition)
            start = self.consumer.position(Partition)
            number_of_messages += end-start

        self.consumer.unsubscribe()
        self.consumer.assign(assignment)
        return number_of_messages
559834
Node nextVictim(Node candidate) {
    if (policy == EvictionPolicy.MFU) {
      // highest, never the candidate
      return freq0.prev.nextNode.next;
    }

    // find the lowest that is not the candidate
    Node victim = freq0.next.nextNode.next;
    if (victim == candidate) {
      victim = (victim.next == victim.prev)
          ? victim.freq.next.nextNode.next
          : victim.next;
    }
    return victim;
  }
628395
def partitions_for(self, topic):
        """"""
        max_wait = self.config['max_block_ms'] / 1000.0
        return self._wait_on_metadata(topic, max_wait)
802612
public void getTopicPartitionLogSize(TopicPartitionInfo stat){
    	BrokerEndPoint leader = findLeader(stat.getTopic(), stat.getPartition()).leader();
    	SimpleConsumer consumer = getConsumerClient(leader.host(), leader.port());	
    	
    	try {			
    		long logsize = getLastOffset(consumer,stat.getTopic(), stat.getPartition(), kafka.api.OffsetRequest.LatestTime());
    		stat.setLogSize(logsize);
		} finally {
			consumer.close();
		}
    }
317662
private void moveToNextPartition() {
    if (this.currentPartitionIdx == INITIAL_PARTITION_IDX) {
      LOG.info("Pulling topic " + this.topicName);
      this.currentPartitionIdx = 0;
    } else {
      updateStatisticsForCurrentPartition();
      this.currentPartitionIdx++;
      this.currentPartitionRecordCount = 0;
      this.currentPartitionTotalSize = 0;
      this.currentPartitionDecodeRecordTime = 0;
      this.currentPartitionFetchMessageBufferTime = 0;
      this.currentPartitionReadRecordTime = 0;
      this.currentPartitionLastSuccessfulRecord = null;
    }

    this.messageIterator = null;
    if (this.currentPartitionIdx < this.partitions.size()) {
      LOG.info(String.format("Pulling partition %s from offset %d to %d, range=%d", this.getCurrentPartition(),
          this.nextWatermark.get(this.currentPartitionIdx), this.highWatermark.get(this.currentPartitionIdx),
          this.highWatermark.get(this.currentPartitionIdx) - this.nextWatermark.get(this.currentPartitionIdx)));
      switchMetricContextToCurrentPartition();
    }

    if (!allPartitionsFinished()) {
      this.startFetchEpochTime.put(this.getCurrentPartition(), System.currentTimeMillis());
    }
  }
158927
int spillInMemoryPartition(Channel.ID targetChannel, IOManager ioManager, LinkedBlockingQueue<MemorySegment> writeBehindBuffers) throws IOException {
		this.initialPartitionBuffersCount = partitionBuffers.length; // for ReOpenableHashMap
		this.initialBuildSideChannel = targetChannel;
		
		initialBuildSideWriter = ioManager.createBlockChannelWriter(targetChannel, writeBehindBuffers);
		
		final int numSegments = this.partitionBuffers.length;
		for (int i = 0; i < numSegments; i++) {
			initialBuildSideWriter.writeBlock(partitionBuffers[i]);
		}
		this.partitionBuffers = null;
		initialBuildSideWriter.close();
		// num partitions are now in the writeBehindBuffers. We propagate this information back
		return numSegments;
		
	}
1051258
def seek_to_beginning(self, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            self.consumer.seek_to_beginning(topic_partition)
        else:
            raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")
1072256
public static void markPartitionAsIndexed(int partitionId, InternalIndex[] indexes) {
        for (InternalIndex index : indexes) {
            index.markPartitionAsIndexed(partitionId);
        }
    }
20694
def partitions_for_topic(self, topic):
        
        if topic not in self._partitions:
            return None
        return set(self._partitions[topic].keys())
877122
def remove_partition(self, partition):
        """"""
        if partition in self._partitions:
            # Remove partition from set
            self._partitions.remove(partition)
            # Remove broker from replica list of partition
            partition.replicas.remove(self)
        else:
            raise ValueError(
                'Partition: {topic_id}:{partition_id} not found in broker '
                '{broker_id}'.format(
                    topic_id=partition.topic.id,
                    partition_id=partition.partition_id,
                    broker_id=self._id,
                )
            )
463163
def push_broks_to_broker(self):  # pragma: no cover - not used!
        
        someone_is_concerned = False
        sent = False
        for broker_link in self.conf.brokers:
            # Send only if the broker is concerned...
            if not broker_link.manage_arbiters:
                continue

            someone_is_concerned = True
            if broker_link.reachable:
                logger.debug("Sending %d broks to the broker %s", len(self.broks), broker_link.name)
                if broker_link.push_broks(self.broks):
                    statsmgr.counter('broks.pushed.count', len(self.broks))
                    sent = True

        if not someone_is_concerned or sent:
            # No one is anymore interested with...
            del self.broks[:]
826494
def get_topic_partition_metadata(hosts):
    
    kafka_client = KafkaToolClient(hosts, timeout=10)
    kafka_client.load_metadata_for_topics()
    topic_partitions = kafka_client.topic_partitions
    resp = kafka_client.send_metadata_request()

    for _, topic, partitions in resp.topics:
        for partition_error, partition, leader, replicas, isr in partitions:
            if topic_partitions.get(topic, {}).get(partition) is not None:
                topic_partitions[topic][partition] = PartitionMetadata(topic, partition, leader,
                                                                       replicas, isr, partition_error)
    return topic_partitions
321642
private String chooseFirstOccurence(String[] among_owners, String[] within_listgroup,
			String prior_tableName) {
		int index = CStringUtils.indexOf(within_listgroup, prior_tableName);
		for(int i = index-1; i >= 0; i-- ){
			String closest = within_listgroup[i];
			if(CStringUtils.indexOf(among_owners, closest) >= 0){
				return closest;
			}
		}
		return null;
	}