def METHOD_NAME(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