fix(plugin-kafka): route every request to the broker that can answer it

[datlechin]

Fixes #2993.

Root cause

TablePro's Kafka driver had one routing model, "send it to the bootstrap connection", and Kafka has four:

Request	Must go to	A wrong broker answers
Metadata, ApiVersions, FindCoordinator	any broker	nothing, these are fine anywhere
Produce, Fetch, ListOffsets	the partition leader	NOT_LEADER_OR_FOLLOWER (6), per partition
OffsetFetch, DescribeGroups	the group coordinator	NOT_COORDINATOR (16)
ListGroups	every broker, unioned	success, with only that broker's groups
DeleteTopics	the controller	forwarded on KRaft (KIP-590)

KafkaCluster.withLeader(of:topic:) implemented the leader rule but took one partition, so only the two single-partition requests used it. ListOffsets is per-partition-batched and had nowhere to go, so it went to controlConnection() (KafkaOffsetsRequest.swift:25). Coordinator routing was never implemented at all: KafkaApiKey.findCoordinator = 10 was declared with version bounds and had zero call sites anywhere in the repo.

On a single-broker cluster every partition's leader and every group's coordinator is the broker you are already holding, so a driver that routes nothing is indistinguishable from a correct one. That is why this shipped, and why the existing integration suite never caught it: scripts/kafka-test-broker.sh started one node.

Measured, against a live 3-broker KRaft cluster (Kafka 4.3.1)

One ListOffsets v7 carrying all six partitions of a 6-partition RF3 topic, sent to each broker in turn, which is exactly what the driver did:

to node 1   p2 NONE offset 5   p3 NONE offset 3   p0 p1 p4 p5 NOT_LEADER_OR_FOLLOWER
to node 2   p0 NONE offset 6   p5 NONE offset 4   p1 p2 p3 p4 NOT_LEADER_OR_FOLLOWER
to node 3   p1 NONE offset 5   p4 NONE offset 7   p0 p2 p3 p5 NOT_LEADER_OR_FOLLOWER

Every broker answers error 6 for exactly the partitions it does not lead, per partition, inside an otherwise successful response. Nothing is forwarded. KafkaOffsetsRequest.swift:84 threw on the first such code, discarding the partitions that had answered.

Three more, measured the same way:

• OffsetFetch is not forwarded. The coordinating broker answered with six partition rows; the other two answered group-level NOT_COORDINATOR with zero rows. The comment at KafkaGroupsRequest.swift:120 claiming a broker "forwards otherwise" was false.
• ListGroups is per-broker. Six consumer groups over three brokers: each broker returned two, each with top-level error 0. SHOW GROUPS showed two of six and reported success.
• DeleteTopics is forwarded, and Metadata's controller_id is a randomly chosen live broker under KRaft: the three brokers answered 1, 2 and 1 for the same cluster. DROP TOPIC was never broken; the controller column was.

The fix

Routing becomes something KafkaCluster owns, one primitive per Kafka rule, so a request builder declares its rule instead of reaching for a connection.

• KafkaCluster+Routing.swift (new): withLeaders groups a topic's partitions by leader, resolves every leader before sending anything, regroups by connection identity so several leaders collapsing onto one socket become one request, runs the distinct connections concurrently, and merges per-partition answers. One retry after a metadata refresh for the partitions whose code says the cluster moved. withLeader is now a one-element wrapper over it, so there is a single retry implementation. withCoordinators and withEveryBroker do the same for the group rules.
• KafkaFindCoordinatorRequest.swift (new), v0 to v4. The response field order changed at v4: up to v3 the body opens with the error and then names the node; v4 (KIP-699) moved the error to the end of each coordinator entry, after the address. Reading v4 in the v3 order parses a node id out of the error code and yields a plausible broker rather than a parse failure, so both orders were measured against a real broker and are now checked by scripts/ci/kafka-protocol-probe.py.
• Errors say which broker and why. "The broker rejected the ListOffsets request: this broker no longer leads the partition" is true and unactionable. It is now the partitions that failed, the broker that could not be reached with its advertised address, or the name of the Broker Addresses setting that pinned the client to one broker.
• KafkaErrorCode: topicDeletionDisabled was 72, which is LISTENER_NOT_FOUND; 73 is TOPIC_DELETION_DISABLED. Nothing read the constant, so nothing broke. The retry set is now three classifications rather than one boolean, because "retry" alone is not an instruction: a moved leader needs a fresh Metadata read, a moved coordinator needs a fresh FindCoordinator, and a transient failure needs neither.

Shipped with it, because routing alone would have made these worse

• CONSUME ... FROM NEWEST returned the oldest messages of its window on any topic with more than one partition. The scan steps each of P partitions back by the page size and reads forward, so the merged run holds up to P pages and its newest records are at the end; the code took the front. With one partition the front and the back are the same rows, which is why every test missed it. This is the reporter's own statement, so fixing the routing alone would have turned their error into silently wrong rows.
• Page two of a tail browse was empty. Page one recorded the start of its window, so page two asked to skip a page inside a window exactly one page long. A tail scan now pins its end and later pages step back from it, through a new FROM TAIL (...) clause. The old test only asserted that no message appeared twice, which empty pages satisfy.
• bounds turned a partition with no answer into earliest 0 / latest 0, a real-looking empty partition. Unreachable while one request either answered for everything or threw; splitting per leader is exactly how a partition goes missing.

