kafka_meta/
routing.rs

1//! Where each RPC has to go.
2//!
3//! Not every request can be sent to any broker, and the failure mode of
4//! getting it wrong is not an error — it is a `NOT_CONTROLLER` or
5//! `NOT_COORDINATOR` retry loop that presents as a flaky cluster. So this is a
6//! first-class table in its own file, next to the error table, rather than a
7//! decision scattered across call sites.
8//!
9//! CLAUDE.md names four classes. [`Routing::Specific`] carries the one
10//! distinction that list glosses over: a broker the *caller* names
11//! (`DescribeLogDirs` against a particular node) and a broker the *metadata
12//! snapshot* names (a partition leader, for `Fetch`) are the same routing class
13//! but need different resolution.
14//!
15//! The wildcard arm is [`Routing::Any`], and that is the safe default in a way
16//! the read-only gate's wildcard is not: sending to the wrong broker at worst
17//! costs a redirect, whereas mis-classifying a mutating API costs the property.
18
19use kafka_conn::ApiKey;
20
21/// Which coordinator a request belongs to.
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
23pub enum CoordinatorKind {
24    /// Resolved per group id.
25    Group,
26    /// Resolved per transactional id.
27    Transaction,
28}
29
30impl CoordinatorKind {
31    /// The `key_type` byte `FindCoordinator` expects.
32    pub const fn key_type(self) -> i8 {
33        match self {
34            CoordinatorKind::Group => 0,
35            CoordinatorKind::Transaction => 1,
36        }
37    }
38}
39
40/// How a specific broker is chosen.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum BrokerSelector {
43    /// The caller names the broker id.
44    Caller,
45    /// The leader of the partition the request names.
46    PartitionLeader,
47}
48
49/// Where a request has to be sent.
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51pub enum Routing {
52    /// Any live broker will answer.
53    Any,
54    /// Only the active controller.
55    Controller,
56    /// The coordinator for a particular group or transactional id.
57    Coordinator(CoordinatorKind),
58    /// One particular broker.
59    Specific(BrokerSelector),
60}
61
62/// The routing class for an api key.
63pub const fn routing(api_key: ApiKey) -> Routing {
64    match api_key {
65        // Controller only. In KRaft a broker will forward most of these, but
66        // "most" is doing a lot of work in that sentence and the forwarding
67        // path has its own failure modes.
68        ApiKey::CreateTopics
69        | ApiKey::DeleteTopics
70        | ApiKey::CreatePartitions
71        | ApiKey::AlterPartitionReassignments
72        | ApiKey::ListPartitionReassignments
73        | ApiKey::ElectLeaders
74        | ApiKey::UpdateFeatures => Routing::Controller,
75
76        // Group coordinator, resolved per group id.
77        ApiKey::OffsetCommit
78        | ApiKey::OffsetFetch
79        | ApiKey::OffsetDelete
80        | ApiKey::JoinGroup
81        | ApiKey::Heartbeat
82        | ApiKey::LeaveGroup
83        | ApiKey::SyncGroup
84        | ApiKey::DescribeGroups
85        | ApiKey::DeleteGroups
86        | ApiKey::ConsumerGroupDescribe
87        | ApiKey::ConsumerGroupHeartbeat
88        | ApiKey::ShareGroupDescribe
89        | ApiKey::ShareGroupHeartbeat
90        | ApiKey::DescribeShareGroupOffsets
91        | ApiKey::AlterShareGroupOffsets
92        | ApiKey::DeleteShareGroupOffsets
93        | ApiKey::TxnOffsetCommit => Routing::Coordinator(CoordinatorKind::Group),
94
95        // Transaction coordinator, resolved per transactional id.
96        ApiKey::InitProducerId
97        | ApiKey::AddPartitionsToTxn
98        | ApiKey::AddOffsetsToTxn
99        | ApiKey::EndTxn
100        | ApiKey::DescribeTransactions => Routing::Coordinator(CoordinatorKind::Transaction),
101
102        // A broker the caller picks: these report that broker's own state and
103        // answering from anywhere else would be answering a different question.
104        ApiKey::DescribeLogDirs | ApiKey::AlterReplicaLogDirs | ApiKey::DescribeProducers => {
105            Routing::Specific(BrokerSelector::Caller)
106        }
107
108        // The partition leader.
109        ApiKey::Produce | ApiKey::Fetch | ApiKey::ListOffsets | ApiKey::OffsetForLeaderEpoch => {
110            Routing::Specific(BrokerSelector::PartitionLeader)
111        }
112
113        // Everything else answers from any broker: metadata, describes, ACLs,
114        // quotas, SCRAM credentials, ListGroups, ListTransactions.
115        _ => Routing::Any,
116    }
117}
118
119#[cfg(test)]
120mod tests {
121    use super::*;
122
123    #[test]
124    fn the_four_classes_from_claude_md() {
125        // Controller only.
126        for key in [
127            ApiKey::CreateTopics,
128            ApiKey::DeleteTopics,
129            ApiKey::CreatePartitions,
130            ApiKey::AlterPartitionReassignments,
131            ApiKey::ElectLeaders,
132            ApiKey::UpdateFeatures,
133        ] {
134            assert_eq!(routing(key), Routing::Controller, "{key}");
135        }
136
137        // Group/txn coordinator.
138        assert_eq!(
139            routing(ApiKey::OffsetFetch),
140            Routing::Coordinator(CoordinatorKind::Group)
141        );
142        assert_eq!(
143            routing(ApiKey::DescribeTransactions),
144            Routing::Coordinator(CoordinatorKind::Transaction)
145        );
146
147        // One specific broker.
148        assert_eq!(
149            routing(ApiKey::DescribeLogDirs),
150            Routing::Specific(BrokerSelector::Caller)
151        );
152        assert_eq!(
153            routing(ApiKey::DescribeProducers),
154            Routing::Specific(BrokerSelector::Caller)
155        );
156
157        // Any broker.
158        for key in [
159            ApiKey::DescribeConfigs,
160            ApiKey::DescribeAcls,
161            ApiKey::ListGroups,
162            ApiKey::Metadata,
163        ] {
164            assert_eq!(routing(key), Routing::Any, "{key}");
165        }
166    }
167
168    #[test]
169    fn the_read_path_goes_to_the_leader() {
170        for key in [ApiKey::Fetch, ApiKey::ListOffsets, ApiKey::Produce] {
171            assert_eq!(
172                routing(key),
173                Routing::Specific(BrokerSelector::PartitionLeader),
174                "{key}"
175            );
176        }
177    }
178
179    #[test]
180    fn share_and_consumer_group_apis_route_like_classic_groups() {
181        // KIP-848 and KIP-932 introduced new RPCs for the same resource. They
182        // are coordinator-routed exactly as DescribeGroups is; treating them as
183        // "any broker" because they are new is a NOT_COORDINATOR loop.
184        for key in [
185            ApiKey::ConsumerGroupDescribe,
186            ApiKey::ShareGroupDescribe,
187            ApiKey::DescribeShareGroupOffsets,
188        ] {
189            assert_eq!(
190                routing(key),
191                Routing::Coordinator(CoordinatorKind::Group),
192                "{key}"
193            );
194        }
195    }
196
197    #[test]
198    fn an_api_key_this_build_cannot_name_routes_anywhere() {
199        // A streams-group RPC, say. "Any broker" is the right default: at worst
200        // the broker redirects us.
201        assert_eq!(routing(ApiKey::Unknown(9_999)), Routing::Any);
202    }
203
204    #[test]
205    fn find_coordinator_key_types_match_the_protocol() {
206        assert_eq!(CoordinatorKind::Group.key_type(), 0);
207        assert_eq!(CoordinatorKind::Transaction.key_type(), 1);
208    }
209}
kafka_meta/routing.rs

kafka_meta/
routing.rs
