Display current and proposed changes in
    topic-partition to replica layout over brokers.
962325
def display_assignment_changes(plan_details, to_log=True):
    
    curr_plan_list, new_plan_list, total_changes = plan_details
    action_cnt = '\n[INFO] Total actions required {0}'.format(total_changes)
    _log_or_display(to_log, action_cnt)
    action_cnt = (
        '[INFO] Total actions that will be executed {0}'
        .format(len(new_plan_list))
    )
    _log_or_display(to_log, action_cnt)
    changes = ('[INFO] Proposed Changes in current cluster-layout:\n')
    _log_or_display(to_log, changes)

    tp_str = 'Topic - Partition'
    curr_repl_str = 'Previous-Assignment'
    new_rep_str = 'Proposed-Assignment'
    tp_list = [tp_repl[0] for tp_repl in curr_plan_list]

    # Display heading
    msg = '=' * 80
    _log_or_display(to_log, msg)
    row = (
        '{tp:^30s}: {curr_rep_str:^20s} ==> {new_rep_str:^20s}' .format(
            tp=tp_str,
            curr_rep_str=curr_repl_str,
            new_rep_str=new_rep_str,
        )
    )
    _log_or_display(to_log, row)
    msg = '=' * 80
    _log_or_display(to_log, msg)

    # Display each topic-partition list with changes
    tp_list_sorted = sorted(tp_list, key=lambda tp: (tp[0], tp[1]))
    for tp in tp_list_sorted:
        curr_repl = [
            tp_repl[1] for tp_repl in curr_plan_list if tp_repl[0] == tp
        ][0]
        proposed_repl = [
            tp_repl[1] for tp_repl in new_plan_list if tp_repl[0] == tp
        ][0]
        tp_str = '{topic} - {partition:<2d}'.format(topic=tp[0], partition=tp[1])
        row = (
            '{tp:<30s}: {curr_repl:<20s} ==> {proposed_repl:<20s}'.format(
                tp=tp_str,
                curr_repl=curr_repl,
                proposed_repl=proposed_repl,
            )
        )
        _log_or_display(to_log, row)
133
def move_partition_replica(self, under_loaded_rg, eligible_partition):
        """"""
        # Evaluate possible source and destination-broker
        source_broker, dest_broker = self._get_eligible_broker_pair(
            under_loaded_rg,
            eligible_partition,
        )
        if source_broker and dest_broker:
            self.log.debug(
                'Moving partition {p_name} from broker {source_broker} to '
                'replication-group:broker {rg_dest}:{dest_broker}'.format(
                    p_name=eligible_partition.name,
                    source_broker=source_broker.id,
                    dest_broker=dest_broker.id,
                    rg_dest=under_loaded_rg.id,
                ),
            )
            # Move partition if eligible brokers found
            source_broker.move_partition(eligible_partition, dest_broker)
689949
def _build_partitions(self, assignment):
        
        self.partitions = {}
        for partition_name, replica_ids in six.iteritems(assignment):
            # Get topic
            topic_id = partition_name[0]
            partition_id = partition_name[1]
            topic = self.topics.setdefault(
                topic_id,
                Topic(topic_id, replication_factor=len(replica_ids))
            )

            # Creating partition object
            partition = Partition(
                topic,
                partition_id,
                weight=self.partition_measurer.get_weight(partition_name),
                size=self.partition_measurer.get_size(partition_name),
            )
            self.partitions[partition_name] = partition
            topic.add_partition(partition)

            # Updating corresponding broker objects
            for broker_id in replica_ids:
                # Check if broker-id is present in current active brokers
                if broker_id not in list(self.brokers.keys()):
                    self.log.warning(
                        "Broker %s containing partition %s is not in "
                        "active brokers.",
                        broker_id,
                        partition,
                    )
                    self.brokers[broker_id] = self._create_broker(broker_id)

                self.brokers[broker_id].add_partition(partition)
