docs: def all_consumed_offsets(self):
          """Returns consumed offsets as {TopicPartition: OffsetAndMetadata}"""
          all_consumed = {}
          for partition, state in six.iteritems(self.assignment):
              if state.has_valid_position:
                  all_consumed[partition] = OffsetAndMetadata(state.position, '')
          return all_consumed