Also fixed, verified by adversarial refutation

• KafkaQL.quote and escapeStringLiteral escaped the quote but not the backslash, while the tokenizer treats a backslash as an escape. PRODUCE INTO "logs" VALUE "C:\temp\app.log" wrote C:tempapp.log and reported success, and VALUE "a\" PARTITION 3 VALUE "b" closed its own quote and set the partition. The same escaper is handed to AI agents by MCPConnectionBridge+Data.swift:22.
• CONSUME ... PARTITION (99) and PARTITION (0,two,2) silently dropped what they could not match, returning a page indistinguishable from an empty topic.
• DESCRIBE GROUP on a group that does not exist returned five columns and no rows, byte for byte what a real group with nothing committed returns. It now asks DescribeGroups first, which is what that function was written for and what nothing had ever called.
• SHOW TOPICS INTERNAL and DESCRIBE TOPIC "a" "b" parsed and discarded what they could not use.
• SHOW BROKERS and SHOW CLUSTER named a different broker as the controller on each run of a KRaft cluster. The column now says what the field actually decides, takes_admin_requests, which is true on a KRaft cluster and on a ZooKeeper one.
• controlConnection() returned the bootstrap connection with no liveness check while connect() refused to re-dial it, so one cancelled statement closed the channel and every later call threw notConnected for the rest of the session.
• KafkaConnection.send wrote its request bytes even after the one-in-flight guard had already failed the caller, so the broker answered a request nobody was waiting for and the orphan frame resumed the next caller on a correlation mismatch. Requests now queue instead of colliding, which is what makes a connection shared by the 30-second health ping and a running statement work at all.
• connection(forLeader:) read the pool, awaited a dial and only then wrote the pool, so two callers routing to the same leader both dialled and the second leaked the first's socket.

Verification

Everything below was run on this branch.

Step	Result
verify.sh test (9 Kafka suites, 3-broker cluster)	see below
verify.sh plugins (AllPlugins, 41 targets)	PASS
verify.sh build (KafkaDriverPlugin scheme)	PASS
verify.sh lint (17 changed Swift files)	0 violations
verify.sh docs	PASS
scripts/check-kafka-protocol.sh	24 of 24 (was 11)
shellcheck --severity=warning scripts/kafka-test-broker.sh	clean

Before and after, with the existing suite unchanged. scripts/kafka-test-broker.sh gains --brokers N, because one broker structurally cannot reproduce this. Running the existing KafkaIntegrationTests against three brokers instead of one, on main:

cases: 28 executed, 15 passed, 13 failed
  Caught error: .broker(code: 6, api: "ListOffsets")
  Caught error: .broker(code: 16, api: "OffsetFetch")
  Expectation failed: groups.rows.contains { $0.first?.asText == "tp-it-group" }

On this branch, with five new integration cases added: all pass.

Three of those 13 were the suite's own single-broker assumptions (brokers.rows.count == 1 and friends) rather than product defects; they are now asserted against however many brokers are running.

New tests: KafkaRoutingTests (pure, no socket) covers the error-code classification, the 72/73 constants, the replay-safety rule, per-partition answer collection, the bootstrap-only translation, partition filters and which end of a window a page comes from. KafkaQLTests covers the backslash round trip and the clause injection. KafkaIntegrationTests gains multi-broker cases for DESCRIBE TOPIC, CONSUME FROM NEWEST content, tail paging and the SHOW GROUPS union.

No before/after screenshots. The user-visible change is an error alert becoming a populated grid, and capturing it needs the registry-only Kafka plugin installed into a real user plugin directory against a live cluster. The integration suite above is the same evidence at the layer that can be run repeatably.

Review

Reviewed by Skill(security-review) (no findings at confidence 7 or above; it confirmed the quote/unquote round trip is now closed in both directions, that every newly dialled broker gets the same TLS verification and SASL handshake as the bootstrap connection, and that KafkaProtocolReader's existing bounds checks cover the new readers) and by Skill(code-review) at high effort, which found twelve issues. Eleven were fixed in a second commit; the twelfth, a claimed SwiftLint syntactic_sugar violation, did not reproduce.

The one that mattered: I had classified REQUEST_TIMED_OUT (7) as retriable, and KafkaProduceRequest asks for acks = -1, where that code means the leader appended the record and its in-sync replicas were late acknowledging. This client sends no producer id, so Kafka cannot deduplicate a replay and the message would have been written twice. Retry safety is now a property of the request: a read may always be repeated, a write only on the codes that prove the broker refused it outright.

Codex was not used: it is rate-limited on this account until 2026-09-22.

[mintlify]

Preview deployment for your docs. Learn more about Mintlify Previews.

Project	Status	Preview	Updated
TablePro	🟢 Ready	View Preview	Sep 20, 2026, 1:56 PM

💡 Tip: Enable Automations to automatically generate PRs for you.