977801
@When("^I increase '(.+?)' partitions in a Kafka topic named '(.+?)'")
    public void modifyPartitions(int numPartitions, String topic_name) throws Exception {
        commonspec.getKafkaUtils().modifyTopicPartitioning(topic_name, numPartitions);
    }
514443
def find_changed():
    

    session_token = request.headers['session_token']
    repository    = request.headers['repository']

    #===
    current_user = have_authenticated_user(request.environ['REMOTE_ADDR'], repository, session_token)
    if current_user is False: return fail(user_auth_fail_msg)

    #===
    repository_path = config['repositories'][repository]['path']
    body_data = request.get_json()

    #===
    data_store = versioned_storage(repository_path)
    head = data_store.get_head()
    if head == 'root': return success({}, {'head' : 'root', 'sorted_changes' : {'none' : []}})

    # Find changed items
    client_changes = json.loads(body_data['client_changes'])
    server_changes = data_store.get_changes_since(request.headers["previous_revision"], head)

    # Resolve conflicts
    conflict_resolutions = json.loads(body_data['conflict_resolutions'])
    if conflict_resolutions != []:
        resolutions = {'server' : {},'client' : {}}
        for r in conflict_resolutions:
            if len(r['4_resolution']) != 1 or r['4_resolution'][0] not in ['client', 'server']: return fail(conflict_msg)
            resolutions[r['4_resolution'][0]][r['1_path']] = None

        client_changes = {k : v for k,v in client_changes.iteritems() if v['path'] not in resolutions['server']}
        server_changes = {k : v for k,v in server_changes.iteritems() if v['path'] not in resolutions['client']}

    sorted_changes = merge_client_and_server_changes(server_changes, client_changes)
    return success({}, {'head' : head, 'sorted_changes': sorted_changes})
62249
def changed_action(mapper, connection, target):
    """"""
    action_history = get_history(target, 'action')
    argument_history = get_history(target, 'argument')
    owner_history = get_history(
        target,
        'user' if isinstance(target, ActionUsers) else
        'role' if isinstance(target, ActionRoles) else 'role_name')

    if action_history.has_changes() or argument_history.has_changes() \
       or owner_history.has_changes():
        current_access.delete_action_cache(
            get_action_cache_key(target.action, target.argument))
        current_access.delete_action_cache(
            get_action_cache_key(
                action_history.deleted[0] if action_history.deleted
                else target.action,
                argument_history.deleted[0] if argument_history.deleted
                else target.argument)
        )
292562
private void submitRevisions(final List<DocumentChange> changes) {
        synchronized (changesLock) {
            try {
                java.net.URI remoteUri = remote.toURI();
                for (DocumentChange change : changes) {
                    // Skip revisions that originally came from the database I'm syncing to:
                    URL source = change.getSource();
                    if (source != null && source.toURI().equals(remoteUri))
                        continue;
                    RevisionInternal rev = change.getAddedRevision();
                    if (rev == null)
                        continue;
                    if (getLocalDatabase().runFilter(filter, filterParams, rev)) {
                        if (doneBeginReplicating) {
                            pauseOrResume();
                            waitIfPaused();
                            // if not running state anymore, exit from loop.
                            if (!isRunning())
                                break;
                            RevisionInternal nuRev = rev.copyWithoutBody();
                            addToInbox(nuRev);
                        } else {
                            RevisionInternal nuRev = rev.copyWithoutBody();
                            queueChanges.add(nuRev);
                        }
                    }
                }
            } catch (java.net.URISyntaxException uriException) {
                // Not possible since it would not be an active replicator.
                // However, until we refactor everything to use java.net,
                // I'm not sure we have a choice but to swallow this.
                Log.e(TAG, "Active replicator found with invalid URI", uriException);
            }
        }
    }
