use crate::error::{Error, Result};
pub const PRODUCE: i16 = 0;
pub const FETCH: i16 = 1;
pub const LIST_OFFSETS: i16 = 2;
pub const METADATA: i16 = 3;
pub const OFFSET_COMMIT: i16 = 8;
pub const OFFSET_FETCH: i16 = 9;
pub const FIND_COORDINATOR: i16 = 10;
pub const JOIN_GROUP: i16 = 11;
pub const HEARTBEAT: i16 = 12;
pub const LEAVE_GROUP: i16 = 13;
pub const SYNC_GROUP: i16 = 14;
pub const DESCRIBE_GROUPS: i16 = 15;
pub const LIST_GROUPS: i16 = 16;
pub const SASL_HANDSHAKE: i16 = 17;
pub const API_VERSIONS: i16 = 18;
pub const CREATE_TOPICS: i16 = 19;
pub const DELETE_TOPICS: i16 = 20;
pub const DELETE_RECORDS: i16 = 21;
pub const INIT_PRODUCER_ID: i16 = 22;
pub const OFFSET_FOR_LEADER_EPOCH: i16 = 23;
pub const ADD_PARTITIONS_TO_TXN: i16 = 24;
pub const ADD_OFFSETS_TO_TXN: i16 = 25;
pub const END_TXN: i16 = 26;
pub const WRITE_TXN_MARKERS: i16 = 27;
pub const TXN_OFFSET_COMMIT: i16 = 28;
pub const DESCRIBE_ACLS: i16 = 29;
pub const CREATE_ACLS: i16 = 30;
pub const DELETE_ACLS: i16 = 31;
pub const DESCRIBE_CONFIGS: i16 = 32;
pub const ALTER_CONFIGS: i16 = 33;
pub const ALTER_REPLICA_LOG_DIRS: i16 = 34;
pub const DESCRIBE_LOG_DIRS: i16 = 35;
pub const SASL_AUTHENTICATE: i16 = 36;
pub const CREATE_PARTITIONS: i16 = 37;
pub const CREATE_DELEGATION_TOKEN: i16 = 38;
pub const RENEW_DELEGATION_TOKEN: i16 = 39;
pub const EXPIRE_DELEGATION_TOKEN: i16 = 40;
pub const DESCRIBE_DELEGATION_TOKEN: i16 = 41;
pub const DELETE_GROUPS: i16 = 42;
pub const INCREMENTAL_ALTER_CONFIGS: i16 = 44;
pub const ALTER_PARTITION_REASSIGNMENTS: i16 = 45;
pub const LIST_PARTITION_REASSIGNMENTS: i16 = 46;
pub const OFFSET_DELETE: i16 = 47;
pub const DESCRIBE_CLIENT_QUOTAS: i16 = 48;
pub const ALTER_CLIENT_QUOTAS: i16 = 49;
pub const DESCRIBE_USER_SCRAM_CREDENTIALS: i16 = 50;
pub const ALTER_USER_SCRAM_CREDENTIALS: i16 = 51;
pub const UPDATE_FEATURES: i16 = 57;
pub const DESCRIBE_CLUSTER: i16 = 60;
pub const DESCRIBE_PRODUCERS: i16 = 61;
pub const UNREGISTER_BROKER: i16 = 64;
pub const DESCRIBE_TRANSACTIONS: i16 = 65;
pub const LIST_TRANSACTIONS: i16 = 66;
pub const ALLOCATE_PRODUCER_IDS: i16 = 67;
pub const CONSUMER_GROUP_HEARTBEAT: i16 = 68;
pub const CONSUMER_GROUP_DESCRIBE: i16 = 69;
pub const GET_TELEMETRY_SUBSCRIPTIONS: i16 = 71;
pub const PUSH_TELEMETRY: i16 = 72;
pub const ASSIGN_REPLICAS_TO_DIRS: i16 = 73;
pub const LIST_CONFIG_RESOURCES: i16 = 74;
pub const DESCRIBE_TOPIC_PARTITIONS: i16 = 75;
pub const SHARE_GROUP_HEARTBEAT: i16 = 76;
pub const SHARE_GROUP_DESCRIBE: i16 = 77;
pub const SHARE_FETCH: i16 = 78;
pub const SHARE_ACKNOWLEDGE: i16 = 79;
pub const DESCRIBE_SHARE_GROUP_OFFSETS: i16 = 90;
pub const ALTER_SHARE_GROUP_OFFSETS: i16 = 91;
pub const DELETE_SHARE_GROUP_OFFSETS: i16 = 92;
#[must_use]
pub const fn name(id: i16) -> Option<&'static str> {
match id {
PRODUCE => Some("PRODUCE"),
FETCH => Some("FETCH"),
LIST_OFFSETS => Some("LIST_OFFSETS"),
METADATA => Some("METADATA"),
4 => Some("LEADER_AND_ISR"),
5 => Some("STOP_REPLICA"),
6 => Some("UPDATE_METADATA"),
7 => Some("CONTROLLED_SHUTDOWN"),
OFFSET_COMMIT => Some("OFFSET_COMMIT"),
OFFSET_FETCH => Some("OFFSET_FETCH"),
FIND_COORDINATOR => Some("FIND_COORDINATOR"),
JOIN_GROUP => Some("JOIN_GROUP"),
HEARTBEAT => Some("HEARTBEAT"),
LEAVE_GROUP => Some("LEAVE_GROUP"),
SYNC_GROUP => Some("SYNC_GROUP"),
DESCRIBE_GROUPS => Some("DESCRIBE_GROUPS"),
LIST_GROUPS => Some("LIST_GROUPS"),
SASL_HANDSHAKE => Some("SASL_HANDSHAKE"),
API_VERSIONS => Some("API_VERSIONS"),
CREATE_TOPICS => Some("CREATE_TOPICS"),
DELETE_TOPICS => Some("DELETE_TOPICS"),
DELETE_RECORDS => Some("DELETE_RECORDS"),
INIT_PRODUCER_ID => Some("INIT_PRODUCER_ID"),
OFFSET_FOR_LEADER_EPOCH => Some("OFFSET_FOR_LEADER_EPOCH"),
ADD_PARTITIONS_TO_TXN => Some("ADD_PARTITIONS_TO_TXN"),
ADD_OFFSETS_TO_TXN => Some("ADD_OFFSETS_TO_TXN"),
END_TXN => Some("END_TXN"),
WRITE_TXN_MARKERS => Some("WRITE_TXN_MARKERS"),
TXN_OFFSET_COMMIT => Some("TXN_OFFSET_COMMIT"),
DESCRIBE_ACLS => Some("DESCRIBE_ACLS"),
CREATE_ACLS => Some("CREATE_ACLS"),
DELETE_ACLS => Some("DELETE_ACLS"),
DESCRIBE_CONFIGS => Some("DESCRIBE_CONFIGS"),
ALTER_CONFIGS => Some("ALTER_CONFIGS"),
ALTER_REPLICA_LOG_DIRS => Some("ALTER_REPLICA_LOG_DIRS"),
DESCRIBE_LOG_DIRS => Some("DESCRIBE_LOG_DIRS"),
SASL_AUTHENTICATE => Some("SASL_AUTHENTICATE"),
CREATE_PARTITIONS => Some("CREATE_PARTITIONS"),
CREATE_DELEGATION_TOKEN => Some("CREATE_DELEGATION_TOKEN"),
RENEW_DELEGATION_TOKEN => Some("RENEW_DELEGATION_TOKEN"),
EXPIRE_DELEGATION_TOKEN => Some("EXPIRE_DELEGATION_TOKEN"),
DESCRIBE_DELEGATION_TOKEN => Some("DESCRIBE_DELEGATION_TOKEN"),
DELETE_GROUPS => Some("DELETE_GROUPS"),
43 => Some("ELECT_LEADERS"),
INCREMENTAL_ALTER_CONFIGS => Some("INCREMENTAL_ALTER_CONFIGS"),
ALTER_PARTITION_REASSIGNMENTS => Some("ALTER_PARTITION_REASSIGNMENTS"),
LIST_PARTITION_REASSIGNMENTS => Some("LIST_PARTITION_REASSIGNMENTS"),
OFFSET_DELETE => Some("OFFSET_DELETE"),
DESCRIBE_CLIENT_QUOTAS => Some("DESCRIBE_CLIENT_QUOTAS"),
ALTER_CLIENT_QUOTAS => Some("ALTER_CLIENT_QUOTAS"),
DESCRIBE_USER_SCRAM_CREDENTIALS => Some("DESCRIBE_USER_SCRAM_CREDENTIALS"),
ALTER_USER_SCRAM_CREDENTIALS => Some("ALTER_USER_SCRAM_CREDENTIALS"),
52 => Some("VOTE"),
53 => Some("BEGIN_QUORUM_EPOCH"),
54 => Some("END_QUORUM_EPOCH"),
55 => Some("DESCRIBE_QUORUM"),
56 => Some("ALTER_PARTITION"),
UPDATE_FEATURES => Some("UPDATE_FEATURES"),
58 => Some("ENVELOPE"),
59 => Some("FETCH_SNAPSHOT"),
DESCRIBE_CLUSTER => Some("DESCRIBE_CLUSTER"),
DESCRIBE_PRODUCERS => Some("DESCRIBE_PRODUCERS"),
62 => Some("BROKER_REGISTRATION"),
63 => Some("BROKER_HEARTBEAT"),
UNREGISTER_BROKER => Some("UNREGISTER_BROKER"),
DESCRIBE_TRANSACTIONS => Some("DESCRIBE_TRANSACTIONS"),
LIST_TRANSACTIONS => Some("LIST_TRANSACTIONS"),
ALLOCATE_PRODUCER_IDS => Some("ALLOCATE_PRODUCER_IDS"),
CONSUMER_GROUP_HEARTBEAT => Some("CONSUMER_GROUP_HEARTBEAT"),
CONSUMER_GROUP_DESCRIBE => Some("CONSUMER_GROUP_DESCRIBE"),
70 => Some("CONTROLLER_REGISTRATION"),
GET_TELEMETRY_SUBSCRIPTIONS => Some("GET_TELEMETRY_SUBSCRIPTIONS"),
PUSH_TELEMETRY => Some("PUSH_TELEMETRY"),
ASSIGN_REPLICAS_TO_DIRS => Some("ASSIGN_REPLICAS_TO_DIRS"),
LIST_CONFIG_RESOURCES => Some("LIST_CLIENT_METRICS_RESOURCES"),
DESCRIBE_TOPIC_PARTITIONS => Some("DESCRIBE_TOPIC_PARTITIONS"),
SHARE_GROUP_HEARTBEAT => Some("SHARE_GROUP_HEARTBEAT"),
SHARE_GROUP_DESCRIBE => Some("SHARE_GROUP_DESCRIBE"),
SHARE_FETCH => Some("SHARE_FETCH"),
SHARE_ACKNOWLEDGE => Some("SHARE_ACKNOWLEDGE"),
80 => Some("ADD_RAFT_VOTER"),
81 => Some("REMOVE_RAFT_VOTER"),
82 => Some("UPDATE_RAFT_VOTER"),
83 => Some("INITIALIZE_SHARE_GROUP_STATE"),
84 => Some("READ_SHARE_GROUP_STATE"),
85 => Some("WRITE_SHARE_GROUP_STATE"),
86 => Some("DELETE_SHARE_GROUP_STATE"),
87 => Some("READ_SHARE_GROUP_STATE_SUMMARY"),
DESCRIBE_SHARE_GROUP_OFFSETS => Some("DESCRIBE_SHARE_GROUP_OFFSETS"),
ALTER_SHARE_GROUP_OFFSETS => Some("ALTER_SHARE_GROUP_OFFSETS"),
DELETE_SHARE_GROUP_OFFSETS => Some("DELETE_SHARE_GROUP_OFFSETS"),
_ => None,
}
}
#[must_use]
pub const fn has_id(id: i16) -> bool {
name(id).is_some()
}
pub fn for_id(id: i16) -> Result<&'static str> {
name(id).ok_or_else(|| Error::protocol(format!("Unexpected api key: {id}")))
}
#[must_use]
pub const fn cluster_action(id: i16) -> bool {
matches!(
id,
4 | 5
| 6
| 7
| WRITE_TXN_MARKERS
| 52
| 53
| 54
| 55
| 56
| UPDATE_FEATURES
| 58
| 62
| 63
| ALLOCATE_PRODUCER_IDS
| 83
| 84
| 85
| 86
| 87
)
}
#[must_use]
pub const fn forwardable(id: i16) -> bool {
matches!(
id,
CREATE_TOPICS
| DELETE_TOPICS
| CREATE_ACLS
| DELETE_ACLS
| ALTER_CONFIGS
| CREATE_PARTITIONS
| CREATE_DELEGATION_TOKEN
| RENEW_DELEGATION_TOKEN
| EXPIRE_DELEGATION_TOKEN
| 43
| INCREMENTAL_ALTER_CONFIGS
| ALTER_PARTITION_REASSIGNMENTS
| LIST_PARTITION_REASSIGNMENTS
| ALTER_CLIENT_QUOTAS
| ALTER_USER_SCRAM_CREDENTIALS
| 55
| UPDATE_FEATURES
| UNREGISTER_BROKER
| ALLOCATE_PRODUCER_IDS
| 80
| 81
)
}
#[must_use]
pub const fn min_required_inter_broker_magic(id: i16) -> i8 {
match id {
ADD_PARTITIONS_TO_TXN
| ADD_OFFSETS_TO_TXN
| END_TXN
| WRITE_TXN_MARKERS
| TXN_OFFSET_COMMIT => 2,
_ => 0,
}
}
pub fn pick_version(
broker_min: i16,
broker_max: i16,
client_min: i16,
client_max: i16,
) -> Option<i16> {
let lo = broker_min.max(client_min);
let hi = broker_max.min(client_max);
(lo <= hi).then_some(hi)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn has_id_and_for_id_match_java() {
assert!(has_id(PRODUCE));
assert!(has_id(43));
assert!(has_id(LIST_CONFIG_RESOURCES));
assert!(has_id(DESCRIBE_SHARE_GROUP_OFFSETS));
assert!(!has_id(999));
assert!(!has_id(-1));
assert_eq!(for_id(PRODUCE).unwrap(), "PRODUCE");
assert_eq!(for_id(43).unwrap(), "ELECT_LEADERS");
assert_eq!(
for_id(LIST_CONFIG_RESOURCES).unwrap(),
"LIST_CLIENT_METRICS_RESOURCES"
);
let err = for_id(999).unwrap_err();
assert!(err.to_string().contains("Unexpected api key: 999"), "{err}");
}
#[test]
fn cluster_action_and_forwardable_match_java_4_0() {
assert!(!cluster_action(PRODUCE));
assert!(!forwardable(PRODUCE));
assert!(cluster_action(WRITE_TXN_MARKERS));
assert!(!forwardable(WRITE_TXN_MARKERS));
assert!(!cluster_action(CREATE_TOPICS));
assert!(forwardable(CREATE_TOPICS));
assert!(cluster_action(UPDATE_FEATURES));
assert!(forwardable(UPDATE_FEATURES));
assert!(!cluster_action(UNREGISTER_BROKER));
assert!(forwardable(UNREGISTER_BROKER));
assert!(cluster_action(ALLOCATE_PRODUCER_IDS));
assert!(forwardable(ALLOCATE_PRODUCER_IDS));
assert!(!cluster_action(43));
assert!(forwardable(43));
assert!(cluster_action(55));
assert!(forwardable(55));
assert!(cluster_action(4));
assert!(!forwardable(4));
assert!(!cluster_action(DESCRIBE_SHARE_GROUP_OFFSETS));
assert!(!forwardable(DESCRIBE_SHARE_GROUP_OFFSETS));
assert!(!cluster_action(999));
assert!(!forwardable(999));
}
#[test]
fn min_required_inter_broker_magic_match_java_4_0() {
use crate::RecordBatch;
assert_eq!(
min_required_inter_broker_magic(ADD_PARTITIONS_TO_TXN),
RecordBatch::MAGIC_VALUE_V2
);
assert_eq!(
min_required_inter_broker_magic(ADD_OFFSETS_TO_TXN),
RecordBatch::MAGIC_VALUE_V2
);
assert_eq!(
min_required_inter_broker_magic(END_TXN),
RecordBatch::MAGIC_VALUE_V2
);
assert_eq!(
min_required_inter_broker_magic(WRITE_TXN_MARKERS),
RecordBatch::MAGIC_VALUE_V2
);
assert_eq!(
min_required_inter_broker_magic(TXN_OFFSET_COMMIT),
RecordBatch::MAGIC_VALUE_V2
);
assert_eq!(
min_required_inter_broker_magic(PRODUCE),
RecordBatch::MAGIC_VALUE_V0
);
assert_eq!(
min_required_inter_broker_magic(CREATE_TOPICS),
RecordBatch::MAGIC_VALUE_V0
);
assert_eq!(
min_required_inter_broker_magic(UNREGISTER_BROKER),
RecordBatch::MAGIC_VALUE_V0
);
assert_eq!(
min_required_inter_broker_magic(ALLOCATE_PRODUCER_IDS),
RecordBatch::MAGIC_VALUE_V0
);
assert_eq!(
min_required_inter_broker_magic(55),
RecordBatch::MAGIC_VALUE_V0
);
assert_eq!(
min_required_inter_broker_magic(DESCRIBE_SHARE_GROUP_OFFSETS),
RecordBatch::MAGIC_VALUE_V0
);
assert_eq!(
min_required_inter_broker_magic(999),
RecordBatch::MAGIC_VALUE_V0
);
}
}