'Returns consumed offsets as {TopicPartition: OffsetAndMetadata}'

def all_consumed_offsets(self):

    all_consumed = {}
    for (partition, state) in six.iteritems(self.assignment):
        if state.has_valid_position:
            all_consumed[partition] = OffsetAndMetadata(state.position, '')
    return all_consumed