19092
def run_command(self, ct, cluster_balancer):
        """"""
        if self.args.topic in ct.topics:
            topic = ct.topics[self.args.topic]
        else:
            self.log.error(
                "Topic {topic} not found. Exiting."
                .format(topic=self.args.topic),
            )
            sys.exit(1)

        if topic.replication_factor == self.args.replication_factor:
            self.log.info(
                "Topic {topic} already has replication factor {rf}. "
                "No action to perform."
                .format(topic=topic.id, rf=self.args.replication_factor),
            )
            return

        if self.args.replication_factor > len(ct.brokers):
            self.log.error(
                "Replication factor {rf} is greater than the total number of "
                "brokers {brokers}. Exiting."
                .format(
                    rf=self.args.replication_factor,
                    brokers=len(ct.brokers)
                ),
            )
            sys.exit(1)

        base_assignment = ct.assignment

        changes_per_partition = abs(
            self.args.replication_factor - topic.replication_factor
        )

        if topic.replication_factor < self.args.replication_factor:
            self.log.info(
                "Increasing topic {topic} replication factor from {old_rf} to "
                "{new_rf}."
                .format(
                    topic=topic.id,
                    old_rf=topic.replication_factor,
                    new_rf=self.args.replication_factor,
                ),
            )
            for partition in topic.partitions:
                cluster_balancer.add_replica(
                    partition.name,
                    changes_per_partition,
                )
        else:
            self.log.info(
                "Decreasing topic {topic} replication factor from {old_rf} to "
                "{new_rf}."
                .format(
                    topic=topic.id,
                    old_rf=topic.replication_factor,
                    new_rf=self.args.replication_factor,
                ),
            )
            topic_data = self.zk.get_topics(topic.id)[topic.id]
            for partition in topic.partitions:
                partition_data = topic_data['partitions'][str(partition.partition_id)]
                isr = partition_data['isr']
                osr_broker_ids = [b.id for b in partition.replicas if b.id not in isr]
                if osr_broker_ids:
                    self.log.info(
                        "The out of sync replica(s) {osr_broker_ids} will be "
                        "prioritized for removal."
                        .format(osr_broker_ids=osr_broker_ids)
                    )
                cluster_balancer.remove_replica(
                    partition.name,
                    osr_broker_ids,
                    changes_per_partition,
                )

        # Each replica addition/removal for each partition counts for one
        # partition movement
        partition_movement_count = len(topic.partitions) * changes_per_partition

        reduced_assignment = self.get_reduced_assignment(
            base_assignment,
            ct,
            max_partition_movements=partition_movement_count,
            max_leader_only_changes=0,
        )
        self.process_assignment(reduced_assignment, allow_rf_change=True)
20694
def partitions_for_topic(self, topic):
        
        if topic not in self._partitions:
            return None
        return set(self._partitions[topic].keys())
167202
def check_topics(client, req_topics):
    """"""
    client.update_cluster()
    logger.debug('Found topics: %r', client.topics.keys())

    for req_topic in req_topics:
        if req_topic not in client.topics.keys():
            err_topic_not_found = 'Topic not found: {}'.format(req_topic)
            logger.warning(err_topic_not_found)
            raise TopicNotFound(err_topic_not_found)

        topic = client.topics[req_topic]
        if not topic.partitions:
            err_topic_no_part = 'Topic has no partitions: {}'.format(req_topic)
            logger.warning(err_topic_no_part)
            raise TopicNoPartition(err_topic_no_part)

        logger.info('Topic is ready: %s', req_topic)
507269
def count_replica(self, partition):
        """"""
        return sum(1 for b in partition.replicas if b in self.brokers)
1014813
def cli(env, sortby, datacenter):
    """"""
    file_manager = SoftLayer.FileStorageManager(env.client)
    mask = "mask[serviceResource[datacenter[name]],"\
           "replicationPartners[serviceResource[datacenter[name]]]]"
    file_volumes = file_manager.list_file_volumes(datacenter=datacenter,
                                                  mask=mask)

    datacenters = dict()
    for volume in file_volumes:
        service_resource = volume['serviceResource']
        if 'datacenter' in service_resource:
            datacenter_name = service_resource['datacenter']['name']
            if datacenter_name not in datacenters.keys():
                datacenters[datacenter_name] = 1
            else:
                datacenters[datacenter_name] += 1

    table = formatting.KeyValueTable(DEFAULT_COLUMNS)
    table.sortby = sortby
    for datacenter_name in datacenters:
        table.add_row([datacenter_name, datacenters[datacenter_name]])
    env.fout(table)
