// ListOffsets v10/v11 — offset resolution by timestamp or sentinel.
// Field order is pinned from the Kafka 4.3 message schemas
// ListOffsetsRequest.json / ListOffsetsResponse.json. v10 appends a
// timeout_ms for tiered-storage reads (KIP-1075); v11 adds the earliest
// pending upload sentinel (KIP-1023) — same wire shape. The response is
// unchanged since v7. A timestamp query (OffsetForTimestamp) is this API
// with a wall-clock timestamp instead of a sentinel.
///|
/// Sentinel timestamps understood by ListOffsets.
pub const OFFSET_EARLIEST : Int64 = -2L
///|
pub const OFFSET_LATEST : Int64 = -1L
///|
/// Offset of the record with the largest timestamp (KIP-734, v7+).
pub const OFFSET_MAX_TIMESTAMP : Int64 = -3L
///|
/// Earliest log start offset still in the local log (KIP-405, v8+).
pub const OFFSET_LOCAL_LOG_START : Int64 = -4L
///|
/// Last tiered offset (KIP-1005, v9+).
pub const OFFSET_LAST_TIERED : Int64 = -5L
///|
/// Earliest pending upload offset (KIP-1023, v11+).
pub const OFFSET_EARLIEST_PENDING_UPLOAD : Int64 = -6L
///|
/// Encode a ListOffsets v10/v11 request body: one topic, one entry per
/// partition. `timestamp` is a sentinel or a wall-clock query.
pub fn encode_list_offsets_request(
version : Int,
topic : String,
partitions : Array[PartitionInfo],
timestamp : Int64,
timeout_ms? : Int = 0,
) -> Bytes raise {
if version < 10 || version > 11 {
raise ProtocolError::ProtocolError(
"moonkafka implements ListOffsets v10-v11, got v\{version}",
)
}
let body = @buf.Encoder::new()
body.write_i32(-1) // replica_id: consumer
body.write_i8(0) // isolation_level: READ_UNCOMMITTED
body.write_compact_len(1) // one topic
body.write_compact_string(topic)
body.write_compact_len(partitions.length())
for p in partitions {
body.write_i32(p.index)
body.write_i32(p.leader_epoch) // current_leader_epoch
body.write_i64(timestamp)
body.write_tag_buffer()
}
body.write_tag_buffer()
body.write_i32(timeout_ms)
body.write_tag_buffer()
body.to_bytes()
}
///|
/// Decode a ListOffsets v10/v11 response body (the shape is unchanged
/// since v7). Returns partition index -> offset plus the throttle hint
/// in milliseconds.
pub fn decode_list_offsets_response(
d : @buf.Decoder,
) -> (Map[Int, Int64], Int) raise {
let throttle = d.read_i32() // throttle_time_ms
let offsets : Map[Int, Int64] = Map([])
let topic_count = d.read_compact_len()
for _ in 0..