(topics, latest)
:param topics: list of topics
    :param latest: True to fetch latest offsets, False to fetch earliest available
    :return: dict: { (topic, partition): offset, ... }
:param topics: list of topics
    :param latest: True to fetch latest offsets, False to fetch earliest available
    :return: dict: { (topic, partition): offset, ... }
: param topics : list of topics : param latest : True to fetch latest offsets False to fetch earliest available : return : dict : { ( topic partition ) : offset ... }
def _get_topic_offsets(topics, latest):
    """
    :param topics: list of topics
    :param latest: True to fetch latest offsets, False to fetch earliest available
    :return: dict: { (topic, partition): offset, ... }
    """

    # https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetRequest
    # https://cfchou.github.io/blog/2015/04/23/a-closer-look-at-kafka-offsetrequest/
    assert set(topics) <= set(ALL)
    with get_simple_kafka_client() as client:
        partition_meta = client.topic_partitions

        # only return the offset of the latest message in the partition
        num_offsets = 1
        time_value = -1 if latest else -2

        offsets = {}
        offset_requests = []
        for topic in topics:
            partitions = list(partition_meta.get(topic, {}))
            for partition in partitions:
                offsets[(topic, partition)] = None
                offset_requests.append(OffsetRequestPayload(topic, partition, time_value, num_offsets))

        responses = client.send_offset_request(offset_requests)
        for r in responses:
            offsets[(r.topic, r.partition)] = r.offsets[0]

        return offsets
def _get_topic_offsets ( topics , latest ) : # https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetRequest # https://cfchou.github.io/blog/2015/04/23/a-closer-look-at-kafka-offsetrequest/ assert set ( topics ) <= set ( ALL ) with get_simple_kafka_client ( ) as client : partition_meta = client . topic_partitions # only return the offset of the latest message in the partition num_offsets = 1 time_value = - 1 if latest else - 2 offsets = { } offset_requests = [ ] for topic in topics : partitions = list ( partition_meta . get ( topic , { } ) ) for partition in partitions : offsets [ ( topic , partition ) ] = None offset_requests . append ( OffsetRequestPayload ( topic , partition , time_value , num_offsets ) ) responses = client . send_offset_request ( offset_requests ) for r in responses : offsets [ ( r . topic , r . partition ) ] = r . offsets [ 0 ] return offsets