// API version negotiation (plan decision D1). This file is the single
// source of truth for two things: which API key numbers and versions this
// driver implements (the matrix), and how a request picks its version
// against the ranges a broker advertised in its ApiVersions response.
//
// Root scope for now; folds into the protocol/ package split together
// with the per-API codecs.
///|
// API keys are pinned from the 4.3 protocol guide as each codec lands;
// they are stable across 4.x.
pub const API_PRODUCE : Int = 0
///|
pub const API_FETCH : Int = 1
///|
pub const API_OFFSET_COMMIT : Int = 8
///|
pub const API_OFFSET_FETCH : Int = 9
///|
pub const API_OFFSET_FOR_LEADER_EPOCH : Int = 23
///|
pub const API_CONSUMER_GROUP_HEARTBEAT : Int = 68
///|
pub const API_JOIN_GROUP : Int = 11
///|
pub const API_SYNC_GROUP : Int = 14
///|
pub const API_HEARTBEAT : Int = 12
///|
pub const API_LEAVE_GROUP : Int = 13
///|
pub const API_DESCRIBE_GROUPS : Int = 15
///|
pub const API_CONSUMER_GROUP_DESCRIBE : Int = 69
///|
pub const API_LIST_GROUPS : Int = 16
///|
pub const API_DELETE_GROUPS : Int = 42
///|
pub const API_CREATE_TOPICS : Int = 19
///|
pub const API_DELETE_TOPICS : Int = 20
///|
pub const API_DELETE_RECORDS : Int = 21
///|
pub const API_CREATE_PARTITIONS : Int = 37
///|
pub const API_OFFSET_DELETE : Int = 47
///|
pub const API_LIST_OFFSETS : Int = 2
///|
pub const API_METADATA : Int = 3
///|
pub const API_FIND_COORDINATOR : Int = 10
///|
pub const API_INIT_PRODUCER_ID : Int = 22
///|
pub const API_ADD_PARTITIONS_TO_TXN : Int = 24
///|
pub const API_ADD_OFFSETS_TO_TXN : Int = 25
///|
pub const API_END_TXN : Int = 26
///|
pub const API_TXN_OFFSET_COMMIT : Int = 28
///|
pub const API_DESCRIBE_PRODUCERS : Int = 61
///|
pub const API_DESCRIBE_TRANSACTIONS : Int = 65
///|
pub const API_LIST_TRANSACTIONS : Int = 66
///|
pub const API_DESCRIBE_TOPIC_PARTITIONS : Int = 75
///|
pub const API_DESCRIBE_ACLS : Int = 29
///|
pub const API_CREATE_ACLS : Int = 30
///|
pub const API_DELETE_ACLS : Int = 31
///|
pub const API_DESCRIBE_CLIENT_QUOTAS : Int = 48
///|
pub const API_ALTER_CLIENT_QUOTAS : Int = 49
///|
pub const API_DESCRIBE_USER_SCRAM_CREDENTIALS : Int = 50
///|
pub const API_ALTER_USER_SCRAM_CREDENTIALS : Int = 51
///|
pub const API_DESCRIBE_CONFIGS : Int = 32
///|
pub const API_ALTER_CONFIGS : Int = 33
///|
pub const API_INCREMENTAL_ALTER_CONFIGS : Int = 44
///|
pub const API_LIST_CONFIG_RESOURCES : Int = 74
///|
pub const API_DESCRIBE_LOG_DIRS : Int = 35
///|
pub const API_ELECT_LEADERS : Int = 43
///|
pub const API_ALTER_PARTITION_REASSIGNMENTS : Int = 45
///|
pub const API_LIST_PARTITION_REASSIGNMENTS : Int = 46
///|
pub const API_DESCRIBE_CLUSTER : Int = 60
///|
pub const API_UNREGISTER_BROKER : Int = 64
///|
pub const API_DESCRIBE_QUORUM : Int = 55
///|
pub const API_UPDATE_FEATURES : Int = 57
///|
pub const API_SASL_HANDSHAKE : Int = 17
///|
pub const API_API_VERSIONS : Int = 18
///|
pub const API_SASL_AUTHENTICATE : Int = 36
///|
pub const API_SHARE_GROUP_HEARTBEAT : Int = 76
///|
pub const API_SHARE_FETCH : Int = 78
///|
pub const API_SHARE_ACKNOWLEDGE : Int = 79
///|
pub const API_DESCRIBE_SHARE_GROUP_OFFSETS : Int = 90
///|
pub const API_ALTER_SHARE_GROUP_OFFSETS : Int = 91
///|
pub const API_DELETE_SHARE_GROUP_OFFSETS : Int = 92
///|
pub const API_GET_TELEMETRY_SUBSCRIPTIONS : Int = 71
///|
pub const API_PUSH_TELEMETRY : Int = 72
///|
// The handshake's own version cannot itself be negotiated: it is the
// newest version every 4.0+ broker accepts.
const API_VERSIONS_VERSION : Int = 3
///|
/// One row of the driver's version matrix: the contiguous range of
/// versions whose codecs are implemented for one API.
pub(all) struct ApiRange {
key : Int
name : String
min : Int
max : Int
} derive(@debug.Debug)
///|
/// The driver's version matrix. Grow a row's range only when codecs for
/// the new versions exist — `pick` may return any version inside a row,
/// and the encoders/decoders must be able to handle what it returns.
pub fn driver_api_range(api_key : Int) -> ApiRange? {
match api_key {
API_PRODUCE =>
Some({ key: API_PRODUCE, name: "Produce", min: 12, max: 13, })
API_FETCH => Some({ key: API_FETCH, name: "Fetch", min: 12, max: 16, })
// OffsetCommit/OffsetFetch v10 address topics by id (KIP-848); v8/v9
// keep the classic wire shape.
API_OFFSET_COMMIT =>
Some({ key: API_OFFSET_COMMIT, name: "OffsetCommit", min: 8, max: 10, })
API_OFFSET_FETCH =>
Some({ key: API_OFFSET_FETCH, name: "OffsetFetch", min: 8, max: 10, })
API_OFFSET_FOR_LEADER_EPOCH =>
Some({
key: API_OFFSET_FOR_LEADER_EPOCH,
name: "OffsetForLeaderEpoch",
min: 4,
max: 4,
})
// v1 carries SubscribedTopicRegex (KIP-848 GA).
API_CONSUMER_GROUP_HEARTBEAT =>
Some({
key: API_CONSUMER_GROUP_HEARTBEAT,
name: "ConsumerGroupHeartbeat",
min: 0,
max: 1,
})
// The classic group coordination quartet: JoinGroup v9 (reason
// field), SyncGroup v5 (protocol name/echo), Heartbeat v4 and
// LeaveGroup v5 (member identity array).
API_JOIN_GROUP =>
Some({ key: API_JOIN_GROUP, name: "JoinGroup", min: 9, max: 9, })
API_SYNC_GROUP =>
Some({ key: API_SYNC_GROUP, name: "SyncGroup", min: 5, max: 5, })
API_HEARTBEAT =>
Some({ key: API_HEARTBEAT, name: "Heartbeat", min: 4, max: 4, })
API_LEAVE_GROUP =>
Some({ key: API_LEAVE_GROUP, name: "LeaveGroup", min: 5, max: 5, })
// DescribeGroups v6 is the 4.3 ceiling and every 4.0+ broker already
// advertises it (4.0 ships validVersions 0-6), so no older shape is
// needed; v5 is the first flexible version and v6 adds the per-group
// error message (KIP-1043).
API_DESCRIBE_GROUPS =>
Some({ key: API_DESCRIBE_GROUPS, name: "DescribeGroups", min: 6, max: 6, })
// ConsumerGroupDescribe is the KIP-848 counterpart: DescribeGroups only
// sees classic groups. v1 adds the per-member MemberType field
// (KIP-1099), and 4.0 brokers already advertise 0-1, so pin to v1.
API_CONSUMER_GROUP_DESCRIBE =>
Some({
key: API_CONSUMER_GROUP_DESCRIBE,
name: "ConsumerGroupDescribe",
min: 1,
max: 1,
})
// ListGroups v5 is the 4.3 ceiling and 4.0 brokers already advertise
// 0-5, so one shape covers every supported broker. v3 is the first
// flexible version, v4 adds the per-group state filter and GroupState
// field, v5 the type filter and GroupType (KIP-848 group types).
API_LIST_GROUPS =>
Some({ key: API_LIST_GROUPS, name: "ListGroups", min: 5, max: 5, })
// DeleteGroups v2 is both the first flexible version and the ceiling;
// 4.0 brokers advertise 0-2.
API_DELETE_GROUPS =>
Some({ key: API_DELETE_GROUPS, name: "DeleteGroups", min: 2, max: 2, })
// Topic admin ops; v7 echoes topic ids, DeleteTopics v6 deletes by
// id (topic-id-first, KIP-516).
API_CREATE_TOPICS =>
Some({ key: API_CREATE_TOPICS, name: "CreateTopics", min: 7, max: 7, })
API_DELETE_TOPICS =>
Some({ key: API_DELETE_TOPICS, name: "DeleteTopics", min: 6, max: 6, })
API_CREATE_PARTITIONS =>
Some({
key: API_CREATE_PARTITIONS,
name: "CreatePartitions",
min: 3,
max: 3,
})
API_DELETE_RECORDS =>
Some({ key: API_DELETE_RECORDS, name: "DeleteRecords", min: 2, max: 2, })
// OffsetDelete v0 is the one non-flexible admin API in the set.
API_OFFSET_DELETE =>
Some({ key: API_OFFSET_DELETE, name: "OffsetDelete", min: 0, max: 0, })
API_LIST_OFFSETS =>
Some({ key: API_LIST_OFFSETS, name: "ListOffsets", min: 10, max: 11, })
API_METADATA =>
Some({ key: API_METADATA, name: "Metadata", min: 12, max: 13, })
API_FIND_COORDINATOR =>
Some({
key: API_FIND_COORDINATOR,
name: "FindCoordinator",
min: 4,
max: 4,
})
API_INIT_PRODUCER_ID =>
Some({
key: API_INIT_PRODUCER_ID,
name: "InitProducerId",
min: 5,
max: 5,
})
// AddPartitionsToTxn v3: the client shape (v4+ is the broker-batch
// API of KIP-890, unused by producers).
API_ADD_PARTITIONS_TO_TXN =>
Some({
key: API_ADD_PARTITIONS_TO_TXN,
name: "AddPartitionsToTxn",
min: 3,
max: 3,
})
API_ADD_OFFSETS_TO_TXN =>
Some({
key: API_ADD_OFFSETS_TO_TXN,
name: "AddOffsetsToTxn",
min: 4,
max: 4,
})
API_END_TXN => Some({ key: API_END_TXN, name: "EndTxn", min: 5, max: 5, })
// TxnOffsetCommit v4: v5 is wire-identical and the schema caps
// non-2PC transactional clients at v4.
API_TXN_OFFSET_COMMIT =>
Some({
key: API_TXN_OFFSET_COMMIT,
name: "TxnOffsetCommit",
min: 4,
max: 4,
})
// Transaction introspection (KIP-890 admin side). Both are v0-only and
// flexible from v0.
API_DESCRIBE_PRODUCERS =>
Some({
key: API_DESCRIBE_PRODUCERS,
name: "DescribeProducers",
min: 0,
max: 0,
})
API_DESCRIBE_TRANSACTIONS =>
Some({
key: API_DESCRIBE_TRANSACTIONS,
name: "DescribeTransactions",
min: 0,
max: 0,
})
// ListTransactions spans two versions: v2 adds the transactional-id
// pattern filter, but 4.0 brokers advertise only 0-1, so the floor is
// v1 (the duration filter) to keep every supported broker reachable.
// The response is identical across 0-2; only the request grows.
API_LIST_TRANSACTIONS =>
Some({
key: API_LIST_TRANSACTIONS,
name: "ListTransactions",
min: 1,
max: 2,
})
API_DESCRIBE_TOPIC_PARTITIONS =>
Some({
key: API_DESCRIBE_TOPIC_PARTITIONS,
name: "DescribeTopicPartitions",
min: 0,
max: 0,
})
// ACL admin: v1 added pattern types, v2 is the first flexible
// version, v3 only adds the USER resource type (same wire shape).
API_DESCRIBE_ACLS =>
Some({ key: API_DESCRIBE_ACLS, name: "DescribeAcls", min: 2, max: 3, })
API_CREATE_ACLS =>
Some({ key: API_CREATE_ACLS, name: "CreateAcls", min: 2, max: 3, })
API_DELETE_ACLS =>
Some({ key: API_DELETE_ACLS, name: "DeleteAcls", min: 2, max: 3, })
// Quota admin: v0 is the pre-flexible shape, v1 enables flexible
// versions; both wire shapes are implemented.
API_DESCRIBE_CLIENT_QUOTAS =>
Some({
key: API_DESCRIBE_CLIENT_QUOTAS,
name: "DescribeClientQuotas",
min: 0,
max: 1,
})
API_ALTER_CLIENT_QUOTAS =>
Some({
key: API_ALTER_CLIENT_QUOTAS,
name: "AlterClientQuotas",
min: 0,
max: 1,
})
// SCRAM credential admin is v0-only, flexible from the start.
API_DESCRIBE_USER_SCRAM_CREDENTIALS =>
Some({
key: API_DESCRIBE_USER_SCRAM_CREDENTIALS,
name: "DescribeUserScramCredentials",
min: 0,
max: 0,
})
API_ALTER_USER_SCRAM_CREDENTIALS =>
Some({
key: API_ALTER_USER_SCRAM_CREDENTIALS,
name: "AlterUserScramCredentials",
min: 0,
max: 0,
})
// DescribeConfigs v4 is the first flexible version and every 4.0+
// broker accepts it; v1-v3 stay non-flexible and unimplemented.
API_DESCRIBE_CONFIGS =>
Some({
key: API_DESCRIBE_CONFIGS,
name: "DescribeConfigs",
min: 4,
max: 4,
})
// AlterConfigs v2 is the first flexible version; v0/v1 are
// pre-flexible legacy shapes.
API_ALTER_CONFIGS =>
Some({ key: API_ALTER_CONFIGS, name: "AlterConfigs", min: 2, max: 2, })
// IncrementalAlterConfigs v1 is the first flexible version.
API_INCREMENTAL_ALTER_CONFIGS =>
Some({
key: API_INCREMENTAL_ALTER_CONFIGS,
name: "IncrementalAlterConfigs",
min: 1,
max: 1,
})
// v1 filters by resource type (KIP-1142); v0 only lists client
// metrics resources on older brokers.
API_LIST_CONFIG_RESOURCES =>
Some({
key: API_LIST_CONFIG_RESOURCES,
name: "ListConfigResources",
min: 0,
max: 1,
})
// v2 is the first flexible version; v3+ only add response fields,
// decoded per version up to v5 (is_cordoned, KIP-1066).
API_DESCRIBE_LOG_DIRS =>
Some({
key: API_DESCRIBE_LOG_DIRS,
name: "DescribeLogDirs",
min: 2,
max: 5,
})
API_ELECT_LEADERS =>
Some({ key: API_ELECT_LEADERS, name: "ElectLeaders", min: 2, max: 2, })
// v1 toggles AllowReplicationFactorChange; both shapes are decoded.
API_ALTER_PARTITION_REASSIGNMENTS =>
Some({
key: API_ALTER_PARTITION_REASSIGNMENTS,
name: "AlterPartitionReassignments",
min: 0,
max: 1,
})
API_LIST_PARTITION_REASSIGNMENTS =>
Some({
key: API_LIST_PARTITION_REASSIGNMENTS,
name: "ListPartitionReassignments",
min: 0,
max: 0,
})
// v1 adds endpoint_type (KIP-919), v2 fenced brokers (KIP-1073).
API_DESCRIBE_CLUSTER =>
Some({
key: API_DESCRIBE_CLUSTER,
name: "DescribeCluster",
min: 0,
max: 2,
})
API_UNREGISTER_BROKER =>
Some({
key: API_UNREGISTER_BROKER,
name: "UnregisterBroker",
min: 0,
max: 0,
})
API_DESCRIBE_QUORUM =>
Some({ key: API_DESCRIBE_QUORUM, name: "DescribeQuorum", min: 2, max: 2, })
// v2 keeps the request (validate_only + upgrade_type) but drops the
// per-feature response results.
API_UPDATE_FEATURES =>
Some({ key: API_UPDATE_FEATURES, name: "UpdateFeatures", min: 2, max: 2, })
API_API_VERSIONS =>
Some({ key: API_API_VERSIONS, name: "ApiVersions", min: 3, max: 3, })
API_SASL_HANDSHAKE =>
Some({ key: API_SASL_HANDSHAKE, name: "SaslHandshake", min: 1, max: 1, })
API_SASL_AUTHENTICATE =>
Some({
key: API_SASL_AUTHENTICATE,
name: "SaslAuthenticate",
min: 2,
max: 2,
})
// Share groups (KIP-932). ShareGroupHeartbeat is v1-only (the
// initial GA shape); ShareFetch/ShareAcknowledge span v1-v2; the
// share-offset admin set uses v0 (Alter/Delete) and v0-v1
// (Describe, the v1 default for Lag = -1).
API_SHARE_GROUP_HEARTBEAT =>
Some({
key: API_SHARE_GROUP_HEARTBEAT,
name: "ShareGroupHeartbeat",
min: 1,
max: 1,
})
API_SHARE_FETCH =>
Some({ key: API_SHARE_FETCH, name: "ShareFetch", min: 1, max: 2, })
API_SHARE_ACKNOWLEDGE =>
Some({
key: API_SHARE_ACKNOWLEDGE,
name: "ShareAcknowledge",
min: 1,
max: 2,
})
API_DESCRIBE_SHARE_GROUP_OFFSETS =>
Some({
key: API_DESCRIBE_SHARE_GROUP_OFFSETS,
name: "DescribeShareGroupOffsets",
min: 0,
max: 1,
})
API_ALTER_SHARE_GROUP_OFFSETS =>
Some({
key: API_ALTER_SHARE_GROUP_OFFSETS,
name: "AlterShareGroupOffsets",
min: 0,
max: 0,
})
API_DELETE_SHARE_GROUP_OFFSETS =>
Some({
key: API_DELETE_SHARE_GROUP_OFFSETS,
name: "DeleteShareGroupOffsets",
min: 0,
max: 0,
})
// Telemetry (KIP-714): both are v0-only and flexible from v0.
API_GET_TELEMETRY_SUBSCRIPTIONS =>
Some({
key: API_GET_TELEMETRY_SUBSCRIPTIONS,
name: "GetTelemetrySubscriptions",
min: 0,
max: 0,
})
API_PUSH_TELEMETRY =>
Some({ key: API_PUSH_TELEMETRY, name: "PushTelemetry", min: 0, max: 0, })
_ => None
}
}
///|
/// The version ranges a broker advertised in its ApiVersions response.
pub struct BrokerVersions {
ranges : Map[Int, (Int, Int)]
}
///|
/// Highest version common to the driver range [dmin, dmax] capped by
/// `want`, and the broker range [bmin, bmax]. None when they do not
/// intersect.
fn pick_common(
dmin : Int,
dmax : Int,
want : Int,
bmin : Int,
bmax : Int,
) -> Int? {
let top = Int::min(want, dmax)
let bottom = Int::max(dmin, bmin)
let upper = Int::min(top, bmax)
if upper >= bottom {
Some(upper)
} else {
None
}
}
///|
/// Choose the request version for `api_key`: the highest version common
/// to the driver's implemented range and this broker's advertised range,
/// capped at `want` (the driver's ceiling when omitted). Raises when the
/// broker does not advertise the API at all, or when no version overlaps —
/// the telltale of a pre-4.0 broker.
pub fn BrokerVersions::pick(
self : BrokerVersions,
api_key : Int,
want? : Int? = None,
) -> Int raise {
guard driver_api_range(api_key) is Some(row) else {
raise ProtocolError::ProtocolError(
"moonkafka implements no version of API key \{api_key}",
)
}
guard self.ranges.get(api_key) is Some((bmin, bmax)) else {
raise ProtocolError::ProtocolError(
"broker does not advertise API key \{api_key} (\{row.name})",
)
}
let want = want.unwrap_or(row.max)
match pick_common(row.min, row.max, want, bmin, bmax) {
Some(version) => version
None =>
raise ProtocolError::ProtocolError(
"no common \{row.name} version: broker supports \{bmin}-\{bmax}, moonkafka implements \{row.min}-\{row.max}; only Kafka 4.x+ brokers are supported",
)
}
}
///|
/// The broker's advertised range for one API, if it advertises the key.
pub fn BrokerVersions::range(
self : BrokerVersions,
api_key : Int,
) -> (Int, Int)? {
self.ranges.get(api_key)
}
///|
/// Negotiated request version for one API on this connection: the highest
/// version common to the driver matrix and the ranges the broker
/// advertised at handshake time. Raises when the handshake has not run on
/// this connection.
pub fn BrokerConnection::api_version(
self : BrokerConnection,
api_key : Int,
want? : Int? = None,
) -> Int raise {
match self.versions {
Some(versions) => versions.pick(api_key, want~)
None =>
raise ProtocolError::ProtocolError(
"API versions have not been negotiated on this connection",
)
}
}