code: 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
docs: 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.