// FindCoordinator v4 — the coordinator lookup the group/transaction
// clients (P3/P4) build on. Pinned from FindCoordinatorRequest.json /
// FindCoordinatorResponse.json in the Kafka 4.3 tree: v4 (KIP-699)
// carries the batched coordinator_keys request and returns the batched
// coordinators array; flexible since v3.
///|
/// What kind of coordinator to look up. Group and Transaction cover the
/// consumer-group and transaction flows; Share arrives with KIP-932.
pub(all) enum CoordinatorType {
Group
Transaction
Share
} derive(@debug.Debug, Eq)
///|
fn CoordinatorType::code(self : CoordinatorType) -> Int {
match self {
Group => 0
Transaction => 1
Share => 2
}
}
///|
/// One coordinator lookup result; error_code names per-key failures
/// (COORDINATOR_NOT_AVAILABLE, GROUP_AUTHORIZATION_FAILED, ...).
pub struct CoordinatorInfo {
key : String
node_id : Int
host : String
port : Int
error_code : Int
error_message : String?
} derive(@debug.Debug)
///|
/// Encode a FindCoordinator v4 request body: the batched keys plus the
/// coordinator type (field order per the v4 schema).
pub fn encode_find_coordinator_request(
keys : Array[String],
coordinator_type : CoordinatorType,
) -> Bytes {
let body = @buf.Encoder::new()
body.write_i8(coordinator_type.code())
body.write_compact_len(keys.length())
for key in keys {
body.write_compact_string(key)
}
body.write_tag_buffer()
body.to_bytes()
}
///|
/// Decode a FindCoordinator v4 response body: one entry per requested
/// key, plus the throttle hint.
pub fn decode_find_coordinator_response(
d : @buf.Decoder,
) -> (Array[CoordinatorInfo], Int) raise {
let throttle = d.read_i32()
let n = d.read_compact_len()
let out : Array[CoordinatorInfo] = []
for _ in 0.. Array[CoordinatorInfo] {
let version = self.api_version(API_FIND_COORDINATOR)
let d = self.request(
API_FIND_COORDINATOR,
version,
encode_find_coordinator_request(keys, coordinator_type),
timeout_ms~,
)
let (coordinators, throttle) = decode_find_coordinator_response(d)
self.note_throttle(throttle)
coordinators
}