'Fetch the current committed offsets for specified partitions
Arguments:
partitions (list of TopicPartition): partitions to fetch
Returns:
dict: {TopicPartition: OffsetAndMetadata}'

def fetch_committed_offsets(self, partitions):

    if (not partitions):
        return {}
    while True:
        self.ensure_coordinator_known()
        future = self._send_offset_fetch_request(partitions)
        self._client.poll(future=future)
        if future.succeeded():
            return future.value
        if (not future.retriable()):
            raise future.exception
        time.sleep((self.config['retry_backoff_ms'] / 1000.0))
