'Fetch committed offsets for assigned partitions.'

def refresh_committed_offsets_if_needed(self):

    if self._subscription.needs_fetch_committed_offsets:
        offsets = self.fetch_committed_offsets(self._subscription.assigned_partitions())
        for (partition, offset) in six.iteritems(offsets):
            if self._subscription.is_assigned(partition):
                self._subscription.assignment[partition].committed = offset.offset
        self._subscription.needs_fetch_committed_offsets = False
