<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def poll(self):
 """ Poll for coordinator events. Only applicable if group_id is set, and broker version supports GroupCoordinators. This ensures that the coordinator is known, and if using automatic partition assignment, ensures that the consumer has joined the group. This also handles periodic offset commits if they are enabled. """

        if self.group_id is None or self.config['api_version'] < (0, 8, 2):
            return

        self._invoke_completed_offset_commit_callbacks()
        self.ensure_coordinator_ready()

        if self.config['api_version'] >= (0, 9) and self._subscription.partitions_auto_assigned():
            if self.need_rejoin():
                # due to a race condition between the initial metadata fetch and the
                # initial rebalance, we need to ensure that the metadata is fresh
                # before joining initially, and then request the metadata update. If
                # metadata update arrives while the rebalance is still pending (for
                # example, when the join group is still inflight), then we will lose
                # track of the fact that we need to rebalance again to reflect the
                # change to the topic subscription. Without ensuring that the
                # metadata is fresh, any metadata update that changes the topic
                # subscriptions and arrives while a rebalance is in progress will
                # essentially be ignored. See KAFKA-3949 for the complete
                # description of the problem.
                if self._subscription.subscribed_pattern:
                    metadata_update = self._client.cluster.request_update()
                    self._client.poll(future=metadata_update)

                self.ensure_active_group()

            self.poll_heartbeat()

        self._maybe_auto_commit_offsets_async()