code: def update_cluster_topology(self, assignment):
          
          try:
              for partition_name, replica_ids in six.iteritems(assignment):
                  try:
                      new_replicas = [self.brokers[b_id] for b_id in replica_ids]
                  except KeyError:
                      self.log.error(
                          "Invalid replicas %s for topic-partition %s-%s.",
                          ', '.join([str(id) for id in replica_ids]),
                          partition_name[0],
                          partition_name[1],
                      )
                      raise InvalidBrokerIdError(
                          "Invalid replicas {0}.".format(
                              ', '.join([str(id) for id in replica_ids])
                          ),
                      )
                  try:
                      partition = self.partitions[partition_name]
                      old_replicas = [broker for broker in partition.replicas]
  
                      # No change needed. Save ourself some CPU time.
                      # Replica order matters as the first one is the leader.
                      if new_replicas == old_replicas:
                          continue
  
                      # Remove old partitions from broker
                      # This also updates partition replicas
                      for broker in old_replicas:
                          broker.remove_partition(partition)
  
                      # Add new partition to brokers
                      for broker in new_replicas:
                          broker.add_partition(partition)
                  except KeyError:
                      self.log.error(
                          "Invalid topic-partition %s-%s.",
                          partition_name[0],
                          partition_name[1],
                      )
                      raise InvalidPartitionError(
                          "Invalid topic-partition {0}-{1}."
                          .format(partition_name[0], partition_name[1]),
                      )
          except KeyError:
              self.log.error("Could not parse given assignment {0}".format(assignment))
              raise
docs: Modify the cluster-topology with given assignment.
  
          Change the replica set of partitions as in given assignment.
  
          :param assignment: dict representing actions to be used to update the current
          cluster-topology
          :raises: InvalidBrokerIdError when broker-id is invalid
          :raises: InvalidPartitionError when partition-name is invalid