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

CHANGELOG.md

@@ -43,6 +43,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - etcd connections failing with `Unexpected HTTP 400 from v3/maintenance/status` once authentication is enabled. (#2994)
+- `DESCRIBE TOPIC` and `CONSUME` on a Kafka cluster of more than one broker failing with `this broker no longer leads the partition`. (#2993)
+- Kafka `SHOW GROUPS` listing only the groups the bootstrap broker coordinates, and reporting success.
+- Kafka `DESCRIBE GROUP` failing with `this broker does not coordinate that group`.
+- Kafka `DESCRIBE GROUP` on a group that does not exist returning an empty table.
+- Kafka `CONSUME ... FROM NEWEST` returning the oldest messages of its window on a topic with more than one partition.
+- Empty second page when paging a Kafka topic from the newest messages.
+- Kafka sidebar row counts and `DESCRIBE TOPIC` reporting zero for a partition whose broker could not be reached.
+- Backslashes dropped from a Kafka `PRODUCE` value, and a value able to close its own quotes and set the partition.
+- Kafka `CONSUME ... PARTITION` silently ignoring a partition the topic does not have, or one that is not a number.
+- Kafka `SHOW TOPICS INTERNAL` and `DESCRIBE TOPIC "a" "b"` discarding the tokens they cannot use.
+- Kafka `SHOW BROKERS` and `SHOW CLUSTER` naming a different broker as the controller on each run of a KRaft cluster.
+- A cancelled Kafka statement leaving every later one failing with `Not connected to the Kafka cluster`.
+- A Kafka connection dialled twice at once when two requests needed the same broker.
+- `two requests overlapped on one Kafka connection` when the health check ran during a long statement.
 - etcd connections carrying a username refusing a server that has authentication disabled.
 - Raw JSON shown instead of etcd's own message when a username or password is wrong.
 - etcd connections reported as unreachable every 30 seconds for a user without the root role.

Plugins/KafkaDriverPlugin/KafkaBrowseEngine.swift

@@ -1,10 +1,20 @@
 import Foundation
 
+/// Where a scan reads from, and which end of the merged run is the page.
+struct KafkaScanWindow: Sendable {
+    /// The first offset to read in each partition.
+    let start: [Int32: Int64]
+    /// The exclusive end each partition was measured back from. Empty for a forward scan.
+    let tail: [Int32: Int64]
+    /// True when the page is the newest rows of the window rather than the oldest.
+    let readsBackward: Bool
+}
+
 struct KafkaBrowsePage: Sendable {
     let records: [KafkaRecord]
-    /// The per-partition offsets this scan started from, so the next page can resume exactly
-    /// where this one began rather than re-resolving a moving anchor.
-    let anchor: [Int32: Int64]
+    /// The window this scan was taken from, so the next page reads the same one rather than
+    /// re-resolving a moving anchor.
+    let window: KafkaScanWindow
     let truncated: Bool
 }
 
@@ -14,6 +24,13 @@ struct KafkaBrowsePage: Sendable {
 /// assembled here: read forward from an anchor in each partition, merge, then slice. That is
 /// the same shape Redis and DynamoDB already use to satisfy the host's integer limit/offset
 /// from a cursor-native store.
+///
+/// Which end of the merged run is the page depends on the start mode, and getting that wrong is
+/// invisible on one partition. `FROM NEWEST` steps each of P partitions back by the page size
+/// and then reads forward, so the merged run holds up to P times the page and its newest rows
+/// are at the END. Taking the front of it returned the oldest rows of the tail window and
+/// called them the newest messages; with one partition the front and the back are the same rows,
+/// which is why every test missed it.
 enum KafkaBrowseEngine {
     /// The host asks for `limit` rows starting at `skip`. Over-fetching `skip + limit` and
     /// slicing is the honest way to answer that, but it has to be bounded or a deep page would
@@ -25,61 +42,116 @@ enum KafkaBrowseEngine {
         let topic = try metadata.requireTopic(named: query.topic)
 
         let available = topic.partitions.map(\.index).sorted()
-        let selected = query.partitions.map { requested in
-            requested.filter { available.contains($0) }
-        } ?? available
+        let selected = try resolvePartitions(query.partitions, available: available, topic: query.topic)
         guard !selected.isEmpty else {
-            return KafkaBrowsePage(records: [], anchor: [:], truncated: false)
+            return KafkaBrowsePage(
+                records: [],
+                window: KafkaScanWindow(start: [:], tail: [:], readsBackward: false),
+                truncated: false
+            )
         }
 
         let wanted = min(query.skip + query.limit, maximumOverFetch)
-        let anchor = try await resolveAnchor(query, partitions: selected, cluster: cluster)
+        let window = try await resolveWindow(query, partitions: selected, cluster: cluster)
 
         // Each partition contributes at most `wanted` records, because the merge cannot know
         // in advance how the messages are distributed: one partition may hold the whole page.
         var collected: [KafkaRecord] = []
         var truncated = false
         for partition in selected {
             try Task.checkCancellation()
-            guard let start = anchor[partition] else { continue }
+            guard let start = window.start[partition] else { continue }
             let result = try await KafkaFetchRequest.fetch(
                 topic: query.topic,
                 partition: partition,
                 startOffset: start,
                 maximumRecords: wanted,
                 cluster: cluster
             )
-            collected.append(contentsOf: result.records)
+            var records = result.records
+            // A tail scan must not read past the end it was anchored to, or page two would show
+            // messages produced after page one and the window would not be a window at all.
+            if let end = window.tail[partition] {
+                records = records.filter { $0.offset < end }
+            }
+            collected.append(contentsOf: records)
             if result.truncated { truncated = true }
         }
 
         let ordered = KafkaRecordOrdering.merge(collected)
-        let page = Array(ordered.dropFirst(query.skip).prefix(query.limit))
+        let page = KafkaRecordOrdering.page(
+            ordered,
+            skip: query.skip,
+            limit: query.limit,
+            readsBackward: window.readsBackward
+        )
         if ordered.count > query.skip + query.limit { truncated = true }
-        return KafkaBrowsePage(records: page, anchor: anchor, truncated: truncated)
+        return KafkaBrowsePage(records: page, window: window, truncated: truncated)
+    }
+
+    /// The partitions to read, or an error naming the ones the topic does not have.
+    ///
+    /// A filter that quietly drops what it cannot match returns an empty page indistinguishable
+    /// from an empty topic. The topic name already refuses to work that way
+    /// (`KafkaClusterMetadata.requireTopic`) and a partition number is no different.
+    static func resolvePartitions(_ requested: [Int32]?, available: [Int32], topic: String) throws -> [Int32] {
+        guard let requested else { return available }
+        let missing = requested.filter { !available.contains($0) }
+        guard missing.isEmpty else {
+            throw KafkaError.unknownPartitions(topic: topic, partitions: missing, available: available)
+        }
+        return Set(requested).sorted()
     }
 
-    /// Resolves a start mode into one concrete offset per partition.
+    /// Resolves a start mode into one concrete window.
     ///
-    /// `.resolved` short-circuits, and that is the point: the browse path bakes the resolved
-    /// anchor into the query string it hands back, so re-running it for page two reads the
-    /// same window rather than re-deriving "newest" against a tail that has since moved.
-    static func resolveAnchor(
+    /// `.resolved` and `.tail` short-circuit, and that is the point: the browse path bakes the
+    /// resolved window into the query string it hands back, so re-running it for page two reads
+    /// the same window rather than re-deriving "newest" against a tail that has since moved.
+    static func resolveWindow(
         _ query: KafkaConsumeQuery,
         partitions: [Int32],
         cluster: KafkaCluster
-    ) async throws -> [Int32: Int64] {
+    ) async throws -> KafkaScanWindow {
+        let step = Int64(min(query.skip + query.limit, maximumOverFetch))
+
         switch query.start {
         case .resolved(let anchors):
-            return anchors.filter { partitions.contains($0.key) }
+            return KafkaScanWindow(
+                start: anchors.filter { partitions.contains($0.key) },
+                tail: [:],
+                readsBackward: false
+            )
+
+        case .tail(let ends):
+            // The window is fixed by the ends page one recorded, so a later page steps back
+            // from the same place. Re-deriving "newest" here would walk the window forward and
+            // page two would show page one's rows again, or none at all.
+            let kept = ends.filter { partitions.contains($0.key) }
+            let earliest = try await KafkaOffsetsRequest.listOffsets(
+                topic: query.topic,
+                partitions: Array(kept.keys),
+                timestamp: KafkaOffsetsRequest.earliestTimestamp,
+                cluster: cluster
+            )
+            var starts: [Int32: Int64] = [:]
+            for (partition, end) in kept {
+                // No fallback offset. A partition whose earliest offset did not come back has
+                // an unknown floor, and anchoring it at zero would send the page to the start of
+                // a log whose first surviving message may be far past it.
+                guard let floor = earliest[partition] else { continue }
+                starts[partition] = max(floor, end - step)
+            }
+            return KafkaScanWindow(start: starts, tail: kept, readsBackward: true)
 
         case .oldest:
-            return try await KafkaOffsetsRequest.listOffsets(
+            let starts = try await KafkaOffsetsRequest.listOffsets(
                 topic: query.topic,
                 partitions: partitions,
                 timestamp: KafkaOffsetsRequest.earliestTimestamp,
                 cluster: cluster
             )
+            return KafkaScanWindow(start: starts, tail: [:], readsBackward: false)
 
         case .offset(let offset):
             let bounds = try await KafkaOffsetsRequest.bounds(
@@ -89,11 +161,11 @@ enum KafkaBrowseEngine {
             )
             // Clamping keeps a hand-typed offset from becoming OFFSET_OUT_OF_RANGE, which
             // reads as a driver failure rather than as "that offset is not in the log".
-            var anchors: [Int32: Int64] = [:]
+            var starts: [Int32: Int64] = [:]
             for bound in bounds {
-                anchors[bound.partition] = min(max(offset, bound.earliest), bound.latest)
+                starts[bound.partition] = min(max(offset, bound.earliest), bound.latest)
             }
-            return anchors
+            return KafkaScanWindow(start: starts, tail: [:], readsBackward: false)
 
         case .timestamp(let milliseconds):
             let resolved = try await KafkaOffsetsRequest.listOffsets(
@@ -110,12 +182,12 @@ enum KafkaBrowseEngine {
                 timestamp: KafkaOffsetsRequest.latestTimestamp,
                 cluster: cluster
             )
-            var anchors: [Int32: Int64] = [:]
+            var starts: [Int32: Int64] = [:]
             for partition in partitions {
                 let candidate = resolved[partition] ?? -1
-                anchors[partition] = candidate >= 0 ? candidate : (latest[partition] ?? 0)
+                starts[partition] = candidate >= 0 ? candidate : (latest[partition] ?? 0)
             }
-            return anchors
+            return KafkaScanWindow(start: starts, tail: [:], readsBackward: false)
 
         case .newest:
             let bounds = try await KafkaOffsetsRequest.bounds(
@@ -127,12 +199,15 @@ enum KafkaBrowseEngine {
             // the page and then reads forward. Sharing the budget evenly is a guess about how
             // messages are distributed, so it is deliberately generous: taking the whole page
             // size from every partition costs one extra read and never misses a recent message.
-            let step = Int64(min(query.skip + query.limit, maximumOverFetch))
-            var anchors: [Int32: Int64] = [:]
+            // The merged run is then read from its newest end, which is what `readsBackward`
+            // says and what the partition count would otherwise hide.
+            var starts: [Int32: Int64] = [:]
+            var ends: [Int32: Int64] = [:]
             for bound in bounds {
-                anchors[bound.partition] = max(bound.earliest, bound.latest - step)
+                starts[bound.partition] = max(bound.earliest, bound.latest - step)
+                ends[bound.partition] = bound.latest
             }
-            return anchors
+            return KafkaScanWindow(start: starts, tail: ends, readsBackward: true)
         }
     }
 }