'Get the last committed offset for the given partition.
This offset will be used as the position for the consumer
in the event of a failure.
This call may block to do a remote call if the partition in question
isn\'t assigned to this consumer or if the consumer hasn\'t yet
initialized its cache of committed offsets.
Arguments:
partition (TopicPartition): The partition to check.
Returns:
The last committed offset, or None if there was no prior commit.'

def committed(self, partition):

    assert (self.config['api_version'] >= (0, 8, 1)), 'Requires    >=    Kafka    0.8.1'
    assert (self.config['group_id'] is not None), 'Requires    group_id'
    if (not isinstance(partition, TopicPartition)):
        raise TypeError('partition    must    be    a    TopicPartition    namedtuple')
    if self._subscription.is_assigned(partition):
        committed = self._subscription.assignment[partition].committed
        if (committed is None):
            self._coordinator.refresh_committed_offsets_if_needed()
            committed = self._subscription.assignment[partition].committed
    else:
        commit_map = self._coordinator.fetch_committed_offsets([partition])
        if (partition in commit_map):
            committed = commit_map[partition].offset
        else:
            committed = None
    return committed
