class_docstr: This class manages the coordination process with the consumer coordinator.
code: def _perform_assignment(
      self, leader_id, assignment_strategy, members
  ):
      assignor = self._lookup_assignor(assignment_strategy)
      assert assignor, "Invalid assignment protocol: %s" % (
          assignment_strategy,
      )
      member_metadata = {}
      all_subscribed_topics = set()
      for member_id, metadata_bytes in members:
          metadata = ConsumerProtocol.METADATA.decode(metadata_bytes)
          member_metadata[member_id] = metadata
          all_subscribed_topics.update(
              metadata.subscription
          )  # pylint: disable-msg=no-member
  
      # the leader will begin watching for changes to any of the topics
      # the group is interested in, which ensures that all metadata changes
      # will eventually be seen
      # Because assignment typically happens within response callbacks,
      # we cannot block on metadata updates here (no recursion into poll())
      self._subscription.group_subscribe(all_subscribed_topics)
      self._client.set_topics(self._subscription.group_subscription())
  
      # keep track of the metadata used for assignment so that we can check
      # after rebalance completion whether anything has changed
      self._cluster.request_update()
      self._is_leader = True
      self._assignment_snapshot = self._metadata_snapshot
  
      log.debug(
          "Performing assignment for group %s using strategy %s"
          " with subscriptions %s",
          self.group_id,
          assignor.name,
          member_metadata,
      )
  
      assignments = assignor.assign(self._cluster, member_metadata)
  
      log.debug(
          "Finished assignment for group %s: %s",
          self.group_id,
          assignments,
      )
  
      group_assignment = {}
      for member_id, assignment in six.iteritems(assignments):
          group_assignment[member_id] = assignment
      return group_assignment
  
imports: from __future__ import absolute_import, division
  import collections
  import copy
  import functools
  import logging
  import time
  from kafka.vendor import six
  from kafka.coordinator.base import BaseCoordinator, Generation
  from kafka.coordinator.assignors.range import RangePartitionAssignor
  from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
  from kafka.coordinator.assignors.sticky.sticky_assignor import StickyPartitionAssignor
  from kafka.coordinator.protocol import ConsumerProtocol
  import kafka.errors
  from kafka.future import Future
  from kafka.metrics import AnonMeasurable
  from kafka.metrics.stats import Avg, Count, Max, Rate
  from kafka.protocol.commit import OffsetCommitRequest, OffsetFetchRequest
  from kafka.structs import OffsetAndMetadata, TopicPartition
  from kafka.util import WeakMethod
label_desc: Use image node_id 12 for calling the ConsumerCoordinator obj's underlying member method code with example usage: obj._perform_assignment(leader_id, assignment_strategy, members) and returns: group_assignment