'Initialize the coordination manager.
Keyword Arguments:
group_id (str): name of the consumer group to join for dynamic
partition assignment (if enabled), and to use for fetching and
committing offsets. Default: \'kafka-python-default-group\'
enable_auto_commit (bool): If true the consumer\'s offset will be
periodically committed in the background. Default: True.
auto_commit_interval_ms (int): milliseconds between automatic
offset commits, if enable_auto_commit is True. Default: 5000.
default_offset_commit_callback (callable): called as
callback(offsets, response) response will be either an Exception
or a OffsetCommitResponse struct. This callback can be used to
trigger custom actions when a commit request completes.
assignors (list): List of objects to use to distribute partition
ownership amongst consumer instances when group management is
used. Default: [RangePartitionAssignor, RoundRobinPartitionAssignor]
heartbeat_interval_ms (int): The expected time in milliseconds
between heartbeats to the consumer coordinator when using
Kafka\'s group management feature. Heartbeats are used to ensure
that the consumer\'s session stays active and to facilitate
rebalancing when new consumers join or leave the group. The
value must be set lower than session_timeout_ms, but typically
should be set no higher than 1/3 of that value. It can be
adjusted even lower to control the expected time for normal
rebalances. Default: 3000
session_timeout_ms (int): The timeout used to detect failures when
using Kafka\'s group managementment facilities. Default: 30000
retry_backoff_ms (int): Milliseconds to backoff when retrying on
errors. Default: 100.
exclude_internal_topics (bool): Whether records from internal topics
(such as offsets) should be exposed to the consumer. If set to
True the only way to receive records from an internal topic is
subscribing to it. Requires 0.10+. Default: True'

def __init__(self, client, subscription, metrics, **configs):

    super(ConsumerCoordinator, self).__init__(client, metrics, **configs)
    self.config = copy.copy(self.DEFAULT_CONFIG)
    for key in self.config:
        if (key in configs):
            self.config[key] = configs[key]
    if ((self.config['api_version'] >= (0, 9)) and (self.config['group_id'] is not None)):
        assert self.config['assignors'], 'Coordinator    requires    assignors'
    self._subscription = subscription
    self._metadata_snapshot = {}
    self._assignment_snapshot = None
    self._cluster = client.cluster
    self._cluster.request_update()
    self._cluster.add_listener(WeakMethod(self._handle_metadata_update))
    self._auto_commit_task = None
    if self.config['enable_auto_commit']:
        if (self.config['api_version'] < (0, 8, 1)):
            log.warning('Broker    version    (%s)    does    not    support    offset    commits;    disabling    auto-commit.', self.config['api_version'])
            self.config['enable_auto_commit'] = False
        elif (self.config['group_id'] is None):
            log.warning('group_id    is    None:    disabling    auto-commit.')
            self.config['enable_auto_commit'] = False
        else:
            interval = (self.config['auto_commit_interval_ms'] / 1000.0)
            self._auto_commit_task = AutoCommitTask(weakref.proxy(self), interval)
            self._auto_commit_task.reschedule()
    self.consumer_sensors = ConsumerCoordinatorMetrics(metrics, self.config['metric_group_prefix'], self._subscription)
