code: def balance(self):
      self._initialize_current_subscriptions()
      initializing = (
          len(
              self.current_assignment[
                  self._get_consumer_with_most_subscriptions()
              ]
          )
          == 0
      )
  
      # assign all unassigned partitions
      for partition in self.unassigned_partitions:
          # skip if there is no potential consumer for the partition
          if not self.partition_to_all_potential_consumers[partition]:
              continue
          self._assign_partition(partition)
  
      # narrow down the reassignment scope to only those partitions that can actually be reassigned
      fixed_partitions = set()
      for partition in six.iterkeys(
          self.partition_to_all_potential_consumers
      ):
          if not self._can_partition_participate_in_reassignment(
              partition
          ):
              fixed_partitions.add(partition)
      for fixed_partition in fixed_partitions:
          remove_if_present(self.sorted_partitions, fixed_partition)
          remove_if_present(self.unassigned_partitions, fixed_partition)
  
      # narrow down the reassignment scope to only those consumers that are subject to reassignment
      fixed_assignments = {}
      for consumer in six.iterkeys(
          self.consumer_to_all_potential_partitions
      ):
          if not self._can_consumer_participate_in_reassignment(
              consumer
          ):
              self._remove_consumer_from_current_subscriptions_and_maintain_order(
                  consumer
              )
              fixed_assignments[consumer] = self.current_assignment[
                  consumer
              ]
              del self.current_assignment[consumer]
  
      # create a deep copy of the current assignment so we can revert to it
      # if we do not get a more balanced assignment later
      prebalance_assignment = deepcopy(self.current_assignment)
      prebalance_partition_consumers = deepcopy(
          self.current_partition_consumer
      )
  
      # if we don't already need to revoke something due to subscription changes,
      # first try to balance by only moving newly added partitions
      if not self.revocation_required:
          self._perform_reassignments(self.unassigned_partitions)
      reassignment_performed = self._perform_reassignments(
          self.sorted_partitions
      )
  
      # if we are not preserving existing assignments and we have made changes to the current assignment
      # make sure we are getting a more balanced assignment; otherwise, revert to previous assignment
      if (
          not initializing
          and reassignment_performed
          and self._get_balance_score(self.current_assignment)
          >= self._get_balance_score(prebalance_assignment)
      ):
          self.current_assignment = prebalance_assignment
          self.current_partition_consumer.clear()
          self.current_partition_consumer.update(
              prebalance_partition_consumers
          )
  
      # add the fixed assignments (those that could not change) back
      for consumer, partitions in six.iteritems(fixed_assignments):
          self.current_assignment[consumer] = partitions
          self._add_consumer_to_current_subscriptions_and_maintain_order(
              consumer
          )
  
imports: import logging
  from collections import defaultdict, namedtuple
  from copy import deepcopy
  from kafka.cluster import ClusterMetadata
  from kafka.coordinator.assignors.abstract import AbstractPartitionAssignor
  from kafka.coordinator.assignors.sticky.partition_movements import PartitionMovements
  from kafka.coordinator.assignors.sticky.sorted_set import SortedSet
  from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment
  from kafka.coordinator.protocol import Schema
  from kafka.protocol.struct import Struct
  from kafka.protocol.types import String, Array, Int32
  from kafka.structs import TopicPartition
  from kafka.vendor import six
label_desc: Use image node_id 3 for calling the StickyAssignmentExecutor obj's underlying member method code with example usage: obj.balance() without return types