Return the next offset that should be fetched for the specified partition.

 @param topic [String] the name of the topic.
 @param partition [Integer] the partition number.
 @return [Integer] the next offset that should be fetched.
765543
def next_offset_for(topic, partition)
      offset = @processed_offsets.fetch(topic, {}).fetch(partition) {
        committed_offset_for(topic, partition)
      }

      # A negative offset means that no offset has been committed, so we need to
      # resolve the default offset for the topic.
      if offset < 0
        resolve_offset(topic, partition)
      else
        # The next offset is the last offset.
        offset
      end
    end
826494
def get_topic_partition_metadata(hosts):
    
    kafka_client = KafkaToolClient(hosts, timeout=10)
    kafka_client.load_metadata_for_topics()
    topic_partitions = kafka_client.topic_partitions
    resp = kafka_client.send_metadata_request()

    for _, topic, partitions in resp.topics:
        for partition_error, partition, leader, replicas, isr in partitions:
            if topic_partitions.get(topic, {}).get(partition) is not None:
                topic_partitions[topic][partition] = PartitionMetadata(topic, partition, leader,
                                                                       replicas, isr, partition_error)
    return topic_partitions
427343
private static List<TopicPartition> convertKafkaPartitions(List<KafkaTopicPartitionState<TopicPartition>> partitions) {
		ArrayList<TopicPartition> result = new ArrayList<>(partitions.size());
		for (KafkaTopicPartitionState<TopicPartition> p : partitions) {
			result.add(p.getKafkaPartitionHandle());
		}
		return result;
	}
544482
def commit_partition_offsets(self, partition_offsets):
        
        self.logger.debug("Committing partition offsets: %s", partition_offsets)

        commit_requests = [
            OffsetCommitRequestPayload(self.consumer.topic, partition, offset, None)
            for partition, offset in partition_offsets.items()
        ]
        commit_responses = self.consumer.client.send_offset_commit_request(
            self.consumer.group,
            commit_requests,
        )
        for commit_response in commit_responses:
            check_error(commit_response)
724412
def pause(list)
      unless list.is_a?(TopicPartitionList)
        raise TypeError.new("list has to be a TopicPartitionList")
      end
      tpl = list.to_native_tpl
      response = Rdkafka::Bindings.rd_kafka_pause_partitions(@native_kafka, tpl)

      if response != 0
        list = TopicPartitionList.from_native_tpl(tpl)
        raise Rdkafka::RdkafkaTopicPartitionListError.new(response, list, "Error pausing '#{list.to_h}'")
      end
    end
250632
def get_number_of_messages_in_topics(self, topics):
        

        if not isinstance(topics, list):
            topics = [topics]

        number_of_messages = 0
        for t in topics:
            part = self.get_kafka_partitions_for_topic(topic=t)
            Partitions = map(lambda p: TopicPartition(topic=t, partition=p), part)
            number_of_messages += self.get_number_of_messages_in_topicpartition(Partitions)

        return number_of_messages
217115
def next_offsets(self):
        # type: (Descriptor) -> Offsets
        
        resume_bytes = self._resume()
        with self._meta_lock:
            if self._chunk_num >= self._total_chunks:
                return None, resume_bytes
            if self._offset + self._chunk_size > self._ase.size:
                num_bytes = self._ase.size - self._offset
            else:
                num_bytes = self._chunk_size
            chunk_num = self._chunk_num
            range_start = self._offset
            range_end = self._offset + num_bytes - 1
            self._offset += num_bytes
            self._chunk_num += 1
            if self._ase.is_encrypted and self._offset >= self._ase.size:
                pad = True
            else:
                pad = False
            return Offsets(
                chunk_num=chunk_num,
                num_bytes=num_bytes,
                range_start=range_start,
                range_end=range_end,
                pad=pad,
            ), resume_bytes
1051258
def seek_to_beginning(self, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            self.consumer.seek_to_beginning(topic_partition)
        else:
            raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")
887241
def get_offset_topic_partition_count(kafka_config):
    """"""
    metadata = get_topic_partition_metadata(kafka_config.broker_list)
    if CONSUMER_OFFSET_TOPIC not in metadata:
        raise UnknownTopic("Consumer offset topic is missing.")
    return len(metadata[CONSUMER_OFFSET_TOPIC])
1018797
def _fetch_partition_info(self, topic_id, partition_id):
        """"""
        info_path = "/brokers/topics/{topic_id}/partitions/{p_id}"
        try:
            _, partition_info = self.get(
                info_path.format(topic_id=topic_id, p_id=partition_id),
            )
            return partition_info
        except NoNodeError:
            return {}
451706
public boolean seekToEnd(String topic) {
        KafkaConsumer<?, ?> consumer = _getConsumer(topic);
        return KafkaHelper.seekToEnd(consumer, topic);
    }
943433
public IndexRecord getIndex(int partition) {
    final int pos = partition * MapTask.MAP_OUTPUT_INDEX_RECORD_LENGTH / 8;
    return new IndexRecord(entries.get(pos), entries.get(pos + 1),
                           entries.get(pos + 2));
  }
1005290
def fetch_committed_offsets(self, partitions):
        
        if not partitions:
            return {}

        while True:
            self.ensure_coordinator_ready()

            # contact coordinator to fetch committed offsets
            future = self._send_offset_fetch_request(partitions)
            self._client.poll(future=future)

            if future.succeeded():
                return future.value

            if not future.retriable():
                raise future.exception # pylint: disable-msg=raising-bad-type

            time.sleep(self.config['retry_backoff_ms'] / 1000)
54125
public List<Long> getOffsets(OffsetRequest offsetRequest) {
        ILog log = getLog(offsetRequest.topic, offsetRequest.partition);
        if (log != null) {
            return log.getOffsetsBefore(offsetRequest);
        }
        return ILog.EMPTY_OFFSETS;
    }
957765
def seek(self, offset, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            self.consumer.seek(topic_partition, offset=offset)
        else:
            raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")
973555
def nullify_offsets(offsets):
    
    result = {}
    for topic, partition_offsets in six.iteritems(offsets):
        result[topic] = _nullify_partition_offsets(partition_offsets)
    return result
222921
public int getNumPartitions(String topicName) {
        List<PartitionInfo> partitionList = getPartitionInfo(topicName);
        return partitionList != null ? partitionList.size() : 0;
    }