212587
def on_change(self, server_description):
        """"""
        # We do no I/O holding the lock.
        with self._lock:
            # Any monitored server was definitely in the topology description
            # once. Check if it's still in the description or if some state-
            # change removed it. E.g., we got a host list from the primary
            # that didn't include this server.
            if self._description.has_server(server_description.address):
                td_old = self._description
                if self._publish_server:
                    old_server_description = td_old._server_descriptions[
                        server_description.address]
                    self._events.put((
                        self._listeners.publish_server_description_changed,
                        (old_server_description, server_description,
                         server_description.address, self._topology_id)))

                self._description = updated_topology_description(
                    self._description, server_description)

                self._update_servers()

                if self._publish_tp:
                    self._events.put((
                        self._listeners.publish_topology_description_changed,
                        (td_old, self._description, self._topology_id)))

                # Wake waiters in select_servers().
                self._condition.notify_all()
887210
def rebalance_brokers(self):
        """"""
        total_partitions = sum(len(b.partitions) for b in self.brokers)
        blacklist = set(b for b in self.brokers if b.decommissioned)
        active_brokers = self.get_active_brokers() - blacklist
        if not active_brokers:
            raise EmptyReplicationGroupError("No active brokers in %s", self._id)
        # Separate brokers based on partition count
        over_loaded_brokers, under_loaded_brokers = separate_groups(
            active_brokers,
            lambda b: len(b.partitions),
            total_partitions,
        )
        # Decommissioned brokers are considered overloaded until they have
        # no more partitions assigned.
        over_loaded_brokers += [b for b in blacklist if not b.empty()]
        if not over_loaded_brokers and not under_loaded_brokers:
            self.log.info(
                'Brokers of replication-group: %s already balanced for '
                'partition-count.',
                self._id,
            )
            return

        sibling_distance = self.generate_sibling_distance()
        while under_loaded_brokers and over_loaded_brokers:
            # Get best-fit source-broker, destination-broker and partition
            broker_source, broker_destination, victim_partition = \
                self._get_target_brokers(
                    over_loaded_brokers,
                    under_loaded_brokers,
                    sibling_distance,
                )
            # No valid source or target brokers found
            if broker_source and broker_destination:
                # Move partition
                self.log.debug(
                    'Moving partition {p_name} from broker {broker_source} to '
                    'broker {broker_destination}'
                    .format(
                        p_name=victim_partition.name,
                        broker_source=broker_source.id,
                        broker_destination=broker_destination.id,
                    ),
                )
                broker_source.move_partition(victim_partition, broker_destination)
                sibling_distance = self.update_sibling_distance(
                    sibling_distance,
                    broker_destination,
                    victim_partition.topic,
                )
            else:
                # Brokers are balanced or could not be balanced further
                break
            # Re-evaluate under and over-loaded brokers
            over_loaded_brokers, under_loaded_brokers = separate_groups(
                active_brokers,
                lambda b: len(b.partitions),
                total_partitions,
            )
            # As before add brokers to decommission.
            over_loaded_brokers += [b for b in blacklist if not b.empty()]
