Admin client is missing list_offsets

[mstojcevich]

The admin client is missing listOffsets from the Java admin client: https://kafka.apache.org/30/javadoc/org/apache/kafka/clients/admin/Admin.html#listOffsets(java.util.Map)

[dingxiong]

It is called OffsetRequest in this Python library.

kafka-python/kafka/protocol/offset.py

Line 93 in 7ac6c6e

class OffsetRequest_v0(Request):

The API_KEY is 2. https://kafka.apache.org/protocol.html#The_Messages_ListOffsets
BTW, I wrote a simple one

import kafka

class AdminClient(kafka.KafkaAdminClient):
    def get_earliest_latest_offsets(
        self, topic_name: str, partitions: list[int] | None = None
    ) -> dict[int, tuple[int, int]]:
        partition_set = set(partitions or [])
        cluster = self._client.cluster
        if not partition_set:
            partition_set: set[
                int
            ] = cluster.available_partitions_for_topic(  # pyright: ignore
                topic_name
            )

        leaders: dict[int, set[int]] = defaultdict(set)
        for partition in partition_set:
            leader = cluster.leader_for_partition(  # pyright: ignore
                TopicPartition(topic=topic_name, partition=partition)
            )
            if not leader:
                continue
            leaders[leader].add(partition)  # pyright: ignore

        if not leaders:
            return {}

        version = 1
        timestamps = [OffsetResetStrategy.EARLIEST, OffsetResetStrategy.LATEST]
        ts_futures: list[tuple[int, Future]] = []
        for ts in timestamps:
            for leader, partition_set in leaders.items():
                by_topic: dict[str, list[tuple[int, int]]] = defaultdict(list)
                for partition in partition_set:
                    by_topic[topic_name].append((partition, ts))
                request = OffsetRequest[version](  # pyright: ignore
                    replica_id=-1, topics=list(by_topic.items())
                )
                f = self._send_request_to_node(  # pyright: ignore
                    node_id=leader, request=request  # pyright: ignore
                )
                ts_futures.append((ts, f))  # pyright: ignore

        self._wait_for_futures([f[1] for f in ts_futures])  # pyright: ignore
        partition_offset_map: dict[int, tuple[int, int]] = defaultdict(lambda: (-1, -1))
        for ts, f in ts_futures:
            response = f.value  # pyright: ignore
            for _topic, partitions_data in response.topics:  # pyright: ignore
                for (
                    partition,  # pyright: ignore
                    error_code,  # pyright: ignore
                    _timestamp,  # pyright: ignore
                    offset,  # pyright: ignore
                ) in partitions_data:
                    error_type = kafka.errors.for_code(error_code)  # pyright: ignore
                    if error_type is kafka.errors.NoError:
                        if ts == OffsetResetStrategy.EARLIEST:
                            partition_offset_map[partition] = (  # pyright: ignore
                                offset,
                                partition_offset_map[partition][1],  # pyright: ignore
                            )
                        else:
                            partition_offset_map[partition] = (  # pyright: ignore
                                partition_offset_map[partition][0],  # pyright: ignore
                                offset,
                            )
                    else:
                        raise RuntimeError(
                            f"Fail to get offset. Response is {response}"
                        )
        return partition_offset_map