Avoid too many return statements within this function.

Open

        return None if val_len == -1 else buf.get(val_len)

Avoid too many return statements within this function.

Open

        return buf.getInt8() == 1

Function add_topic_with_replicas has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def add_topic_with_replicas(cluster, topic, topic_data):
    newtopic = Topic(topic, len(topic_data['partitions']))
    for partition in topic_data['partitions']:
        for i, replica in enumerate(topic_data['partitions'][partition]):
            if replica not in cluster.brokers:

• Read up
  Read up

Function _try_pick_swap_partition has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _try_pick_swap_partition(self, partition, pos, small_partitions, large_partitions):
        """
        Given a partition and a replica position, try and find a partition that has a replica in the same position
        that can be swapped to improve rack separation. The partition selected should be the closest in size possible.

• Read up
  Read up

Function _maybe_update_metadata_for_topics has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _maybe_update_metadata_for_topics(self, topics, cache=True):
        """
        Fetch metadata for the topics specified from the cluster if cache is False or if the cached metadata has expired.
        The cluster brokers and topics are updated with the metadata information.

• Read up
  Read up

Function process_cluster has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def process_cluster(self):
        # For each broker specified, remove it from the replica list for all its partitions
        for broker_id in self.brokers:
            broker = self.cluster.brokers[broker_id]
            for position in broker.partitions:

• Read up
  Read up

Function pmap_matches_target has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def pmap_matches_target(pmap, target):
    for pos in pmap:
        for broker_id in pos:
            if pos[broker_id] != target:
                return False

• Read up
  Read up

Function run_plugins_at_step has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def run_plugins_at_step(plugins, step_name, *args):
    for plugin in plugins:
        try:
            func = getattr(plugin, step_name)
        except AttributeError:

• Read up
  Read up

Function _update_groups_from_lists has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _update_groups_from_lists(self, responses):
        """
        Given a list of ListGroups responses, make sure that all the groups are in the cluster correctly

        Args:

• Read up
  Read up

Function _parse_next_topic has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def _parse_next_topic(cmd_args):
    topic = {'topic': cmd_args.pop(0), 'partitions': []}
    while True:
        try:
            cmd_args[0].index(',')

• Read up
  Read up

Function _parse_next_topic has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def _parse_next_topic(cmd_args):
    topic = {'topic': cmd_args.pop(0), 'partitions': []}
    while True:
        try:
            cmd_args[0].index(',')

• Read up
  Read up

Function find_path_containing has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def find_path_containing(fname):
    """
    Search the PATH for the given executable filename

    :param fname: the filename to check

• Read up
  Read up

Function _parse_next_topic has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def _parse_next_topic(cmd_args):
    topic = {'topic': cmd_args.pop(0), 'partitions': []}
    while True:
        try:
            cmd_args[0].index(',')

• Read up
  Read up

Function _process_partitions_at_pos has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _process_partitions_at_pos(self, pos):
        # Create a sorted list of partitions to use at this position (descending size)
        partitions = self._get_sorted_partition_list_at_pos(pos)

        for i, partition in enumerate(partitions):

• Read up
  Read up

Function create_from_zookeeper has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def create_from_zookeeper(cls, zkconnect, default_retention=1, fetch_topics=True):
        log.info("Connecting to zookeeper {0}".format(zkconnect))
        try:
            zk = KazooClient(zkconnect)
            zk.start()

• Read up
  Read up

Function _update_brokers_from_metadata has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def _update_brokers_from_metadata(self, metadata):
        """
        Given a Metadata response (either V0 or V1), update the broker information for this
        cluster. We don't delete brokers because we don't know if the brokers is gone temporarily
        (crashed or maintenance) or permanently.

• Read up
  Read up

Function _parse_next_topic has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def _parse_next_topic(cmd_args):
    topic = {'topic': cmd_args.pop(0), 'partitions': []}
    while True:
        try:
            cmd_args[0].index(',')

• Read up
  Read up