code: 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):
                  # verify assignment is still active
                  if self._subscription.is_assigned(partition):
                      self._subscription.assignment[partition].committed = offset.offset
              self._subscription.needs_fetch_committed_offsets = False
docs: Fetch committed offsets for assigned partitions.