837160
def present(name, database, duration="7d",
            replication=1, default=False,
            **client_args):
    '''
    
    '''
    ret = {'name': name,
           'changes': {},
           'result': True,
           'comment': 'retention policy {0} is already present'.format(name)}

    if not __salt__['influxdb.retention_policy_exists'](name=name,
                                                        database=database,
                                                        **client_args):
        if __opts__['test']:
            ret['result'] = None
            ret['comment'] = ' {0} is absent and will be created'\
                .format(name)
            return ret
        if __salt__['influxdb.create_retention_policy'](
            database, name,
            duration, replication, default, **client_args
        ):
            ret['comment'] = 'retention policy {0} has been created'\
                .format(name)
            ret['changes'][name] = 'Present'
            return ret
        else:
            ret['comment'] = 'Failed to create retention policy {0}'\
                .format(name)
            ret['result'] = False
            return ret

    else:
        current_policy = __salt__['influxdb.get_retention_policy'](database=database, name=name)
        update_policy = False
        if current_policy['duration'] != convert_duration(duration):
            update_policy = True
            ret['changes']['duration'] = "Retention changed from {0} to {1}.".format(current_policy['duration'], duration)

        if current_policy['replicaN'] != replication:
            update_policy = True
            ret['changes']['replication'] = "Replication changed from {0} to {1}.".format(current_policy['replicaN'], replication)

        if current_policy['default'] != default:
            update_policy = True
            ret['changes']['default'] = "Default changed from {0} to {1}.".format(current_policy['default'], default)

        if update_policy:
            if __opts__['test']:
                ret['result'] = None
                ret['comment'] = ' {0} is present and set to be changed'\
                    .format(name)
                return ret
            else:
                if __salt__['influxdb.alter_retention_policy'](
                    database, name,
                    duration, replication, default, **client_args
                ):
                    ret['comment'] = 'retention policy {0} has been changed'\
                        .format(name)
                    return ret
                else:
                    ret['comment'] = 'Failed to update retention policy {0}'\
                        .format(name)
                    ret['result'] = False
                    return ret

    return ret
606783
private void shiftUpReplicaVersions() {
        int partitionId = getPartitionId();
        int currentReplicaIndex = migrationInfo.getDestinationCurrentReplicaIndex();
        int lostReplicaIndex = currentReplicaIndex - 1;

        try {
            InternalPartitionServiceImpl partitionService = getService();
            PartitionReplicaVersionManager partitionReplicaVersionManager = partitionService.getPartitionReplicaVersionManager();

            for (ServiceNamespace namespace : partitionReplicaVersionManager.getNamespaces(partitionId)) {
                // returns the internal array itself, not the copy
                long[] versions = partitionReplicaVersionManager.getPartitionReplicaVersions(partitionId, namespace);

                if (currentReplicaIndex > 1) {
                    long[] versionsCopy = Arrays.copyOf(versions, versions.length);
                    long version = versions[lostReplicaIndex];
                    Arrays.fill(versions, 0, lostReplicaIndex, version);

                    if (logger.isFinestEnabled()) {
                        logger.finest(
                                "Partition replica is lost! partitionId=" + partitionId
                                        + " namespace: " + namespace + " lost replicaIndex=" + lostReplicaIndex
                                        + " replica versions before shift up=" + Arrays.toString(versionsCopy)
                                        + " replica versions after shift up=" + Arrays.toString(versions));
                    }
                } else if (logger.isFinestEnabled()) {
                    logger.finest("PROMOTE partitionId=" + getPartitionId() + " namespace: " + namespace
                            + " from currentReplicaIndex=" + currentReplicaIndex);
                }
            }

            PartitionEventManager partitionEventManager = partitionService.getPartitionEventManager();
            partitionEventManager.sendPartitionLostEvent(partitionId, lostReplicaIndex);
        } catch (Throwable e) {
            logger.warning("Promotion failed. partitionId=" + partitionId + " replicaIndex=" + currentReplicaIndex, e);
        }
    }
230219
def get_unavailable_brokers(zk, partition_metadata):
    
    topic_data = zk.get_topics(partition_metadata.topic)
    topic = partition_metadata.topic
    partition = partition_metadata.partition
    expected_replicas = set(topic_data[topic]['partitions'][str(partition)]['replicas'])
    available_replicas = set(partition_metadata.replicas)
    return expected_replicas - available_replicas