diff --git a/CHANGELOG.md b/CHANGELOG.md index 77be3f6c8..e2b2c599e 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/Plugins/KafkaDriverPlugin/KafkaBrowseEngine.swift b/Plugins/KafkaDriverPlugin/KafkaBrowseEngine.swift index 9b4db5460..2d0ed6bfe 100644 --- a/Plugins/KafkaDriverPlugin/KafkaBrowseEngine.swift +++ b/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,15 +42,17 @@ 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. @@ -41,7 +60,7 @@ enum KafkaBrowseEngine { 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, @@ -49,37 +68,90 @@ enum KafkaBrowseEngine { 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) } } } diff --git a/Plugins/KafkaDriverPlugin/KafkaCluster+Routing.swift b/Plugins/KafkaDriverPlugin/KafkaCluster+Routing.swift new file mode 100644 index 000000000..8e667b795 --- /dev/null +++ b/Plugins/KafkaDriverPlugin/KafkaCluster+Routing.swift @@ -0,0 +1,427 @@ +import Foundation + +/// One partition's answer inside a reply that carries several. +/// +/// Kafka batches by partition and answers by partition: a ListOffsets sent to a broker that +/// leads four of a topic's six partitions comes back with four offsets and two error codes, in +/// one successful response. Reading that as a single throwing result is what shipped #2993, +/// where the first error code discarded the partitions that had answered correctly. +enum KafkaPartitionOutcome: Sendable { + case value(Value) + case rejected(code: Int16) + case failed(KafkaError) +} + +/// Whether sending a request a second time can repeat work the broker already did. +/// +/// A read can always be repeated. A write cannot: `KafkaProduceRequest` asks for `acks = -1`, +/// where REQUEST_TIMED_OUT means the leader appended the record and its in-sync replicas did +/// not acknowledge in time, and this client sends no producer id for Kafka to deduplicate on. +/// Retrying that appends the message twice. +enum KafkaRequestRepeatability: Sendable { + case safeToRepeat + case onlyWhenBrokerRefusedIt + + /// Whether a partition that came back with this code should be sent again. + func allowsRetry(after code: Int16) -> Bool { + switch KafkaErrorCode.retryAction(for: code) { + case .report, .findCoordinatorAgain: + return false + case .resolveLeaderAgain, .retrySameBroker: + return self == .safeToRepeat || KafkaErrorCode.provesRequestWasNotApplied(code) + } + } +} + +extension KafkaPartitionOutcome { + /// Every partition's value, or the error that best explains the ones missing. + /// + /// The bootstrap-only translation lives here because it is the one place that knows both + /// the routing setting and which partitions came back short. "This broker no longer leads + /// the partition" is a true statement a user can do nothing with; naming the setting that + /// kept the client on one broker is actionable. + static func requireAll( + _ outcomes: [Int32: KafkaPartitionOutcome], + topic: String, + api: String, + routing: KafkaBrokerRouting + ) throws -> [Int32: Value] { + var values: [Int32: Value] = [:] + var ledElsewhere: [Int32] = [] + var rejected: [Int32: Int16] = [:] + var firstFailure: KafkaError? + + for (partition, outcome) in outcomes { + switch outcome { + case .value(let value): + values[partition] = value + case .rejected(let code): + if routing == .bootstrapOnly, code == KafkaErrorCode.notLeaderOrFollower { + ledElsewhere.append(partition) + } else { + rejected[partition] = code + } + case .failed(let error): + if firstFailure == nil { firstFailure = error } + } + } + + if !ledElsewhere.isEmpty { + throw KafkaError.partitionsLedElsewhere(topic: topic, partitions: ledElsewhere) + } + // Ranked, because the partitions of one topic can come back with different codes and + // only one of them can be reported. Picking the numerically smallest let one partition's + // OFFSET_OUT_OF_RANGE (1) hide TOPIC_AUTHORIZATION_FAILED (29) on the other five, which + // is the least actionable of the failures rather than the most. + if let code = chosenFailure(among: rejected) { + let affected = rejected.filter { $0.value == code }.map(\.key) + throw KafkaError.partitionsRejected(topic: topic, partitions: affected, api: api, code: code) + } + if let firstFailure { throw firstFailure } + return values + } + + /// Which of several partitions' error codes to report. + /// + /// A permission failure first: it is the one the user has to do something about, and it is + /// usually the cause of whatever else came back. Then any other real answer, then a code + /// that only says the cluster is moving. Ties go to the lowest partition so the message does + /// not change between runs of the same broken query. + private static func chosenFailure(among rejected: [Int32: Int16]) -> Int16? { + rejected + .sorted { $0.key < $1.key } + .map(\.value) + .min { rank(of: $0) < rank(of: $1) } + } + + private static func rank(of code: Int16) -> Int { + switch code { + case KafkaErrorCode.topicAuthorizationFailed, + KafkaErrorCode.groupAuthorizationFailed, + KafkaErrorCode.clusterAuthorizationFailed: + return 0 + default: + return KafkaErrorCode.retryAction(for: code) == .report ? 1 : 2 + } + } + + /// How an error thrown by one broker's sub-request is recorded against its partitions. + /// + /// A body that reports an error code by throwing still has to reach the retry. Fetch and + /// Produce name one partition and check its code inline, so folding their throw into + /// `.failed` would take away the refresh-and-retry they have always had. + static func outcome(for error: KafkaError) -> KafkaPartitionOutcome { + guard case .broker(let code, _) = error else { return .failed(error) } + return .rejected(code: code) + } +} + +extension KafkaCluster { + /// Runs a request against the leader of every partition it names, in as few requests as + /// there are leaders. + /// + /// A Kafka client is told each partition's leader by Metadata and must address the leader + /// directly; a broker answers NOT_LEADER_OR_FOLLOWER for a partition it does not lead and + /// forwards nothing. The partitions of one topic have different leaders, so a batched + /// per-partition request has to be split along that boundary, which is the whole of #2993. + /// + /// One retry, after re-reading Metadata, for the partitions whose code says the cluster + /// moved rather than that the request was wrong. Re-resolving the leader covers the + /// transient codes too: where the leader has not changed, the retry lands on the same + /// connection, which is what `retrySameBroker` asks for anyway. + func withLeaders( + topic: String, + partitions: [Int32], + repeatability: KafkaRequestRepeatability = .safeToRepeat, + _ body: @Sendable @escaping (KafkaConnection, [Int32]) async throws + -> [Int32: KafkaPartitionOutcome] + ) async throws -> [Int32: KafkaPartitionOutcome] { + let wanted = Set(partitions).sorted() + guard !wanted.isEmpty else { return [:] } + + var merged = try await routeOneRound(topic: topic, partitions: wanted, refresh: false, body: body) + let moved = wanted.filter { partition in + guard case .rejected(let code) = merged[partition] else { return false } + return repeatability.allowsRetry(after: code) + } + guard !moved.isEmpty else { return merged } + + invalidateMetadata() + let second = try await routeOneRound(topic: topic, partitions: moved, refresh: true, body: body) + merged.merge(second) { _, retried in retried } + return merged + } + + /// The single-partition case, so there is one implementation of resolve, split and retry. + func withLeader( + of partition: Int32, + topic: String, + api: String, + repeatability: KafkaRequestRepeatability = .safeToRepeat, + _ body: @Sendable @escaping (KafkaConnection) async throws -> Value + ) async throws -> Value { + let outcomes = try await withLeaders( + topic: topic, + partitions: [partition], + repeatability: repeatability + ) { connection, _ in + [partition: .value(try await body(connection))] + } + let values = try KafkaPartitionOutcome.requireAll( + outcomes, + topic: topic, + api: api, + routing: routing + ) + guard let value = values[partition] else { + throw KafkaError.partitionsUnanswered(topic: topic, partitions: [partition], api: api) + } + return value + } + + private func routeOneRound( + topic: String, + partitions: [Int32], + refresh: Bool, + body: @Sendable @escaping (KafkaConnection, [Int32]) async throws + -> [Int32: KafkaPartitionOutcome] + ) async throws -> [Int32: KafkaPartitionOutcome] { + try Task.checkCancellation() + let leaders = try await leaderMap(topic: topic, refresh: refresh) + + var outcomes: [Int32: KafkaPartitionOutcome] = [:] + var unknown: [Int32] = [] + var byLeader: [Int32: [Int32]] = [:] + for partition in partitions { + guard let leader = leaders[partition] else { + unknown.append(partition) + continue + } + // A partition between leader elections reports -1. It is not an unreachable broker + // and not a rejection, so it is reported as itself rather than routed anywhere. + guard leader >= 0 else { + outcomes[partition] = .failed(.partitionsHaveNoLeader(topic: topic, partitions: [partition])) + continue + } + byLeader[leader, default: []].append(partition) + } + if !unknown.isEmpty { + throw KafkaError.unknownPartitions( + topic: topic, + partitions: unknown, + available: leaders.keys.sorted() + ) + } + + // Resolve every leader before sending anything. A resolve can dial, and dialling + // suspends, so interleaving one leader's resolve with another's send would let two + // requests reach one connection. Grouping by connection identity rather than by leader + // is what makes bootstrap-only routing and a shared endpoint safe: several leaders + // collapsing onto one connection become one request, not several racing ones. + var plan: [(connection: KafkaConnection, partitions: [Int32])] = [] + var slotForConnection: [ObjectIdentifier: Int] = [:] + for leader in byLeader.keys.sorted() { + let group = byLeader[leader] ?? [] + do { + let connection = try await connection(forLeader: leader) + let identity = ObjectIdentifier(connection) + if let slot = slotForConnection[identity] { + plan[slot].partitions.append(contentsOf: group) + } else { + slotForConnection[identity] = plan.count + plan.append((connection, group)) + } + } catch let error as KafkaError { + for partition in group { outcomes[partition] = .failed(error) } + } + } + + // No child throws. A thrown error would exit the group and cancel its siblings, and + // `KafkaConnection.send` closes its channel on cancellation, so one leader's failure + // would tear down the healthy connections serving the others. + await withTaskGroup(of: [Int32: KafkaPartitionOutcome].self) { group in + for entry in plan { + let connection = entry.connection + let slice = entry.partitions.sorted() + group.addTask { + do { + return try await body(connection, slice) + } catch let error as KafkaError { + let outcome = KafkaPartitionOutcome.outcome(for: error) + return Dictionary(uniqueKeysWithValues: slice.map { ($0, outcome) }) + } catch { + let wrapped = KafkaError.connectionFailed(error.localizedDescription) + return Dictionary(uniqueKeysWithValues: slice.map { ($0, .failed(wrapped)) }) + } + } + } + for await answered in group { + outcomes.merge(answered) { _, new in new } + } + } + return outcomes + } + + /// Each partition's leader, from the routing cache or a fresh Metadata read. + func leaderMap(topic: String, refresh: Bool) async throws -> [Int32: Int32] { + if !refresh, let cached = leadersByTopic[topic], !cached.isEmpty { return cached } + if refresh { leadersByTopic[topic] = nil } + let metadata = try await metadata(topics: [topic], refresh: true) + _ = try metadata.requireTopic(named: topic) + guard let leaders = leadersByTopic[topic] else { throw KafkaError.unknownTopic(topic) } + return leaders + } + + // MARK: - Coordinator routing + + /// Runs a request against the broker that coordinates a consumer group. + /// + /// A group's committed offsets and its membership live on its coordinator, which is the + /// broker owning the `__consumer_offsets` partition the group id hashes to. Every other + /// broker answers NOT_COORDINATOR and forwards nothing, so FindCoordinator is not optional + /// on a cluster with more than one broker. One retry, because a coordinator that has just + /// moved or is still loading its log answers a code that says exactly that. + func withCoordinator( + of group: String, + _ body: @Sendable (KafkaConnection) async throws -> Value + ) async throws -> Value { + let answers = try await withCoordinators(of: [group]) { connection, _ in + [try await body(connection)] + } + guard let answer = answers.first else { throw KafkaError.unknownGroup(group) } + return answer + } + + /// Runs a request against every coordinator the named groups belong to. + /// + /// DescribeGroups is batched but coordinator-scoped, so a list spanning three coordinators + /// is three requests. The retry lives here rather than at each call site because a cached + /// coordinator that has moved answers NOT_COORDINATOR forever otherwise: the cache is only + /// cleared by a disconnect, so one broker restart used to break DESCRIBE GROUP for the rest + /// of the session. + func withCoordinators( + of groups: [String], + _ body: @Sendable (KafkaConnection, [String]) async throws -> [Value] + ) async throws -> [Value] { + // The lookup itself can draw a coordinator code, because a cluster where no group has + // ever committed has no __consumer_offsets topic to own one yet. Retrying here and per + // bucket, rather than around the whole loop, is what keeps a bucket that already + // answered from being asked twice and its results counted twice. + var plan: [(connection: KafkaConnection, groups: [String])] + do { + plan = try await groupsByCoordinator(groups) + } catch let error as KafkaError where Self.saysTheCoordinatorMoved(error) { + forgetCoordinators(of: groups) + plan = try await groupsByCoordinator(groups) + } + + var collected: [Value] = [] + for entry in plan { + do { + collected.append(contentsOf: try await body(entry.connection, entry.groups)) + } catch let error as KafkaError where Self.saysTheCoordinatorMoved(error) { + forgetCoordinators(of: entry.groups) + for retry in try await groupsByCoordinator(entry.groups) { + collected.append(contentsOf: try await body(retry.connection, retry.groups)) + } + } + } + return collected + } + + private static func saysTheCoordinatorMoved(_ error: KafkaError) -> Bool { + guard case .broker(let code, _) = error else { return false } + return KafkaErrorCode.retryAction(for: code) == .findCoordinatorAgain + } + + private func forgetCoordinators(of groups: [String]) { + for group in groups { coordinatorsByGroup[group] = nil } + } + + /// The groups each broker coordinates, for a request that names several. + /// + /// DescribeGroups is batched but coordinator-scoped, so a list spanning three coordinators + /// is three requests, not one, and asking any single broker for all of them answers + /// NOT_COORDINATOR per group for the ones it does not hold. + func groupsByCoordinator(_ groups: [String]) async throws -> [(connection: KafkaConnection, groups: [String])] { + var byNode: [Int32: [String]] = [:] + for group in Set(groups).sorted() { + let nodeId = try await coordinatorNode(for: group, refresh: false) + byNode[nodeId, default: []].append(group) + } + var plan: [(connection: KafkaConnection, groups: [String])] = [] + var slotForConnection: [ObjectIdentifier: Int] = [:] + for nodeId in byNode.keys.sorted() { + let held = byNode[nodeId] ?? [] + let connection = try await connection(forLeader: nodeId) + let identity = ObjectIdentifier(connection) + if let slot = slotForConnection[identity] { + plan[slot].groups.append(contentsOf: held) + } else { + slotForConnection[identity] = plan.count + plan.append((connection, held)) + } + } + return plan + } + + private func coordinatorNode(for group: String, refresh: Bool) async throws -> Int32 { + if !refresh, let cached = coordinatorsByGroup[group] { return cached } + let nodeId = try await KafkaFindCoordinatorRequest.coordinator( + forGroup: group, + on: try await controlConnection() + ) + coordinatorsByGroup[group] = nodeId + return nodeId + } + + private func coordinatorConnection(for group: String, refresh: Bool) async throws -> KafkaConnection { + if !refresh, let cached = coordinatorsByGroup[group] { + if let connection = try? await connection(forLeader: cached) { return connection } + coordinatorsByGroup[group] = nil + } + return try await connection(forLeader: try await coordinatorNode(for: group, refresh: true)) + } + + // MARK: - Cluster-wide sweeps + + /// Runs a request against every broker and collects what each one says. + /// + /// ListGroups is the reason this exists: a broker answers it with the groups it coordinates + /// and nothing else, with no error and no hint that the rest of the cluster holds more, so + /// a client that asks one broker reports a fraction of the groups as though it were all of + /// them. `reachedEveryBroker` is what lets the caller say the list is short rather than + /// present a partial answer as complete. + func withEveryBroker( + _ body: @Sendable @escaping (KafkaConnection) async throws -> Value + ) async throws -> (results: [Value], reachedEveryBroker: Bool) { + let reachable = try await everyBrokerConnection() + var collected: [Value] = [] + var failures: [KafkaError] = [] + + await withTaskGroup(of: Result.self) { group in + for connection in reachable.connections { + group.addTask { + do { + return .success(try await body(connection)) + } catch { + return .failure(error) + } + } + } + for await answer in group { + switch answer { + case .success(let value): + collected.append(value) + case .failure(let error): + // Kept, because when no broker answers this is the only thing that says + // why. Discarding it reported a missing group permission as a dead + // connection. + failures.append(error as? KafkaError ?? .connectionFailed(error.localizedDescription)) + } + } + } + guard !collected.isEmpty else { throw failures.first ?? KafkaError.notConnected } + return (collected, collected.count >= reachable.expected) + } +} diff --git a/Plugins/KafkaDriverPlugin/KafkaCluster.swift b/Plugins/KafkaDriverPlugin/KafkaCluster.swift index 027cafca8..12388ec94 100644 --- a/Plugins/KafkaDriverPlugin/KafkaCluster.swift +++ b/Plugins/KafkaDriverPlugin/KafkaCluster.swift @@ -1,8 +1,8 @@ import Foundation import NIOCore import NIOPosix -import TableProPluginKit import os +import TableProPluginKit /// Owns the connections to a cluster: the bootstrap dial, the per-broker pool, and the routing /// rule that decides whether a partition leader is reachable at its advertised address. @@ -13,20 +13,45 @@ import os /// `kafka-1.internal:9092` cannot be dialled at all. TablePro already solves the same problem /// for its other topology-aware drivers by pinning them (MongoDB to `directConnection`, Redis /// to `standalone`) whenever a tunnel rewrites the host, and Kafka joins that rule. +/// +/// Which broker a request goes to is a property of the request, not a preference. Kafka has +/// four answers and this type offers one primitive for each: any broker (`controlConnection`), +/// the partition leader (`withLeaders`), the group coordinator (`withCoordinator`), and every +/// broker at once (`withEveryBroker`). See `KafkaCluster+Routing.swift`. actor KafkaCluster { private static let logger = Logger(subsystem: "com.TablePro", category: "KafkaCluster") private let bootstrap: [KafkaEndpoint] private let ssl: SSLConfiguration private let credentials: KafkaCredentials - private let routing: KafkaBrokerRouting + let routing: KafkaBrokerRouting private let connectTimeout: TimeAmount private let group: EventLoopGroup private var connections: [KafkaEndpoint: KafkaConnection] = [:] + /// The dial in progress for an endpoint, so a second caller awaits the first instead of + /// opening a rival socket. `connection(forLeader:)` reads the pool, awaits `open()` and + /// only then writes the pool, and an actor releases its executor across that await: two + /// callers routing to the same leader both saw an empty pool, both dialled, and the second + /// overwrote the first's entry, leaking its socket. Fanning one request out per leader + /// makes that race routine rather than rare. + private var dialsInFlight: [KafkaEndpoint: Task] = [:] + /// Bumped by every disconnect. A dial suspends, so one that succeeds after the pool has been + /// drained would otherwise put its socket back into an emptied pool that nothing will ever + /// close, and its cleanup would remove a newer caller's entry. + private var poolGeneration = 0 + /// Why a broker could not be dialled, remembered so a six-partition browse does not pay the + /// connect timeout once per partition. Cleared whenever Metadata is re-read, because that is + /// when a broker's advertised address can have changed. + private var unreachableBrokers: [Int32: String] = [:] private var bootstrapConnection: KafkaConnection? private var cachedMetadata: KafkaClusterMetadata? private var brokersById: [Int32: KafkaBroker] = [:] + /// Partition to leader, per topic, for routing only. Deliberately separate from + /// `cachedMetadata`: six callers read `metadata(topics:)` expecting a live answer for + /// DESCRIBE TOPIC and the sidebar counts, and caching that call would change all of them. + var leadersByTopic: [String: [Int32: Int32]] = [:] + var coordinatorsByGroup: [String: Int32] = [:] init( bootstrap: [KafkaEndpoint], @@ -48,8 +73,14 @@ actor KafkaCluster { /// Dials the first bootstrap endpoint that answers. A cluster is usually given as several /// addresses precisely because any one of them may be down. + /// + /// A bootstrap connection that has since closed is re-dialled rather than kept. The guard + /// used to be `bootstrapConnection == nil`, so one cancelled statement closed the channel + /// (`send`'s cancellation handler does) and every later call threw `notConnected` for the + /// life of the session with nothing able to heal it. func connect() async throws { - guard bootstrapConnection == nil else { return } + if let bootstrapConnection, await bootstrapConnection.isOpen { return } + bootstrapConnection = nil var failures: [String] = [] for endpoint in bootstrap { try Task.checkCancellation() @@ -68,6 +99,9 @@ actor KafkaCluster { } func disconnect() async { + poolGeneration &+= 1 + for dial in dialsInFlight.values { dial.cancel() } + dialsInFlight.removeAll() for connection in connections.values { await connection.close() } @@ -75,95 +109,175 @@ actor KafkaCluster { bootstrapConnection = nil cachedMetadata = nil brokersById.removeAll() + leadersByTopic.removeAll() + coordinatorsByGroup.removeAll() + unreachableBrokers.removeAll() } func bootstrapEndpointDescription() -> String { bootstrap.map(\.description).joined(separator: ",") } - func controlConnection() throws -> KafkaConnection { + /// A connection to a broker that can answer a request addressed to no broker in particular: + /// Metadata, ApiVersions and FindCoordinator. + func controlConnection() async throws -> KafkaConnection { + if let bootstrapConnection, await bootstrapConnection.isOpen { return bootstrapConnection } + try await connect() guard let bootstrapConnection else { throw KafkaError.notConnected } return bootstrapConnection } /// The connection to the broker that is currently the controller. /// - /// Only the controller accepts an admin request such as DeleteTopics; any other broker answers - /// NOT_CONTROLLER. On a single-broker cluster this is the bootstrap connection. Under - /// `bootstrapOnly` it has to be, and a cluster whose controller is elsewhere will say so - /// rather than have the client dial an address it cannot reach. + /// A ZooKeeper-mode cluster accepts an admin request such as DeleteTopics only on the + /// controller and answers NOT_CONTROLLER anywhere else. A KRaft cluster forwards it + /// (KIP-590) and fills Metadata's `controllerId` with a randomly chosen live broker, so this + /// dials an arbitrary broker there and it does not matter. The silent fall back to the + /// bootstrap connection is kept for that reason, and only here: an admin request reaching + /// the "wrong" broker is answered on every cluster this driver supports, which is not true + /// of a partition read. func controllerConnection() async throws -> KafkaConnection { let metadata = try await metadata() - guard metadata.controllerId >= 0 else { return try controlConnection() } - return try await connection(forLeader: metadata.controllerId) + guard metadata.controllerId >= 0 else { return try await controlConnection() } + do { + return try await connection(forLeader: metadata.controllerId) + } catch { + return try await controlConnection() + } } /// The connection to use for a partition whose leader is `nodeId`. /// - /// Under `bootstrapOnly` this always returns the bootstrap connection. That can mean - /// sending a Fetch to a broker that does not lead the partition, which answers - /// NOT_LEADER_OR_FOLLOWER, and that is the honest outcome: the alternative is dialling an - /// address that cannot be reached and reporting a timeout instead. + /// Under `bootstrapOnly` this always returns the bootstrap connection, which will answer + /// NOT_LEADER_OR_FOLLOWER for a partition it does not lead. The caller turns that into an + /// error naming the setting rather than the error code, because "this broker no longer leads + /// the partition" tells a user nothing they can act on. + /// + /// An advertised address that cannot be dialled is reported. It used to fall back to the + /// bootstrap connection with a log line, which guaranteed NOT_LEADER_OR_FOLLOWER and blamed + /// leadership for what was a reachability problem. func connection(forLeader nodeId: Int32) async throws -> KafkaConnection { - guard routing == .advertised else { return try controlConnection() } - guard let broker = brokersById[nodeId] else { return try controlConnection() } + guard routing == .advertised else { return try await controlConnection() } + guard let broker = brokersById[nodeId] else { return try await controlConnection() } + if let reason = unreachableBrokers[nodeId] { + throw KafkaError.brokerUnreachable(nodeId: nodeId, address: broker.endpoint.description, reason: reason) + } if let existing = connections[broker.endpoint], await existing.isOpen { return existing } - let connection = KafkaConnection(endpoint: broker.endpoint, clientId: KafkaClientInfo.clientId) do { - try await connection.open(ssl: ssl, credentials: credentials, group: group, timeout: connectTimeout) + return try await dial(broker.endpoint) } catch { - await connection.close() + let reason = (error as? LocalizedError)?.errorDescription ?? error.localizedDescription + unreachableBrokers[nodeId] = reason Self.logger.warning(""" - Broker \(nodeId, privacy: .public) advertises an address this client cannot reach; \ - falling back to the bootstrap connection + Broker \(nodeId, privacy: .public) advertises an address this client cannot reach """) - return try controlConnection() + throw KafkaError.brokerUnreachable(nodeId: nodeId, address: broker.endpoint.description, reason: reason) } - connections[broker.endpoint] = connection + } + + /// Every broker the cluster reports, for a request each broker answers only for itself. + /// + /// Brokers that cannot be dialled are skipped rather than failing the sweep, and the count + /// that came back is returned so the caller can say the answer is partial. Under + /// `bootstrapOnly` that is one broker out of however many the cluster has. + func everyBrokerConnection() async throws -> (connections: [KafkaConnection], expected: Int) { + let metadata = try await metadata() + let expected = max(1, metadata.brokers.count) + guard routing == .advertised else { + return ([try await controlConnection()], expected) + } + // Deduplicated by connection identity, not by broker id. Several brokers can advertise + // one address, and asking the same socket three times and counting three answers is how + // a partial sweep would report itself as complete, which is the defect this exists to + // fix rather than repeat. + var reachable: [KafkaConnection] = [] + var seen: Set = [] + for broker in metadata.brokers.sorted(by: { $0.nodeId < $1.nodeId }) { + guard let connection = try? await connection(forLeader: broker.nodeId) else { continue } + guard seen.insert(ObjectIdentifier(connection)).inserted else { continue } + reachable.append(connection) + } + if reachable.isEmpty { reachable = [try await controlConnection()] } + return (reachable, expected) + } + + private func dial(_ endpoint: KafkaEndpoint) async throws -> KafkaConnection { + let generation = poolGeneration + if let running = dialsInFlight[endpoint] { + let connection = try await running.value + return try await install(connection, at: endpoint, from: generation) + } + let ssl = ssl + let credentials = credentials + let group = group + let timeout = connectTimeout + let dial = Task { () async throws -> KafkaConnection in + let connection = KafkaConnection(endpoint: endpoint, clientId: KafkaClientInfo.clientId) + do { + try await connection.open(ssl: ssl, credentials: credentials, group: group, timeout: timeout) + } catch { + await connection.close() + throw error + } + return connection + } + dialsInFlight[endpoint] = dial + let connection: KafkaConnection + do { + connection = try await dial.value + } catch { + if poolGeneration == generation { dialsInFlight[endpoint] = nil } + throw error + } + if poolGeneration == generation { dialsInFlight[endpoint] = nil } + return try await install(connection, at: endpoint, from: generation) + } + + /// Puts a freshly dialled connection into the pool, unless the pool moved on while it was + /// being dialled. A connection nobody will own is closed here rather than leaked. + private func install( + _ connection: KafkaConnection, + at endpoint: KafkaEndpoint, + from generation: Int + ) async throws -> KafkaConnection { + guard poolGeneration == generation else { + await connection.close() + throw KafkaError.notConnected + } + connections[endpoint] = connection return connection } func metadata(topics: [String]? = nil, refresh: Bool = false) async throws -> KafkaClusterMetadata { if !refresh, let cachedMetadata, topics == nil { return cachedMetadata } - let connection = try controlConnection() + let connection = try await controlConnection() let fetched = try await KafkaMetadataRequest.fetch(topics: topics, on: connection) - if topics == nil { - cachedMetadata = fetched - } - for broker in fetched.brokers { brokersById[broker.nodeId] = broker } + adopt(fetched, cacheAll: topics == nil) return fetched } - func invalidateMetadata() { - cachedMetadata = nil - } - - /// Runs a request against the leader of a partition, refreshing metadata and retrying once - /// when the cluster says the leadership moved. One retry, because a second failure is a - /// real answer rather than a race. - func withLeader( - of partition: Int32, - topic: String, - _ body: (KafkaConnection) async throws -> T - ) async throws -> T { - let leader = try await leaderNode(topic: topic, partition: partition) - do { - return try await body(try await connection(forLeader: leader)) - } catch let error as KafkaError { - guard case .broker(let code, _) = error, KafkaErrorCode.requiresMetadataRefresh(code) else { throw error } - invalidateMetadata() - let refreshed = try await leaderNode(topic: topic, partition: partition, refresh: true) - return try await body(try await connection(forLeader: refreshed)) + /// Records what a Metadata reply says about the cluster's shape. + /// + /// A fresh reply is also the only moment a broker's advertised address can have changed, so + /// it is where the unreachable list is cleared: a broker that moves to an address this client + /// can reach must not stay marked from the old one. + func adopt(_ metadata: KafkaClusterMetadata, cacheAll: Bool) { + if cacheAll { cachedMetadata = metadata } + for broker in metadata.brokers { brokersById[broker.nodeId] = broker } + for topic in metadata.topics where !topic.name.isEmpty { + leadersByTopic[topic.name] = Dictionary( + topic.partitions.map { ($0.index, $0.leader) }, + uniquingKeysWith: { first, _ in first } + ) } + unreachableBrokers.removeAll() } - private func leaderNode(topic: String, partition: Int32, refresh: Bool = false) async throws -> Int32 { - let meta = try await metadata(topics: [topic], refresh: refresh) - let found = try meta.requireTopic(named: topic) - guard let match = found.partitions.first(where: { $0.index == partition }) else { - throw KafkaError.producedToUnknownPartition(topic: topic, partition: partition) - } - return match.leader + func invalidateMetadata() { + cachedMetadata = nil + leadersByTopic.removeAll() + coordinatorsByGroup.removeAll() + unreachableBrokers.removeAll() } } diff --git a/Plugins/KafkaDriverPlugin/KafkaConnection.swift b/Plugins/KafkaDriverPlugin/KafkaConnection.swift index 6ea82d062..a424e91ec 100644 --- a/Plugins/KafkaDriverPlugin/KafkaConnection.swift +++ b/Plugins/KafkaDriverPlugin/KafkaConnection.swift @@ -12,7 +12,7 @@ import TableProPluginKit private final class KafkaFrameDecoder: ByteToMessageDecoder { typealias InboundOut = ByteBuffer - private static let maximumFrameLength = 256 * 1024 * 1024 + private static let maximumFrameLength = 256 * 1_024 * 1_024 func decode(context: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState { guard buffer.readableBytes >= 4 else { return .needMoreData } @@ -42,10 +42,15 @@ private final class KafkaResponseHandler: ChannelInboundHandler, @unchecked Send private var failure: Error? private let lock = NIOLock() - /// Registers the one in-flight request. A second registration is refused rather than - /// allowed to overwrite the first: dropping a continuation leaks it and hangs its caller - /// forever, which is far harder to diagnose than an error naming the overlap. - func expect(_ continuation: CheckedContinuation) { + /// Registers the one in-flight request, and says whether the caller may now write. + /// + /// A second registration is refused rather than allowed to overwrite the first: dropping a + /// continuation leaks it and hangs its caller forever, which is far harder to diagnose than + /// an error naming the overlap. The return value is what stops the refusal being worse than + /// the overlap: the caller used to be resumed with the error and then write its request + /// anyway, so the broker answered a request nobody was waiting for and the orphan frame + /// resumed the NEXT caller, failing it on a correlation id mismatch. + func expect(_ continuation: CheckedContinuation) -> Bool { let resolved: Error? = lock.withLock { if let failure { return failure } guard pending == nil else { @@ -56,7 +61,9 @@ private final class KafkaResponseHandler: ChannelInboundHandler, @unchecked Send pending = continuation return nil } - if let resolved { continuation.resume(throwing: resolved) } + guard let resolved else { return true } + continuation.resume(throwing: resolved) + return false } func channelRead(context: ChannelHandlerContext, data: NIOAny) { @@ -141,6 +148,8 @@ actor KafkaConnection { private var channel: Channel? private var handler: KafkaResponseHandler? private var correlationId: Int32 = 0 + private var wireSlotIsTaken = false + private var waitingForWireSlot: [CheckedContinuation] = [] private(set) var apiVersions: KafkaApiVersionTable = .preNegotiation let endpoint: KafkaEndpoint @@ -248,7 +257,17 @@ actor KafkaConnection { } /// Sends one request and reads its reply. + /// + /// Requests queue rather than collide. An actor releases its executor across every `await`, + /// so two callers that both reach this method interleave, and the wire slot the response + /// handler guards holds exactly one of them. The queue is what makes a connection shared by + /// the health monitor's 30-second ping and a running statement work at all; without it + /// whichever arrived second failed with "two requests overlapped", and a fan-out across + /// leaders makes a shared connection far more common than it used to be. func send(_ request: KafkaRequest) async throws -> KafkaProtocolReader { + await claimWireSlot() + defer { releaseWireSlot() } + guard let channel, let handler, channel.isActive else { throw KafkaError.notConnected } try Task.checkCancellation() @@ -260,7 +279,7 @@ actor KafkaConnection { let frame = try await withTaskCancellationHandler { try await withCheckedThrowingContinuation { continuation in - handler.expect(continuation) + guard handler.expect(continuation) else { return } channel.writeAndFlush(buffer).whenFailure { error in handler.finish(with: KafkaError.connectionFailed(error.localizedDescription)) } @@ -277,6 +296,29 @@ actor KafkaConnection { return body } + /// Waits for the wire slot. + /// + /// Deliberately not cancellable. A waiter that unwound on cancellation would have to resume + /// itself out of a nonisolated handler and race the holder's release, and the wait it would + /// be escaping is already bounded: the request ahead ends when the broker answers or when + /// its own cancellation closes the channel. A cancelled waiter takes the slot and gives it + /// straight back at the `Task.checkCancellation()` below. + private func claimWireSlot() async { + guard wireSlotIsTaken else { + wireSlotIsTaken = true + return + } + await withCheckedContinuation { waitingForWireSlot.append($0) } + } + + private func releaseWireSlot() { + guard !waitingForWireSlot.isEmpty else { + wireSlotIsTaken = false + return + } + waitingForWireSlot.removeFirst().resume() + } + /// Asks the broker what it supports, then never sends a version outside that range. /// /// A broker that rejects even v3 answers UNSUPPORTED_VERSION with a v0-shaped body, so the diff --git a/Plugins/KafkaDriverPlugin/KafkaError.swift b/Plugins/KafkaDriverPlugin/KafkaError.swift index b46baa620..a88efca1b 100644 --- a/Plugins/KafkaDriverPlugin/KafkaError.swift +++ b/Plugins/KafkaDriverPlugin/KafkaError.swift @@ -13,7 +13,14 @@ enum KafkaError: LocalizedError { case decompressionFailed(codec: String, reason: String) case syntax(String) case unknownTopic(String) + case unknownGroup(String) case producedToUnknownPartition(topic: String, partition: Int32) + case unknownPartitions(topic: String, partitions: [Int32], available: [Int32]) + case partitionsRejected(topic: String, partitions: [Int32], api: String, code: Int16) + case partitionsUnanswered(topic: String, partitions: [Int32], api: String) + case brokerUnreachable(nodeId: Int32, address: String, reason: String) + case partitionsLedElsewhere(topic: String, partitions: [Int32]) + case partitionsHaveNoLeader(topic: String, partitions: [Int32]) var errorDescription: String? { switch self { @@ -66,17 +73,86 @@ enum KafkaError: LocalizedError { return detail case .unknownTopic(let name): return String(format: String(localized: "No topic named %@ on this cluster."), name) + case .unknownGroup(let name): + return String(format: String(localized: "No consumer group named %@ on this cluster."), name) case .producedToUnknownPartition(let topic, let partition): return String( format: String(localized: "Topic %@ has no partition %d."), topic, Int(partition) ) + case .unknownPartitions(let topic, let partitions, let available): + return String( + format: String(localized: "Topic %@ has no partition %@. It has %@."), + topic, + KafkaPartitionList.describe(partitions), + KafkaPartitionList.describe(available) + ) + case .partitionsRejected(let topic, let partitions, let api, let code): + return String( + format: String(localized: "The broker rejected %@ for partition %@ of %@: %@"), + api, + KafkaPartitionList.describe(partitions), + topic, + KafkaErrorCode.describe(code) + ) + case .partitionsUnanswered(let topic, let partitions, let api): + return String( + format: String(localized: "The broker's %@ reply left out partition %@ of %@."), + api, + KafkaPartitionList.describe(partitions), + topic + ) + case .brokerUnreachable(let nodeId, let address, let reason): + return String( + format: String(localized: "Broker %d advertises %@, which could not be reached: %@"), + Int(nodeId), + address, + reason + ) + case .partitionsLedElsewhere(let topic, let partitions): + return String( + format: String(localized: """ + Partition %@ of %@ is led by another broker, and Broker Addresses is set to use only the \ + bootstrap address. + """), + KafkaPartitionList.describe(partitions), + topic + ) + case .partitionsHaveNoLeader(let topic, let partitions): + return String( + format: String(localized: "Partition %@ of %@ has no elected leader."), + KafkaPartitionList.describe(partitions), + topic + ) } } } -/// The subset of Kafka's error codes this client can meet, with the retryable ones marked. +/// Renders a partition set for a message. +/// +/// A routing failure is about a set of partitions rather than one, and "partitions 1, 4, 5" is +/// what tells a user the request was split and which part of it failed. Sorted, because the +/// set arrives from a dictionary and an error that reorders itself between runs reads as noise. +enum KafkaPartitionList { + static func describe(_ partitions: [Int32]) -> String { + partitions.sorted().map(String.init).joined(separator: ", ") + } +} + +/// What to do about an error code before giving it to the user. +enum KafkaRetryAction: Sendable { + /// A real answer. Report it. + case report + /// The client's view of the cluster is stale. Re-read Metadata, re-resolve the leader, retry. + case resolveLeaderAgain + /// The client's view of the group's coordinator is stale. Re-run FindCoordinator, retry. + case findCoordinatorAgain + /// Transient on the broker that answered. Retry the same request there. + case retrySameBroker +} + +/// The subset of Kafka's error codes this client can meet, with what to do about each. /// A code that is not listed still reports its number rather than being swallowed. enum KafkaErrorCode { static let none: Int16 = 0 @@ -85,29 +161,67 @@ enum KafkaErrorCode { static let leaderNotAvailable: Int16 = 5 static let notLeaderOrFollower: Int16 = 6 static let requestTimedOut: Int16 = 7 + static let brokerNotAvailable: Int16 = 8 + static let replicaNotAvailable: Int16 = 9 static let messageTooLarge: Int16 = 10 + static let networkException: Int16 = 13 static let coordinatorLoadInProgress: Int16 = 14 static let coordinatorNotAvailable: Int16 = 15 static let notCoordinator: Int16 = 16 static let illegalSaslState: Int16 = 34 static let unsupportedVersion: Int16 = 35 static let notController: Int16 = 41 - static let topicDeletionDisabled: Int16 = 72 static let topicAuthorizationFailed: Int16 = 29 static let groupAuthorizationFailed: Int16 = 30 static let clusterAuthorizationFailed: Int16 = 31 + static let kafkaStorageError: Int16 = 56 static let saslAuthenticationFailed: Int16 = 58 - static let unknownTopicId: Int16 = 100 + /// 72 is LISTENER_NOT_FOUND and 73 is TOPIC_DELETION_DISABLED, which is the opposite of what + /// this file said until #2993. Nothing read the constant, so nothing broke; the two are + /// spelled out here because a hand-transcribed code table is exactly the thing that drifts. + static let listenerNotFound: Int16 = 72 + static let topicDeletionDisabled: Int16 = 73 static let fencedLeaderEpoch: Int16 = 74 static let unknownLeaderEpoch: Int16 = 75 + static let offsetNotAvailable: Int16 = 78 + static let unknownTopicId: Int16 = 100 + + /// What a code means for the request that drew it. + /// + /// Three outcomes rather than one boolean, because "retry" alone is not an instruction: a + /// moved leader needs a fresh Metadata read before the retry can go anywhere different, a + /// moved coordinator needs a fresh FindCoordinator, and a transient failure needs neither + /// and would only be delayed by them. 8 BROKER_NOT_AVAILABLE is deliberately not retried: + /// Kafka does not classify it retriable, and a client that retries it hides a broker that + /// is genuinely down. + static func retryAction(for code: Int16) -> KafkaRetryAction { + switch code { + case unknownTopicOrPartition, leaderNotAvailable, notLeaderOrFollower, replicaNotAvailable, + networkException, kafkaStorageError, listenerNotFound, fencedLeaderEpoch, + unknownLeaderEpoch, unknownTopicId: + return .resolveLeaderAgain + case coordinatorLoadInProgress, coordinatorNotAvailable, notCoordinator: + return .findCoordinatorAgain + case requestTimedOut, offsetNotAvailable: + return .retrySameBroker + default: + return .report + } + } - /// The codes that mean the cluster moved rather than that the request was wrong. These are - /// the only ones worth one metadata refresh and a retry; everything else is a real answer, - /// and retrying it just delays the error the user needs to see. - static func requiresMetadataRefresh(_ code: Int16) -> Bool { + /// True when the code proves the broker did not take the request, so sending it again + /// cannot repeat work the broker already did. + /// + /// The distinction only matters for a write. `KafkaProduceRequest` asks for + /// `acks = -1`, and REQUEST_TIMED_OUT there means the leader appended the record and the + /// in-sync replicas did not acknowledge in time; the same is true of a connection that + /// broke and of a log directory that failed mid-append. This client sends no producer id, + /// so Kafka cannot deduplicate a replay and the message is appended twice. The codes below + /// all mean the broker refused the request outright, which is safe to send elsewhere. + static func provesRequestWasNotApplied(_ code: Int16) -> Bool { switch code { - case leaderNotAvailable, notLeaderOrFollower, unknownTopicOrPartition, - unknownTopicId, fencedLeaderEpoch, unknownLeaderEpoch: + case unknownTopicOrPartition, leaderNotAvailable, notLeaderOrFollower, replicaNotAvailable, + listenerNotFound, fencedLeaderEpoch, unknownLeaderEpoch, unknownTopicId: return true default: return false @@ -126,8 +240,14 @@ enum KafkaErrorCode { return String(localized: "this broker no longer leads the partition") case requestTimedOut: return String(localized: "the request timed out on the broker") + case brokerNotAvailable: + return String(localized: "the broker is not available") + case replicaNotAvailable: + return String(localized: "the replica is not available on this broker") case messageTooLarge: return String(localized: "the message is larger than the broker accepts") + case networkException: + return String(localized: "the connection to the broker broke") case coordinatorLoadInProgress: return String(localized: "the group coordinator is still loading") case coordinatorNotAvailable: @@ -144,8 +264,22 @@ enum KafkaErrorCode { return String(localized: "not authorized for this consumer group") case clusterAuthorizationFailed: return String(localized: "not authorized for this cluster operation") + case notController: + return String(localized: "this broker is not the cluster controller") + case kafkaStorageError: + return String(localized: "the broker could not read the partition's log directory") case saslAuthenticationFailed: return String(localized: "the credentials were rejected") + case listenerNotFound: + return String(localized: "the broker has no listener for this connection's security protocol") + case topicDeletionDisabled: + return String(localized: "topic deletion is disabled on this cluster") + case fencedLeaderEpoch: + return String(localized: "the partition's leader epoch has moved on") + case unknownLeaderEpoch: + return String(localized: "the broker has not caught up to the partition's leader epoch") + case offsetNotAvailable: + return String(localized: "the offset is not available on this broker yet") case unknownTopicId: return String(localized: "unknown topic id") default: diff --git a/Plugins/KafkaDriverPlugin/KafkaFetchRequest.swift b/Plugins/KafkaDriverPlugin/KafkaFetchRequest.swift index 269882b77..9af9972d8 100644 --- a/Plugins/KafkaDriverPlugin/KafkaFetchRequest.swift +++ b/Plugins/KafkaDriverPlugin/KafkaFetchRequest.swift @@ -15,8 +15,8 @@ enum KafkaFetchRequest { /// a rolled-back transaction is worse than useless, so this client always reads committed. private static let readCommitted: Int8 = 1 - private static let defaultPartitionBudget: Int32 = 1 * 1024 * 1024 - private static let maximumPartitionBudget: Int32 = 64 * 1024 * 1024 + private static let defaultPartitionBudget: Int32 = 1 * 1_024 * 1_024 + private static let maximumPartitionBudget: Int32 = 64 * 1_024 * 1_024 /// Fetches forward from `startOffset` in one partition. /// @@ -82,7 +82,7 @@ enum KafkaFetchRequest { budget: Int32, cluster: KafkaCluster ) async throws -> KafkaFetchResult { - try await cluster.withLeader(of: partition, topic: topic) { connection in + try await cluster.withLeader(of: partition, topic: topic, api: "Fetch") { connection in let version = try await connection.negotiatedVersion(for: .fetch) let flexible = KafkaApiKey.fetch.isFlexible(version: version) diff --git a/Plugins/KafkaDriverPlugin/KafkaFindCoordinatorRequest.swift b/Plugins/KafkaDriverPlugin/KafkaFindCoordinatorRequest.swift new file mode 100644 index 000000000..12ef4350c --- /dev/null +++ b/Plugins/KafkaDriverPlugin/KafkaFindCoordinatorRequest.swift @@ -0,0 +1,76 @@ +import Foundation + +/// Asks any broker which broker coordinates a consumer group. +/// +/// Every broker answers this one, which is what makes it the entry point to coordinator +/// routing: the client has no way to work out the coordinator itself, because it is the broker +/// owning the `__consumer_offsets` partition the group id hashes to, and the partition count of +/// that topic is not something a client is told. +/// +/// The response field order changed at v4 and it is not guessable from the versions either side +/// of it. Up to v3 the body opens with the error, then names the node; v4 (KIP-699) made the +/// request batched and 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 a port out of the +/// host's length prefix, which yields a plausible broker id rather than a parse failure. Both +/// orders below were measured against a live Kafka 4.3.1 broker. +enum KafkaFindCoordinatorRequest { + /// `keyType` 0 is a consumer group. 1 is a transaction coordinator, which this driver does + /// not use: it neither produces transactionally nor reads transaction state. + private static let groupKeyType: Int8 = 0 + + static func coordinator(forGroup group: String, on connection: KafkaConnection) async throws -> Int32 { + let version = try await connection.negotiatedVersion(for: .findCoordinator) + let flexible = KafkaApiKey.findCoordinator.isFlexible(version: version) + + let request = KafkaRequest(api: .findCoordinator, version: version) { writer, _ in + if version >= 4 { + writer.int8(groupKeyType) + writer.compactArrayCount(1) + writer.compactString(group) + writer.emptyTaggedFields() + return + } + if flexible { + writer.compactString(group) + } else { + writer.legacyString(group) + } + if version >= 1 { writer.int8(groupKeyType) } + if flexible { writer.emptyTaggedFields() } + } + + var body = try await connection.send(request) + if version >= 1 { _ = try body.int32() } // throttleTimeMs + + if version >= 4 { + let coordinators = try body.array(compact: true) { reader -> (String, Int32, Int16) in + let key = try reader.compactString() + let nodeId = try reader.int32() + _ = try reader.compactString() // host + _ = try reader.int32() // port + let errorCode = try reader.int16() + _ = try reader.nullableCompactString() // errorMessage + try reader.taggedFields() + return (key, nodeId, errorCode) + } + guard let entry = coordinators.first(where: { $0.0 == group }) ?? coordinators.first else { + throw KafkaError.malformedResponse( + String(localized: "The broker answered a coordinator lookup with no coordinator.") + ) + } + try KafkaErrorCode.check(entry.2, api: KafkaApiKey.findCoordinator.name) + return entry.1 + } + + let errorCode = try body.int16() + if version >= 1 { + _ = flexible ? try body.nullableCompactString() : try body.nullableLegacyString() + } + try KafkaErrorCode.check(errorCode, api: KafkaApiKey.findCoordinator.name) + let nodeId = try body.int32() + _ = flexible ? try body.compactString() : try body.legacyString() // host + _ = try body.int32() // port + if flexible { try body.taggedFields() } + return nodeId + } +} diff --git a/Plugins/KafkaDriverPlugin/KafkaGroupsRequest.swift b/Plugins/KafkaDriverPlugin/KafkaGroupsRequest.swift index 8df1680d6..21f804dd1 100644 --- a/Plugins/KafkaDriverPlugin/KafkaGroupsRequest.swift +++ b/Plugins/KafkaDriverPlugin/KafkaGroupsRequest.swift @@ -6,6 +6,13 @@ struct KafkaGroupSummary: Sendable { let state: String } +struct KafkaGroupListing: Sendable { + let groups: [KafkaGroupSummary] + /// False when at least one broker could not be asked. A group list assembled from some of + /// the cluster's brokers is missing whole groups rather than being merely out of date. + let isComplete: Bool +} + struct KafkaGroupMember: Sendable { let memberId: String let clientId: String @@ -18,6 +25,11 @@ struct KafkaGroupDetail: Sendable { let protocolType: String let assignmentProtocol: String let members: [KafkaGroupMember] + + /// Kafka answers a group it has never heard of with state "Dead" and no members rather + /// than with an error, so this is the only thing that separates a typo from a group that + /// simply has not committed anything yet. + var isKnown: Bool { state.caseInsensitiveCompare("Dead") != .orderedSame } } struct KafkaGroupOffset: Sendable { @@ -28,9 +40,31 @@ struct KafkaGroupOffset: Sendable { /// The consumer-group side of the cluster, which is what makes Kafka debuggable: a group's lag /// is the gap between what has been written and what the group has acknowledged. +/// +/// Every request here is coordinator-scoped. A group's state and committed offsets live on the +/// broker that owns the `__consumer_offsets` partition its id hashes to, and no other broker +/// will answer for it or forward the question: measured on a three-broker cluster, a +/// non-coordinator answers OffsetFetch with a group-level NOT_COORDINATOR and DescribeGroups +/// with NOT_COORDINATOR per group. ListGroups is worse than that, because it succeeds: a broker +/// lists the groups it coordinates and says nothing about the rest, so asking one broker +/// reports a third of a three-broker cluster's groups as though that were all of them. enum KafkaGroupsRequest { - static func listGroups(cluster: KafkaCluster) async throws -> [KafkaGroupSummary] { - let connection = try await cluster.controlConnection() + /// Every group on the cluster, by asking every broker. + static func listGroups(cluster: KafkaCluster) async throws -> KafkaGroupListing { + let swept = try await cluster.withEveryBroker { connection in + try await listGroups(on: connection) + } + var byId: [String: KafkaGroupSummary] = [:] + for summary in swept.results.flatMap({ $0 }) { + byId[summary.groupId] = summary + } + return KafkaGroupListing( + groups: byId.values.sorted { $0.groupId < $1.groupId }, + isComplete: swept.reachedEveryBroker + ) + } + + private static func listGroups(on connection: KafkaConnection) async throws -> [KafkaGroupSummary] { let version = try await connection.negotiatedVersion(for: .listGroups) let flexible = KafkaApiKey.listGroups.isFlexible(version: version) @@ -62,9 +96,19 @@ enum KafkaGroupsRequest { } } + /// A group's state and its members, asked of each group's coordinator. static func describeGroups(_ groupIds: [String], cluster: KafkaCluster) async throws -> [KafkaGroupDetail] { guard !groupIds.isEmpty else { return [] } - let connection = try await cluster.controlConnection() + let details = try await cluster.withCoordinators(of: groupIds) { connection, held in + try await describeGroups(held, on: connection) + } + return details.sorted { $0.groupId < $1.groupId } + } + + private static func describeGroups( + _ groupIds: [String], + on connection: KafkaConnection + ) async throws -> [KafkaGroupDetail] { let version = try await connection.negotiatedVersion(for: .describeGroups) let flexible = KafkaApiKey.describeGroups.isFlexible(version: version) @@ -83,7 +127,10 @@ enum KafkaGroupsRequest { var body = try await connection.send(request) if version >= 1 { _ = try body.int32() } // throttleTimeMs - return try body.array(compact: flexible) { reader -> KafkaGroupDetail in + // Parsed in full before any error code is acted on. Throwing from inside the array + // closure abandons the reader mid-reply, so one group's NOT_COORDINATOR used to discard + // every group already read beside it. + let parsed = try body.array(compact: flexible) { reader -> (KafkaGroupDetail, Int16) in let errorCode = try reader.int16() let groupId = flexible ? try reader.compactString() : try reader.legacyString() let state = flexible ? try reader.compactString() : try reader.legacyString() @@ -105,22 +152,35 @@ enum KafkaGroupsRequest { } if version >= 3 { _ = try reader.int32() } // authorizedOperations if flexible { try reader.taggedFields() } - try KafkaErrorCode.check(errorCode, api: "DescribeGroups") - return KafkaGroupDetail( - groupId: groupId, - state: state, - protocolType: protocolType, - assignmentProtocol: assignmentProtocol, - members: members + return ( + KafkaGroupDetail( + groupId: groupId, + state: state, + protocolType: protocolType, + assignmentProtocol: assignmentProtocol, + members: members + ), + errorCode ) } + + if let rejected = parsed.first(where: { $0.1 != KafkaErrorCode.none }) { + try KafkaErrorCode.check(rejected.1, api: "DescribeGroups") + } + return parsed.map(\.0) } - /// A group's committed offsets. These live on the group's coordinator rather than on any - /// partition leader, which is why FindCoordinator exists; the bootstrap broker answers it - /// correctly on a single-node cluster and forwards otherwise. + /// A group's committed offsets, asked of its coordinator. static func fetchCommittedOffsets(group: String, cluster: KafkaCluster) async throws -> [KafkaGroupOffset] { - let connection = try await cluster.controlConnection() + try await cluster.withCoordinator(of: group) { connection in + try await fetchCommittedOffsets(group: group, on: connection) + } + } + + private static func fetchCommittedOffsets( + group: String, + on connection: KafkaConnection + ) async throws -> [KafkaGroupOffset] { let version = try await connection.negotiatedVersion(for: .offsetFetch) let flexible = KafkaApiKey.offsetFetch.isFlexible(version: version) @@ -150,33 +210,44 @@ enum KafkaGroupsRequest { if version >= 3 { _ = try body.int32() } // throttleTimeMs if version >= 8 { - let groups = try body.array(compact: true) { reader -> [KafkaGroupOffset] in + let groups = try body.array(compact: true) { reader -> ([KafkaGroupOffset], [Int16]) in _ = try reader.compactString() // groupId - let offsets = try readTopics(&reader, version: version, flexible: true) + let read = try readTopics(&reader, version: version, flexible: true) let errorCode = try reader.int16() try reader.taggedFields() - try KafkaErrorCode.check(errorCode, api: "OffsetFetch") - return offsets + return (read.offsets, read.errorCodes + [errorCode]) } - return groups.flatMap { $0 } + try reportFirstFailure(groups.flatMap(\.1)) + return groups.flatMap(\.0) } - let offsets = try readTopics(&body, version: version, flexible: flexible) - if version >= 2 { - let errorCode = try body.int16() - try KafkaErrorCode.check(errorCode, api: "OffsetFetch") - } - return offsets + let read = try readTopics(&body, version: version, flexible: flexible) + var codes = read.errorCodes + if version >= 2 { codes.append(try body.int16()) } + try reportFirstFailure(codes) + return read.offsets + } + + private static func reportFirstFailure(_ codes: [Int16]) throws { + guard let failed = codes.first(where: { $0 != KafkaErrorCode.none }) else { return } + try KafkaErrorCode.check(failed, api: "OffsetFetch") } + /// Reads the topic list and carries every partition's error code back rather than throwing + /// mid-parse. + /// + /// The reader has to reach the end of the reply either way: throwing from inside the array + /// closure abandons it part-read, and at v8 the group's own error code sits AFTER its + /// topics, so the caller would never see it. A partition that did report an error still + /// fails the call, at `reportFirstFailure`, once everything has been read. private static func readTopics( _ reader: inout KafkaProtocolReader, version: Int16, flexible: Bool - ) throws -> [KafkaGroupOffset] { - let topics = try reader.array(compact: flexible) { topicReader -> [KafkaGroupOffset] in + ) throws -> (offsets: [KafkaGroupOffset], errorCodes: [Int16]) { + let topics = try reader.array(compact: flexible) { topicReader -> [(KafkaGroupOffset, Int16)] in let name = flexible ? try topicReader.compactString() : try topicReader.legacyString() - let partitions = try topicReader.array(compact: flexible) { partitionReader -> KafkaGroupOffset in + let partitions = try topicReader.array(compact: flexible) { partitionReader -> (KafkaGroupOffset, Int16) in let index = try partitionReader.int32() let committed = try partitionReader.int64() if version >= 5 { _ = try partitionReader.int32() } // committedLeaderEpoch @@ -185,12 +256,15 @@ enum KafkaGroupsRequest { : try partitionReader.nullableLegacyString() // metadata let errorCode = try partitionReader.int16() if flexible { try partitionReader.taggedFields() } - try KafkaErrorCode.check(errorCode, api: "OffsetFetch") - return KafkaGroupOffset(topic: name, partition: index, committedOffset: committed) + return ( + KafkaGroupOffset(topic: name, partition: index, committedOffset: committed), + errorCode + ) } if flexible { try topicReader.taggedFields() } return partitions } - return topics.flatMap { $0 } + let flat = topics.flatMap { $0 } + return (flat.filter { $0.1 == KafkaErrorCode.none }.map(\.0), flat.map(\.1)) } } diff --git a/Plugins/KafkaDriverPlugin/KafkaOffsetsRequest.swift b/Plugins/KafkaDriverPlugin/KafkaOffsetsRequest.swift index 40e2fb7d8..edc8989e6 100644 --- a/Plugins/KafkaDriverPlugin/KafkaOffsetsRequest.swift +++ b/Plugins/KafkaDriverPlugin/KafkaOffsetsRequest.swift @@ -15,6 +15,15 @@ enum KafkaOffsetsRequest { static let earliestTimestamp: Int64 = -2 static let latestTimestamp: Int64 = -1 + static let api = "ListOffsets" + + /// Each partition's offset, asked of each partition's leader. + /// + /// The routing is the whole point. Only the leader of a partition can answer for it: every + /// other broker replies NOT_LEADER_OR_FOLLOWER for that partition, inside an otherwise + /// successful response, and forwards nothing. Sending one request carrying every partition + /// to the bootstrap broker therefore worked on a single-broker cluster and failed on every + /// other one, which is #2993. static func listOffsets( topic: String, partitions: [Int32], @@ -22,7 +31,69 @@ enum KafkaOffsetsRequest { cluster: KafkaCluster ) async throws -> [Int32: Int64] { guard !partitions.isEmpty else { return [:] } - let connection = try await cluster.controlConnection() + let outcomes = try await cluster.withLeaders( + topic: topic, + partitions: partitions + ) { connection, slice in + try await send(topic: topic, partitions: slice, timestamp: timestamp, on: connection) + } + return try KafkaPartitionOutcome.requireAll( + outcomes, + topic: topic, + api: api, + routing: await cluster.routing + ) + } + + /// The earliest and latest offset of every partition, which together give the topic's + /// message count and the anchors every browse mode seeks from. + /// + /// A partition with no answer is an error rather than a zero. It used to read + /// `earliest[partition] ?? 0`, which was unreachable while one request either answered for + /// every partition or threw; once the request is split per leader, one leader failing would + /// have turned its partitions into real-looking empty ones, under-counting the sidebar and + /// sending a tail scan back to the start of the log. + static func bounds(topic: String, partitions: [Int32], cluster: KafkaCluster) async throws -> [KafkaPartitionOffsets] { + guard !partitions.isEmpty else { return [] } + // Sequential, not concurrent. Both calls ask about the same partitions, so they route to + // the same leaders, and overlapping them would put two requests on each of those + // connections at once. + let earliest = try await listOffsets( + topic: topic, + partitions: partitions, + timestamp: earliestTimestamp, + cluster: cluster + ) + let latest = try await listOffsets( + topic: topic, + partitions: partitions, + timestamp: latestTimestamp, + cluster: cluster + ) + let missing = partitions.filter { earliest[$0] == nil || latest[$0] == nil } + guard missing.isEmpty else { + throw KafkaError.partitionsUnanswered(topic: topic, partitions: missing, api: api) + } + return partitions.sorted().map { partition in + KafkaPartitionOffsets( + partition: partition, + earliest: earliest[partition] ?? 0, + latest: latest[partition] ?? 0 + ) + } + } + + /// One ListOffsets to one broker, for the partitions that broker leads. + /// + /// Every partition's error code is carried back rather than thrown, because the reply is a + /// list of per-partition answers and throwing on the first bad one discards the good ones + /// parsed beside it. + private static func send( + topic: String, + partitions: [Int32], + timestamp: Int64, + on connection: KafkaConnection + ) async throws -> [Int32: KafkaPartitionOutcome] { let version = try await connection.negotiatedVersion(for: .listOffsets) let flexible = KafkaApiKey.listOffsets.isFlexible(version: version) @@ -57,7 +128,6 @@ enum KafkaOffsetsRequest { var body = try await connection.send(request) if version >= 2 { _ = try body.int32() } // throttleTimeMs - var offsets: [Int32: Int64] = [:] let topics = try body.array(compact: flexible) { reader -> [(Int32, Int64, Int16)] in _ = flexible ? try reader.compactString() : try reader.legacyString() let partitions = try reader.array(compact: flexible) { partitionReader -> (Int32, Int64, Int16) in @@ -80,37 +150,10 @@ enum KafkaOffsetsRequest { return partitions } + var outcomes: [Int32: KafkaPartitionOutcome] = [:] for entry in topics.flatMap({ $0 }) { - try KafkaErrorCode.check(entry.2, api: "ListOffsets") - offsets[entry.0] = entry.1 - } - return offsets - } - - /// The earliest and latest offset of every partition, which together give the topic's - /// message count and the anchors every browse mode seeks from. - static func bounds(topic: String, partitions: [Int32], cluster: KafkaCluster) async throws -> [KafkaPartitionOffsets] { - // Sequential, not concurrent. Both calls land on the same KafkaConnection, which holds - // exactly one in-flight request: overlapping them makes the second overwrite the - // first's continuation, so the first never resumes and the browse hangs. - let earliest = try await listOffsets( - topic: topic, - partitions: partitions, - timestamp: earliestTimestamp, - cluster: cluster - ) - let latest = try await listOffsets( - topic: topic, - partitions: partitions, - timestamp: latestTimestamp, - cluster: cluster - ) - return partitions.sorted().map { partition in - KafkaPartitionOffsets( - partition: partition, - earliest: earliest[partition] ?? 0, - latest: latest[partition] ?? 0 - ) + outcomes[entry.0] = entry.2 == KafkaErrorCode.none ? .value(entry.1) : .rejected(code: entry.2) } + return outcomes } } diff --git a/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift b/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift index f92bd3440..b7bc25e1e 100644 --- a/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift +++ b/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift @@ -232,8 +232,14 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { nil } + /// The backslash goes first, and it is not optional: the tokenizer treats a backslash + /// inside quotes as an escape, so a value that carried one came back without it, and a + /// value ending in one escaped the closing delimiter and let the rest of the value parse + /// as further clauses. func escapeStringLiteral(_ value: String) -> String { - value.replacingOccurrences(of: "\"", with: "\\\"") + value + .replacingOccurrences(of: "\\", with: "\\\\") + .replacingOccurrences(of: "\"", with: "\\\"") } // MARK: - Browse @@ -258,15 +264,19 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { anchors.clear(table) return "CONSUME \(KafkaQL.quote(table)) FROM NEWEST LIMIT \(page)" } - guard let anchor = anchors.anchor(for: table), !anchor.isEmpty else { + guard let anchor = anchors.anchor(for: table), !anchor.offsets.isEmpty else { // Nothing was recorded, so there is no window to continue. Re-deriving is worse // than nothing here, but it is what the host asked for. return "CONSUME \(KafkaQL.quote(table)) FROM NEWEST LIMIT \(page) SKIP \(offset)" } - let pairs = anchor.sorted { $0.key < $1.key } + let pairs = anchor.offsets.sorted { $0.key < $1.key } .map { "\($0.key):\($0.value)" } .joined(separator: ",") - return "CONSUME \(KafkaQL.quote(table)) FROM ANCHOR (\(pairs)) LIMIT \(page) SKIP \(offset)" + // A tail scan continues by stepping further back from the same end, not by reading + // forward from the same start. Page one of a NEWEST browse recorded its start, so page + // two asked to skip a page inside a window exactly one page long and came back empty. + let clause = anchor.readsBackward ? "TAIL" : "ANCHOR" + return "CONSUME \(KafkaQL.quote(table)) FROM \(clause) (\(pairs)) LIMIT \(page) SKIP \(offset)" } /// Kafka has no server-side WHERE over a log. Returning nil rather than a filtered query @@ -358,9 +368,12 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { private func runConsume(_ query: KafkaConsumeQuery) async throws -> PluginQueryResult { let page = try await KafkaBrowseEngine.consume(query, cluster: cluster) // Only a fresh scan sets the anchor. A continuation page was handed one already, and - // overwriting it with its own start would walk the window forward a page at a time. - if case .resolved = query.start {} else { - anchors.record(page.anchor, for: query.topic) + // overwriting it with its own window would walk that window a page at a time. + switch query.start { + case .resolved, .tail: + break + default: + anchors.record(page.window, for: query.topic) } let kinds = KafkaMessageFlattener.payloadKinds(for: page.records) return Self.makeResult( @@ -397,7 +410,7 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { key: query.key.map { Data($0.utf8) }, value: query.value.map { Data($0.utf8) }, headers: query.headers, - timestamp: Int64(Date().timeIntervalSince1970 * 1000), + timestamp: Int64(Date().timeIntervalSince1970 * 1_000), cluster: cluster ) return Self.makeResult( @@ -435,6 +448,15 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { ) } + /// The cluster's brokers. + /// + /// The last column says which broker this connection would send an admin request to, and it + /// is deliberately not called "controller". Under KRaft a broker fills Metadata's + /// `controllerId` with a randomly chosen live broker and forwards admin requests itself + /// (KIP-590), so the answer moves between runs and names the real controller only by + /// accident: measured on one three-node cluster, the three brokers answered 1, 2 and 1 for + /// the same question. Naming what the field actually decides is true on a KRaft cluster and + /// on a ZooKeeper one, where it is also the controller. private func runShowBrokers() async throws -> PluginQueryResult { let metadata = try await cluster.metadata(refresh: true) let rows = metadata.brokers.sorted { $0.nodeId < $1.nodeId }.map { broker -> [PluginCellValue] in @@ -452,16 +474,22 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { PluginColumnInfo(name: "host", dataType: "TEXT", isNullable: false), PluginColumnInfo(name: "port", dataType: "INTEGER", isNullable: false), PluginColumnInfo(name: "rack", dataType: "TEXT", isNullable: true), - PluginColumnInfo(name: "controller", dataType: "TEXT", isNullable: false) + PluginColumnInfo(name: "takes_admin_requests", dataType: "TEXT", isNullable: false) ], rows: rows, rowsAffected: 0 ) } + /// Every consumer group, gathered from every broker. + /// + /// A broker answers ListGroups with the groups it coordinates and says nothing about the + /// rest, with no error, so asking one broker reported a fraction of a cluster's groups as + /// though that were all of them. When a broker could not be asked the list really is short, + /// and the truncation flag is what says so instead of presenting it as complete. private func runShowGroups() async throws -> PluginQueryResult { - let groups = try await KafkaGroupsRequest.listGroups(cluster: cluster) - let rows = groups.sorted { $0.groupId < $1.groupId }.map { group -> [PluginCellValue] in + let listing = try await KafkaGroupsRequest.listGroups(cluster: cluster) + let rows = listing.groups.map { group -> [PluginCellValue] in [.text(group.groupId), .text(group.state), .text(group.protocolType)] } return Self.makeResult( @@ -471,13 +499,21 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { PluginColumnInfo(name: "protocol_type", dataType: "TEXT", isNullable: false) ], rows: rows, - rowsAffected: 0 + rowsAffected: 0, + isTruncated: !listing.isComplete ) } /// A group's lag, per partition. This is the number a Kafka debugging session is usually /// after: how far behind the consumers are, and on which partition. private func runDescribeGroup(_ group: String) async throws -> PluginQueryResult { + // Asked first, and only to tell a typo apart from a group with nothing committed. Kafka + // answers OffsetFetch for a group it has never heard of with an empty topic list and no + // error, which is byte for byte what a real group that has not committed yet returns, so + // the lag table alone reported both as five columns and no rows. + let detail = try await KafkaGroupsRequest.describeGroups([group], cluster: cluster).first + guard detail?.isKnown ?? true else { throw KafkaError.unknownGroup(group) } + let committed = try await KafkaGroupsRequest.fetchCommittedOffsets(group: group, cluster: cluster) var rows: [[PluginCellValue]] = [] let byTopic = Dictionary(grouping: committed, by: \.topic) @@ -575,7 +611,7 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { let metadata = try await cluster.metadata(refresh: true) let rows: [[PluginCellValue]] = [ [.text("cluster_id"), metadata.clusterId.map { PluginCellValue.text($0) } ?? .null], - [.text("controller"), .text(String(metadata.controllerId))], + [.text("admin_requests_to"), .text(String(metadata.controllerId))], [.text("brokers"), .text(String(metadata.brokers.count))], [.text("topics"), .text(String(metadata.topics.filter { !$0.isInternal }.count))], [.text("bootstrap"), .text(await cluster.bootstrapEndpointDescription())] @@ -591,21 +627,32 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { } } -/// Where each topic's current browse window starts, so page two continues page one. +/// Where each topic's current browse window sits, so page two continues page one. /// /// A plain lock rather than an actor: `buildBrowseQuery` is a synchronous protocol requirement /// and cannot await. private final class KafkaAnchorStore: @unchecked Sendable { - private var anchors: [String: [Int32: Int64]] = [:] + /// One partition offset per partition, plus which end of the window it is. A forward scan + /// pins its start and reads on from there; a tail scan pins its end and later pages step + /// back from it. + struct Anchor { + let offsets: [Int32: Int64] + let readsBackward: Bool + } + + private var anchors: [String: Anchor] = [:] private let lock = NSLock() - func record(_ anchor: [Int32: Int64], for table: String) { + func record(_ window: KafkaScanWindow, for table: String) { lock.lock() defer { lock.unlock() } - anchors[table] = anchor + anchors[table] = Anchor( + offsets: window.readsBackward ? window.tail : window.start, + readsBackward: window.readsBackward + ) } - func anchor(for table: String) -> [Int32: Int64]? { + func anchor(for table: String) -> Anchor? { lock.lock() defer { lock.unlock() } return anchors[table] diff --git a/Plugins/KafkaDriverPlugin/KafkaProduceRequest.swift b/Plugins/KafkaDriverPlugin/KafkaProduceRequest.swift index a3dec15fa..1560b4520 100644 --- a/Plugins/KafkaDriverPlugin/KafkaProduceRequest.swift +++ b/Plugins/KafkaDriverPlugin/KafkaProduceRequest.swift @@ -26,7 +26,12 @@ enum KafkaProduceRequest { timestamp: timestamp ) - return try await cluster.withLeader(of: partition, topic: topic) { connection in + return try await cluster.withLeader( + of: partition, + topic: topic, + api: "Produce", + repeatability: .onlyWhenBrokerRefusedIt + ) { connection in let version = try await connection.negotiatedVersion(for: .produce) let flexible = KafkaApiKey.produce.isFlexible(version: version) diff --git a/Plugins/KafkaDriverPlugin/KafkaQL.swift b/Plugins/KafkaDriverPlugin/KafkaQL.swift index 04731d2a2..f91151648 100644 --- a/Plugins/KafkaDriverPlugin/KafkaQL.swift +++ b/Plugins/KafkaDriverPlugin/KafkaQL.swift @@ -11,10 +11,17 @@ enum KafkaStartMode: Sendable, Equatable { case offset(Int64) /// The first message at or after a wall-clock time, resolved by ListOffsets. case timestamp(Int64) - /// Per-partition offsets resolved by an earlier call. This is what makes paging stable: - /// page two reads the anchor page one resolved instead of re-deriving "newest" against a - /// tail that has moved on. + /// Per-partition offsets resolved by an earlier call, read forward. This is what makes + /// paging stable: page two reads the anchor page one resolved instead of re-deriving + /// "newest" against a tail that has moved on. case resolved([Int32: Int64]) + /// Per-partition exclusive end offsets resolved by an earlier call, read backward. + /// + /// A tail scan needs its fixed end rather than its fixed start, because each later page + /// steps further back from the same end. Recording the start instead pinned the window to + /// the newest page's worth of messages, and page two then skipped past all of it and + /// showed nothing. + case tail([Int32: Int64]) } struct KafkaConsumeQuery: Sendable { @@ -150,7 +157,7 @@ enum KafkaQL { private static func parseStart(_ tokens: inout Tokenizer) throws -> KafkaStartMode { guard let mode = tokens.next()?.uppercased() else { - throw KafkaError.syntax(String(localized: "FROM needs NEWEST, OLDEST, OFFSET, TIME or ANCHOR.")) + throw KafkaError.syntax(String(localized: "FROM needs NEWEST, OLDEST, OFFSET, TIME, ANCHOR or TAIL.")) } switch mode { case "NEWEST", "LATEST", "END": @@ -163,16 +170,21 @@ enum KafkaQL { return .timestamp(try timestampValue(&tokens)) case "ANCHOR": return .resolved(try anchorMap(&tokens)) + case "TAIL": + return .tail(try anchorMap(&tokens)) default: throw KafkaError.syntax(String( - format: String(localized: "%@ is not a start position. Use NEWEST, OLDEST, OFFSET, TIME or ANCHOR."), + format: String(localized: """ + %@ is not a start position. Use NEWEST, OLDEST, OFFSET, TIME, ANCHOR or TAIL. + """), mode )) } } - /// `ANCHOR(0:120,1:80)` pins one offset per partition. It is machine-written by the browse - /// path rather than typed, and it is what makes page two continue page one exactly. + /// `ANCHOR(0:120,1:80)` and `TAIL(0:400,1:250)` pin one offset per partition. Both are + /// machine-written by the browse path rather than typed, and they are what make page two + /// continue page one exactly. private static func anchorMap(_ tokens: inout Tokenizer) throws -> [Int32: Int64] { guard let raw = tokens.next() else { throw KafkaError.syntax(String(localized: "ANCHOR needs a partition:offset list.")) @@ -199,7 +211,19 @@ enum KafkaQL { throw KafkaError.syntax(String(localized: "PARTITION needs at least one partition number.")) } let body = raw.hasPrefix("(") ? String(raw.dropFirst().dropLast()) : raw - let values = body.split(separator: ",").compactMap { Int32($0.trimmingCharacters(in: .whitespaces)) } + var values: [Int32] = [] + for entry in body.split(separator: ",") { + let text = entry.trimmingCharacters(in: .whitespaces) + // Dropping what does not parse turned a typo into a silently narrower read: a page + // half the size it asked for, reported as a success. + guard let value = Int32(text) else { + throw KafkaError.syntax(String( + format: String(localized: "%@ is not a partition number."), + text + )) + } + values.append(value) + } guard !values.isEmpty else { throw KafkaError.syntax(String(localized: "PARTITION needs at least one partition number.")) } @@ -247,10 +271,18 @@ enum KafkaQL { throw KafkaError.syntax(String(localized: "SHOW needs TOPICS, BROKERS, GROUPS or CLUSTER.")) } switch what { - case "TOPICS": return .showTopics - case "BROKERS": return .showBrokers - case "GROUPS", "CONSUMERS": return .showGroups - case "CLUSTER": return .showCluster + case "TOPICS": + try requireEnd(&tokens, statement: "SHOW TOPICS") + return .showTopics + case "BROKERS": + try requireEnd(&tokens, statement: "SHOW BROKERS") + return .showBrokers + case "GROUPS", "CONSUMERS": + try requireEnd(&tokens, statement: "SHOW GROUPS") + return .showGroups + case "CLUSTER": + try requireEnd(&tokens, statement: "SHOW CLUSTER") + return .showCluster default: throw KafkaError.syntax(String( format: String(localized: "SHOW %@ is not supported. Try TOPICS, BROKERS, GROUPS or CLUSTER."), @@ -267,8 +299,12 @@ enum KafkaQL { throw KafkaError.syntax(String(localized: "DESCRIBE needs a name.")) } switch what { - case "GROUP": return .describeGroup(unquote(name)) - case "TOPIC": return .describeTopic(unquote(name)) + case "GROUP": + try requireEnd(&tokens, statement: "DESCRIBE GROUP") + return .describeGroup(unquote(name)) + case "TOPIC": + try requireEnd(&tokens, statement: "DESCRIBE TOPIC") + return .describeTopic(unquote(name)) default: throw KafkaError.syntax(String( format: String(localized: "DESCRIBE %@ is not supported. Try GROUP or TOPIC."), @@ -279,6 +315,20 @@ enum KafkaQL { // MARK: - Helpers + /// Refuses anything after the statement has been read. + /// + /// `DROP TOPIC` has always done this; SHOW and DESCRIBE did not, so + /// `DESCRIBE TOPIC "orders" "payments"` described the first and discarded the second + /// without a word, and a modifier nobody implemented read as though it had been applied. + private static func requireEnd(_ tokens: inout Tokenizer, statement: String) throws { + guard let extra = tokens.next() else { return } + throw KafkaError.syntax(String( + format: String(localized: "%@ takes nothing after it, and %@ was given."), + statement, + extra + )) + } + private static func requireValue(_ tokens: inout Tokenizer, keyword: String) throws -> String { guard let value = tokens.next() else { throw KafkaError.syntax(String(format: String(localized: "%@ needs a value."), keyword)) @@ -305,9 +355,9 @@ enum KafkaQL { if let milliseconds = Int64(raw) { return milliseconds } let formatter = ISO8601DateFormatter() formatter.formatOptions = [.withInternetDateTime, .withFractionalSeconds] - if let date = formatter.date(from: raw) { return Int64(date.timeIntervalSince1970 * 1000) } + if let date = formatter.date(from: raw) { return Int64(date.timeIntervalSince1970 * 1_000) } formatter.formatOptions = [.withInternetDateTime] - if let date = formatter.date(from: raw) { return Int64(date.timeIntervalSince1970 * 1000) } + if let date = formatter.date(from: raw) { return Int64(date.timeIntervalSince1970 * 1_000) } throw KafkaError.syntax(String( format: String(localized: "%@ is not a time. Use milliseconds since the epoch or an ISO 8601 instant."), raw @@ -340,8 +390,17 @@ enum KafkaQL { return unescaped } + /// Wraps a value so the tokenizer gives back exactly what went in. + /// + /// The backslash has to be escaped first and it is not optional: the tokenizer treats a + /// backslash inside quotes as an escape and `unquote` strips it, so a value carrying one + /// came back with it missing, and a value ending in one swallowed the closing delimiter and + /// let the rest of the value parse as further clauses. static func quote(_ value: String) -> String { - "\"\(value.replacingOccurrences(of: "\"", with: "\\\""))\"" + let escaped = value + .replacingOccurrences(of: "\\", with: "\\\\") + .replacingOccurrences(of: "\"", with: "\\\"") + return "\"\(escaped)\"" } /// Splits on whitespace while keeping quoted strings and parenthesised lists whole. diff --git a/Plugins/KafkaDriverPlugin/KafkaRecordOrdering.swift b/Plugins/KafkaDriverPlugin/KafkaRecordOrdering.swift index 847ecbc19..7e1950ed1 100644 --- a/Plugins/KafkaDriverPlugin/KafkaRecordOrdering.swift +++ b/Plugins/KafkaDriverPlugin/KafkaRecordOrdering.swift @@ -18,4 +18,21 @@ enum KafkaRecordOrdering { return lhs.offset < rhs.offset } } + + /// The page to show, from the end of the merged run the scan was anchored to. + /// + /// A tail scan 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 records are at the END. + /// Taking the front of it returns the oldest records of the tail window, which on one + /// partition is the same rows and on three is the wrong end of the log. This is a separate + /// function because that is a presentation decision with no I/O in it, and the defect + /// survived because the code that made it could only be reached through a broker. + static func page(_ ordered: [KafkaRecord], skip: Int, limit: Int, readsBackward: Bool) -> [KafkaRecord] { + guard limit > 0 else { return [] } + let dropped = max(0, skip) + guard readsBackward else { + return Array(ordered.dropFirst(dropped).prefix(limit)) + } + return Array(ordered.dropLast(dropped).suffix(limit)) + } } diff --git a/TableProTests/Plugins/KafkaIntegrationTests.swift b/TableProTests/Plugins/KafkaIntegrationTests.swift index 114eee422..1c1db8bf4 100644 --- a/TableProTests/Plugins/KafkaIntegrationTests.swift +++ b/TableProTests/Plugins/KafkaIntegrationTests.swift @@ -178,14 +178,15 @@ struct KafkaIntegrationTests { #expect(Set(rows.compactMap { $0[0].asText }) == ["0"]) } - /// Paging is the reason the ANCHOR clause exists: page two has to continue page one rather + /// Paging is the reason the TAIL clause exists: page two has to continue page one rather /// than re-deriving its start against a tail that has moved. /// - /// The invariant is that no message is shown twice, not that three pages of ten cover - /// exactly thirty. A `NEWEST` scan anchors each partition at its own tail, and messages - /// are not spread evenly across partitions, so the window a `NEWEST` anchor opens holds - /// however many messages happen to be in it. Coverage is asserted from `OLDEST`, where the - /// anchor is the start of the log and paging forward really does reach everything. + /// The invariant is that no message is shown twice and that no page comes back short, not + /// that three pages of ten cover exactly thirty. A `NEWEST` scan anchors each partition at + /// its own tail, and messages are not spread evenly across partitions, so the window it + /// opens holds however many messages happen to be in it. Coverage is asserted from + /// `OLDEST`, where the anchor is the start of the log and paging forward really does reach + /// everything. @Test("Paging from the tail never shows the same message twice") func pagingNeverRepeats() async throws { let harness = try await KafkaTestBroker.harness(topic: "tp-it-paging", partitions: 3) @@ -203,11 +204,14 @@ struct KafkaIntegrationTests { offset: page * 10 )) if page > 0 { - // Later pages must name the window page one resolved. - #expect(query.contains("FROM ANCHOR")) + // Later pages must name the window page one resolved, and a tail browse pins + // the END of that window so each page steps further back from it. + #expect(query.contains("FROM TAIL")) } let result = try await harness.driver.execute(query: query) - seen.append(contentsOf: result.rows.map { "\($0[0].asText ?? "?"):\($0[1].asText ?? "?")" }) + let rows = result.rows.map { "\($0[0].asText ?? "?"):\($0[1].asText ?? "?")" } + #expect(rows.count == 10, "page \(page + 1) of a tail browse came back short") + seen.append(contentsOf: rows) } #expect(seen.isEmpty == false) @@ -335,17 +339,22 @@ struct KafkaIntegrationTests { #expect(topics.columns == ["topic", "partitions", "replication_factor", "internal"]) #expect(topics.rows.contains { $0.first?.asText == harness.topic }) + // Asserted against however many brokers are actually running rather than against one. + // Pinning the count to 1 is what made these three assertions fail the moment the suite + // met a cluster big enough to reproduce the routing bug they sit beside. let brokers = try await harness.driver.execute(query: "SHOW BROKERS") - #expect(brokers.rows.count == 1) - #expect(brokers.rows.first?[4].asText == "yes") // the controller + #expect(brokers.columns.last == "takes_admin_requests") + #expect(brokers.rows.isEmpty == false) + #expect(brokers.rows.filter { $0[4].asText == "yes" }.count == 1) let cluster = try await harness.driver.execute(query: "SHOW CLUSTER") let properties = Dictionary(uniqueKeysWithValues: cluster.rows.compactMap { row -> (String, String)? in guard let key = row.first?.asText else { return nil } return (key, row[1].asText ?? "") }) - #expect(properties["brokers"] == "1") + #expect(properties["brokers"] == String(brokers.rows.count)) #expect(properties["cluster_id"]?.isEmpty == false) + #expect(properties["admin_requests_to"]?.isEmpty == false) let described = try await harness.driver.execute(query: "DESCRIBE TOPIC \(harness.quoted)") #expect(described.rows.count == 2) @@ -408,22 +417,150 @@ struct KafkaIntegrationTests { } /// The lag report is what a Kafka debugging session is usually after. + /// + /// Three partitions rather than one, so the end offsets come from ListOffsets sent to three + /// different leaders on a multi-broker cluster. With one partition this passed while every + /// request went to whichever broker the connection happened to hold. @Test("Consumer group lag is reported per partition") func consumerGroupLag() async throws { - let harness = try await KafkaTestBroker.harness(topic: "tp-it-lag", partitions: 1) + let harness = try await KafkaTestBroker.harness(topic: "tp-it-lag", partitions: 3) defer { harness.tearDown() } + // The group consumes everything first, so every partition it was assigned has a + // committed offset; the six produced afterwards are the lag. Committing only part of a + // topic leaves the partitions it never reached out of the report entirely, which is + // correct and not something to assert arithmetic against. try await harness.produce(count: 10) - try harness.commitGroup(named: "tp-it-group", messages: 4) + try harness.commitGroup(named: "tp-it-group", messages: 10) + try await harness.produce(count: 6, startingAt: 10) let groups = try await harness.driver.execute(query: "SHOW GROUPS") #expect(groups.rows.contains { $0.first?.asText == "tp-it-group" }) let lag = try await harness.driver.execute(query: "DESCRIBE GROUP \"tp-it-group\"") #expect(lag.columns == ["topic", "partition", "committed_offset", "end_offset", "lag"]) - let row = try #require(lag.rows.first { $0.first?.asText == harness.topic }) - #expect(row[3].asText == "10") - // Four consumed of ten leaves six behind. - #expect(row[4].asText == "6") + let rows = lag.rows.filter { $0.first?.asText == harness.topic } + #expect(rows.count == 3, "every partition the group committed must be reported") + let written = rows.compactMap { Int($0[3].asText ?? "") }.reduce(0, +) + let consumed = rows.compactMap { Int($0[2].asText ?? "") }.reduce(0, +) + let behind = rows.compactMap { Int($0[4].asText ?? "") }.reduce(0, +) + #expect(written == 16) + #expect(consumed == 10) + // Ten consumed of sixteen leaves six behind, wherever the sixteen landed. + #expect(behind == 6) + } + + /// A group the cluster has never heard of used to come back as five columns and no rows, + /// which is byte for byte what a real group with nothing committed returns. + @Test("Describing a group that does not exist says so") + func describeUnknownGroupReports() async throws { + let harness = try await KafkaTestBroker.harness(topic: "tp-it-nogroup", partitions: 1) + defer { harness.tearDown() } + + await #expect(throws: (any Error).self) { + try await harness.driver.execute(query: "DESCRIBE GROUP \"tp-it-no-such-group-4c71\"") + } + } + + // MARK: - Routing across brokers + + /// The reported bug (#2993), and the reason the rest of this suite could not catch it. + /// + /// A topic's partitions are spread across the cluster's brokers, so every one of these + /// statements has to reach a broker the connection was not opened to. On a single-broker + /// cluster they all pass without any routing at all, which is why they are asserted here + /// against whatever the harness is running and are worth running against three. + @Test("Every statement that needs an offset works wherever the leaders are") + func offsetsReachEveryLeader() async throws { + let harness = try await KafkaTestBroker.harness(topic: "tp-it-routing", partitions: 6) + defer { harness.tearDown() } + try await harness.produce(count: 30) + + let described = try await harness.driver.execute(query: "DESCRIBE TOPIC \(harness.quoted)") + #expect(described.rows.count == 6) + let counted = described.rows.compactMap { Int($0.last?.asText ?? "") }.reduce(0, +) + #expect(counted == 30) + + let metadata = try await harness.driver.fetchTableMetadata(table: harness.topic, schema: nil) + #expect(metadata.rowCount == 30) + + let consumed = try await harness.rows("CONSUME \(harness.quoted) FROM NEWEST LIMIT 100") + #expect(consumed.count == 30) + } + + /// `FROM NEWEST` means the newest messages. On more than one partition the scan steps every + /// partition back by the page size and reads forward, so the merged run holds several pages + /// and the newest rows sit at its end; taking the front of it returned the oldest rows of + /// the tail window and called them the newest. + @Test("FROM NEWEST returns the newest messages, not the oldest of its window") + func newestReturnsTheNewest() async throws { + let harness = try await KafkaTestBroker.harness(topic: "tp-it-newest", partitions: 3) + defer { harness.tearDown() } + try await harness.produce(count: 60) + + let everything = try await harness.rows("CONSUME \(harness.quoted) FROM OLDEST LIMIT 100") + #expect(everything.count == 60) + let newestKeys = Set(everything.suffix(10).compactMap { $0[3].asText }) + + let newest = try await harness.rows("CONSUME \(harness.quoted) FROM NEWEST LIMIT 10") + #expect(newest.count == 10) + #expect(Set(newest.compactMap { $0[3].asText }) == newestKeys) + } + + /// Page two of a tail browse used to be empty: page one recorded the start of its window + /// rather than the end, so page two skipped a page inside a window exactly one page long. + @Test("The second page of a tail browse holds the messages before the first") + func tailPagingWalksBackwards() async throws { + let harness = try await KafkaTestBroker.harness(topic: "tp-it-tailpage", partitions: 3) + defer { harness.tearDown() } + try await harness.produce(count: 60) + + var pages: [[String]] = [] + for page in 0 ..< 2 { + let query = try #require(harness.driver.buildBrowseQuery( + table: harness.topic, + schema: nil, + sortColumns: [], + columns: [], + limit: 10, + offset: page * 10 + )) + if page > 0 { #expect(query.contains("FROM TAIL")) } + let rows = try await harness.driver.execute(query: query).rows + pages.append(rows.compactMap { $0[3].asText }) + } + + #expect(pages[0].count == 10) + #expect(pages[1].count == 10, "page two of a tail browse must hold the page before it") + #expect(Set(pages[0]).isDisjoint(with: Set(pages[1]))) + } + + /// A broker answers ListGroups with the groups it coordinates and nothing else, so a client + /// that asks one broker reports a fraction of them with no error. + @Test("SHOW GROUPS lists groups from every broker, not just the one connected to") + func showGroupsCoversEveryBroker() async throws { + let harness = try await KafkaTestBroker.harness(topic: "tp-it-groups", partitions: 3) + defer { harness.tearDown() } + try await harness.produce(count: 6) + + // Several names, because which broker coordinates a group is a hash of its id: one + // group lands on one broker and says nothing about whether the sweep happened. + let names = (0 ..< 6).map { "tp-it-sweep-\($0)" } + for name in names { + try harness.commitGroup(named: name, messages: 1) + } + + let listed = try await harness.driver.execute(query: "SHOW GROUPS") + let found = Set(listed.rows.compactMap { $0.first?.asText }) + for name in names { + #expect(found.contains(name), "SHOW GROUPS did not list \(name)") + } + + // And each one is describable, which needs its own coordinator rather than the + // bootstrap broker. + for name in names { + let lag = try await harness.driver.execute(query: "DESCRIBE GROUP \(KafkaQL.quote(name))") + #expect(lag.rows.isEmpty == false, "DESCRIBE GROUP \(name) returned nothing") + } } // MARK: - Compression, end to end through the driver diff --git a/TableProTests/Plugins/KafkaQLTests.swift b/TableProTests/Plugins/KafkaQLTests.swift index 77e403f8a..adb86fd14 100644 --- a/TableProTests/Plugins/KafkaQLTests.swift +++ b/TableProTests/Plugins/KafkaQLTests.swift @@ -140,13 +140,35 @@ struct KafkaQLTests { /// backslashes onto the topic. @Test("A quoted value round-trips through quote and unquote") func quotingRoundTrips() throws { - for value in ["plain", "with space", "say \"hi\"", "a,b", "trailing\\"] { - #expect(KafkaQL.unquote(KafkaQL.quote(value)) == value) + let values = [ + "plain", "with space", "say \"hi\"", "a,b", "trailing\\", + "C:\\temp\\app.log", "\\\\server\\\\share", "a\\\"b", "\\", "\\\\" + ] + for value in values { + #expect(KafkaQL.unquote(KafkaQL.quote(value)) == value, "round trip of \(value)") } let produced = try produce("PRODUCE INTO t VALUE \"say \\\"hi\\\"\"") #expect(produced.value == "say \"hi\"") } + /// A value carrying a Windows path used to arrive on the topic with its separators gone, + /// because `quote` escaped the delimiter but not the escape character the tokenizer honours. + @Test("A value full of backslashes reaches the topic intact") + func backslashesSurviveTheParser() throws { + let statement = "PRODUCE INTO \"logs\" VALUE \(KafkaQL.quote("C:\\temp\\app.log"))" + #expect(try produce(statement).value == "C:\\temp\\app.log") + } + + /// The same gap let a value close its own quote and have the rest of itself parsed as + /// further clauses, so a string chose the partition it was written to. + @Test("A value cannot escape its quotes and inject a clause") + func aValueCannotInjectAClause() throws { + let hostile = "a\\\" PARTITION 3 VALUE \"b" + let query = try produce("PRODUCE INTO \"orders\" VALUE \(KafkaQL.quote(hostile))") + #expect(query.value == hostile) + #expect(query.partition == nil) + } + @Test("Bad input is reported with a message rather than silently ignored") func syntaxErrors() { #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("") } @@ -163,6 +185,41 @@ struct KafkaQLTests { #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("DROP TOPIC a b") } } + /// Every one of these used to parse. A token the grammar has no use for is a typo or a + /// modifier the driver does not implement, and reading the statement without it reports + /// success for something other than what was asked. + @Test("A token the statement has no use for is refused, not discarded") + func trailingTokensAreRefused() { + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("SHOW TOPICS INTERNAL") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("SHOW BROKERS ALL") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("SHOW GROUPS STABLE") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("SHOW CLUSTER VERBOSE") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("DESCRIBE TOPIC \"orders\" \"payments\"") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("DESCRIBE GROUP \"a\" \"b\"") } + } + + /// A partition number that does not parse used to be dropped before anything checked it, so + /// one typo quietly halved the read and the grid reported a full success. + @Test("A partition list refuses what it cannot read as a number") + func partitionListRefusesNonNumbers() throws { + #expect(try consume("CONSUME t PARTITION (0,2)").partitions == [0, 2]) + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("CONSUME t PARTITION (0,two,2)") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("CONSUME t PARTITION (first)") } + } + + /// A tail scan pins the end it steps back from, so later pages of a NEWEST browse read the + /// window page one measured rather than re-deriving it against a moving tail. + @Test("FROM TAIL parses the exclusive end of each partition") + func parsesTailAnchors() throws { + let query = try consume("CONSUME \"orders\" FROM TAIL (0:400,1:250) LIMIT 10 SKIP 10") + guard case .tail(let ends) = query.start else { + Issue.record("expected a tail start mode") + return + } + #expect(ends == [0: 400, 1: 250]) + #expect(query.skip == 10) + } + /// A topic delete is KafkaQL's own verb. `DROP TABLE` stays a syntax error above, because a /// topic is not a table and that text is what the app used to invent for engines with no SQL. @Test("DROP TOPIC names the topic to delete") diff --git a/TableProTests/Plugins/KafkaRoutingTests.swift b/TableProTests/Plugins/KafkaRoutingTests.swift new file mode 100644 index 000000000..fc17ca367 --- /dev/null +++ b/TableProTests/Plugins/KafkaRoutingTests.swift @@ -0,0 +1,390 @@ +import Foundation +import TableProPluginKit +import Testing + +@testable import TablePro + +/// The parts of Kafka request routing that can be decided without a broker. +/// +/// Everything here failed silently on a multi-broker cluster and could not fail at all on a +/// single-broker one, which is why issue #2993 survived a full integration suite: with one +/// broker every partition's leader and every group's coordinator is the broker the client is +/// already holding, so a driver that routes nothing is indistinguishable from a correct one. +@Suite("Kafka routing") +struct KafkaRoutingTests { + // MARK: - Error code classification + + /// The constants are a hand transcription of Kafka's own table and nothing at runtime checks + /// them, so the two that were transposed are pinned by number. + @Test("72 is LISTENER_NOT_FOUND and 73 is TOPIC_DELETION_DISABLED") + func listenerAndDeletionCodesAreNotTransposed() { + #expect(KafkaErrorCode.listenerNotFound == 72) + #expect(KafkaErrorCode.topicDeletionDisabled == 73) + } + + @Test("A code that says the cluster moved asks for a fresh leader") + func staleMetadataCodesResolveTheLeaderAgain() { + let stale: [Int16] = [ + KafkaErrorCode.unknownTopicOrPartition, + KafkaErrorCode.leaderNotAvailable, + KafkaErrorCode.notLeaderOrFollower, + KafkaErrorCode.replicaNotAvailable, + KafkaErrorCode.networkException, + KafkaErrorCode.kafkaStorageError, + KafkaErrorCode.listenerNotFound, + KafkaErrorCode.fencedLeaderEpoch, + KafkaErrorCode.unknownLeaderEpoch, + KafkaErrorCode.unknownTopicId + ] + for code in stale { + #expect(KafkaErrorCode.retryAction(for: code) == .resolveLeaderAgain, "code \(code)") + } + } + + @Test("A coordinator code asks for a fresh coordinator, not a fresh leader") + func coordinatorCodesFindTheCoordinatorAgain() { + let coordinator: [Int16] = [ + KafkaErrorCode.coordinatorLoadInProgress, + KafkaErrorCode.coordinatorNotAvailable, + KafkaErrorCode.notCoordinator + ] + for code in coordinator { + #expect(KafkaErrorCode.retryAction(for: code) == .findCoordinatorAgain, "code \(code)") + } + } + + /// A broker that is genuinely down is not a stale-metadata problem, and retrying it hides + /// the outage behind a slower failure. Kafka does not classify it retriable either. + @Test("A broker that is down and a real answer are both reported, not retried") + func nonRetriableCodesAreReported() { + #expect(KafkaErrorCode.retryAction(for: KafkaErrorCode.brokerNotAvailable) == .report) + #expect(KafkaErrorCode.retryAction(for: KafkaErrorCode.offsetOutOfRange) == .report) + #expect(KafkaErrorCode.retryAction(for: KafkaErrorCode.topicAuthorizationFailed) == .report) + #expect(KafkaErrorCode.retryAction(for: KafkaErrorCode.none) == .report) + } + + @Test("A transient code is retried where it is, without re-reading metadata") + func transientCodesRetryTheSameBroker() { + #expect(KafkaErrorCode.retryAction(for: KafkaErrorCode.requestTimedOut) == .retrySameBroker) + #expect(KafkaErrorCode.retryAction(for: KafkaErrorCode.offsetNotAvailable) == .retrySameBroker) + } + + // MARK: - Repeating a request + + /// A read can always be sent again. The retry exists for it. + @Test("A read is retried on every code that says the cluster moved") + func aReadRetriesEveryMovedCode() { + for code in [KafkaErrorCode.notLeaderOrFollower, KafkaErrorCode.leaderNotAvailable, + KafkaErrorCode.requestTimedOut, KafkaErrorCode.networkException, + KafkaErrorCode.kafkaStorageError, KafkaErrorCode.offsetNotAvailable] { + #expect(KafkaRequestRepeatability.safeToRepeat.allowsRetry(after: code), "code \(code)") + } + } + + /// A Produce asks for acks = -1 and sends no producer id, so REQUEST_TIMED_OUT can mean the + /// leader appended the record and its replicas were late acknowledging it. Sending it again + /// writes the message twice, with nothing on the cluster able to tell that it was one + /// message. + @Test("A write is not repeated on a code that leaves the outcome unknown") + func aWriteIsNotRepeatedWhenTheOutcomeIsUnknown() { + let ambiguous = [ + KafkaErrorCode.requestTimedOut, + KafkaErrorCode.networkException, + KafkaErrorCode.kafkaStorageError, + KafkaErrorCode.offsetNotAvailable + ] + for code in ambiguous { + #expect(!KafkaRequestRepeatability.onlyWhenBrokerRefusedIt.allowsRetry(after: code), "code \(code)") + #expect(!KafkaErrorCode.provesRequestWasNotApplied(code), "code \(code)") + } + } + + /// The codes that mean the broker turned the request away are safe for a write, and they + /// are the ones that carry the reported bug. + @Test("A write is repeated when the broker refused it outright") + func aWriteIsRepeatedWhenRefused() { + let refused = [ + KafkaErrorCode.notLeaderOrFollower, + KafkaErrorCode.leaderNotAvailable, + KafkaErrorCode.unknownTopicOrPartition, + KafkaErrorCode.replicaNotAvailable, + KafkaErrorCode.listenerNotFound, + KafkaErrorCode.fencedLeaderEpoch, + KafkaErrorCode.unknownLeaderEpoch, + KafkaErrorCode.unknownTopicId + ] + for code in refused { + #expect(KafkaRequestRepeatability.onlyWhenBrokerRefusedIt.allowsRetry(after: code), "code \(code)") + } + } + + @Test("Neither kind repeats a real answer") + func neitherKindRepeatsARealAnswer() { + for code in [KafkaErrorCode.offsetOutOfRange, KafkaErrorCode.topicAuthorizationFailed, + KafkaErrorCode.brokerNotAvailable] { + #expect(!KafkaRequestRepeatability.safeToRepeat.allowsRetry(after: code), "code \(code)") + #expect(!KafkaRequestRepeatability.onlyWhenBrokerRefusedIt.allowsRetry(after: code), "code \(code)") + } + } + + // MARK: - Collecting per-partition answers + + @Test("Every partition answering gives every partition's value") + func everyPartitionAnswers() throws { + let outcomes: [Int32: KafkaPartitionOutcome] = [ + 0: .value(10), 1: .value(20), 2: .value(30) + ] + let values = try KafkaPartitionOutcome.requireAll( + outcomes, + topic: "orders", + api: "ListOffsets", + routing: .advertised + ) + #expect(values == [0: 10, 1: 20, 2: 30]) + } + + /// The shape of #2993: a broker answers for the partitions it leads and rejects the rest in + /// the same reply. The rejection has to name the partitions rather than being reported as + /// though the whole request failed. + @Test("A rejected partition names itself in the error") + func aRejectedPartitionNamesItself() { + let outcomes: [Int32: KafkaPartitionOutcome] = [ + 0: .value(10), + 1: .rejected(code: KafkaErrorCode.notLeaderOrFollower), + 2: .rejected(code: KafkaErrorCode.notLeaderOrFollower) + ] + do { + _ = try KafkaPartitionOutcome.requireAll( + outcomes, + topic: "orders", + api: "ListOffsets", + routing: .advertised + ) + Issue.record("expected the rejected partitions to be reported") + } catch let error as KafkaError { + guard case .partitionsRejected(let topic, let partitions, let api, let code) = error else { + Issue.record("expected partitionsRejected, got \(error)") + return + } + #expect(topic == "orders") + #expect(partitions.sorted() == [1, 2]) + #expect(api == "ListOffsets") + #expect(code == KafkaErrorCode.notLeaderOrFollower) + } catch { + Issue.record("unexpected error \(error)") + } + } + + private func rejectionCode(_ outcomes: [Int32: KafkaPartitionOutcome]) -> Int16? { + do { + _ = try KafkaPartitionOutcome.requireAll( + outcomes, + topic: "orders", + api: "ListOffsets", + routing: .advertised + ) + Issue.record("expected a rejection") + return nil + } catch let error as KafkaError { + guard case .partitionsRejected(_, _, _, let code) = error else { + Issue.record("expected partitionsRejected, got \(error)") + return nil + } + return code + } catch { + Issue.record("unexpected error \(error)") + return nil + } + } + + /// Only one of several partitions' codes can be reported. Picking the numerically smallest + /// let OFFSET_OUT_OF_RANGE (1) mask TOPIC_AUTHORIZATION_FAILED (29), which is the one the + /// user has to act on. + @Test("A permission failure is named ahead of anything else") + func aPermissionFailureIsNamedFirst() { + #expect(rejectionCode([ + 0: .rejected(code: KafkaErrorCode.offsetOutOfRange), + 1: .rejected(code: KafkaErrorCode.topicAuthorizationFailed) + ]) == KafkaErrorCode.topicAuthorizationFailed) + + #expect(rejectionCode([ + 0: .rejected(code: KafkaErrorCode.notLeaderOrFollower), + 5: .rejected(code: KafkaErrorCode.groupAuthorizationFailed) + ]) == KafkaErrorCode.groupAuthorizationFailed) + } + + @Test("A real answer is named ahead of a code that only says the cluster is moving") + func aRealAnswerOutranksATransientOne() { + #expect(rejectionCode([ + 0: .rejected(code: KafkaErrorCode.leaderNotAvailable), + 1: .rejected(code: KafkaErrorCode.offsetOutOfRange) + ]) == KafkaErrorCode.offsetOutOfRange) + } + + /// Two codes of the same rank tie on the lowest partition, so the message does not change + /// between runs of the same broken query. + @Test("Equally useful codes are broken by partition, not by number") + func equalCodesTieOnTheLowestPartition() { + #expect(rejectionCode([ + 2: .rejected(code: KafkaErrorCode.offsetOutOfRange), + 1: .rejected(code: KafkaErrorCode.messageTooLarge) + ]) == KafkaErrorCode.messageTooLarge) + } + + /// Under bootstrap-only routing the same code means something the user can act on, so it is + /// reported as the setting rather than as Kafka's wording. + @Test("Bootstrap-only routing names the setting rather than the error code") + func bootstrapOnlyNamesTheSetting() { + let outcomes: [Int32: KafkaPartitionOutcome] = [ + 0: .value(10), + 3: .rejected(code: KafkaErrorCode.notLeaderOrFollower) + ] + do { + _ = try KafkaPartitionOutcome.requireAll( + outcomes, + topic: "orders", + api: "ListOffsets", + routing: .bootstrapOnly + ) + Issue.record("expected the partitions led elsewhere to be reported") + } catch let error as KafkaError { + guard case .partitionsLedElsewhere(let topic, let partitions) = error else { + Issue.record("expected partitionsLedElsewhere, got \(error)") + return + } + #expect(topic == "orders") + #expect(partitions == [3]) + } catch { + Issue.record("unexpected error \(error)") + } + } + + /// A partition with no answer used to become earliest 0 / latest 0, which reads as a real + /// empty partition: the sidebar under-counted and a tail scan restarted from the log's + /// beginning, with nothing raised. + @Test("A partition that could not be reached is an error, not a zero") + func anUnreachablePartitionIsNotZero() { + let outcomes: [Int32: KafkaPartitionOutcome] = [ + 0: .value(10), + 1: .failed(.brokerUnreachable(nodeId: 2, address: "kafka-2.internal:9092", reason: "timed out")) + ] + #expect(throws: KafkaError.self) { + _ = try KafkaPartitionOutcome.requireAll( + outcomes, + topic: "orders", + api: "ListOffsets", + routing: .advertised + ) + } + } + + /// Fetch and Produce report a partition's error code by throwing it, so the fan-out has to + /// record that as a rejection rather than a failure or those two lose their retry. + @Test("A thrown broker code is recorded as a rejection, so it still retries") + func aThrownBrokerCodeStaysRetriable() { + let thrown = KafkaError.broker(code: KafkaErrorCode.notLeaderOrFollower, api: "Fetch") + guard case .rejected(let code) = KafkaPartitionOutcome.outcome(for: thrown) else { + Issue.record("a thrown broker code must become a rejection") + return + } + #expect(code == KafkaErrorCode.notLeaderOrFollower) + #expect(KafkaErrorCode.retryAction(for: code) == .resolveLeaderAgain) + } + + @Test("A transport failure is recorded as a failure, and is not retried as a moved leader") + func aTransportFailureIsNotARejection() { + let thrown = KafkaError.brokerUnreachable(nodeId: 2, address: "kafka-2:9092", reason: "timed out") + guard case .failed = KafkaPartitionOutcome.outcome(for: thrown) else { + Issue.record("an unreachable broker must not be recorded as a broker rejection") + return + } + } + + // MARK: - Partition filters + + @Test("A partition filter that names every partition is kept") + func aValidPartitionFilterIsKept() throws { + let selected = try KafkaBrowseEngine.resolvePartitions([2, 0], available: [0, 1, 2], topic: "orders") + #expect(selected == [0, 2]) + } + + @Test("No filter reads every partition") + func noFilterReadsEverything() throws { + let selected = try KafkaBrowseEngine.resolvePartitions(nil, available: [0, 1, 2], topic: "orders") + #expect(selected == [0, 1, 2]) + } + + /// Filtering a partition the topic does not have used to return an empty page, which is + /// exactly what an empty topic returns. The topic name has refused to work that way since + /// the driver shipped and a partition number is no different. + @Test("A partition the topic does not have is reported, not dropped") + func anUnknownPartitionIsReported() { + #expect(throws: KafkaError.self) { + _ = try KafkaBrowseEngine.resolvePartitions([0, 99], available: [0, 1, 2], topic: "orders") + } + } + + // MARK: - Which end of the window is the page + + private func record(partition: Int32, offset: Int64) -> KafkaRecord { + KafkaRecord( + offset: offset, + timestamp: offset, + timestampIsLogAppendTime: false, + key: nil, + value: nil, + headers: [], + partition: partition + ) + } + + /// The defect that made `FROM NEWEST` a lie on any multi-partition topic. Each partition is + /// stepped back by the page size and read forward, so the merged run holds several pages and + /// its newest records are at the end. + @Test("A tail scan shows the newest records of its window, not the oldest") + func aTailScanTakesTheTail() { + let merged = KafkaRecordOrdering.merge((0 ..< 9).map { record(partition: Int32($0 % 3), offset: Int64($0)) }) + let page = KafkaRecordOrdering.page(merged, skip: 0, limit: 3, readsBackward: true) + #expect(page.map(\.offset) == [6, 7, 8]) + } + + @Test("A forward scan shows the oldest records of its window") + func aForwardScanTakesTheHead() { + let merged = KafkaRecordOrdering.merge((0 ..< 9).map { record(partition: Int32($0 % 3), offset: Int64($0)) }) + let page = KafkaRecordOrdering.page(merged, skip: 0, limit: 3, readsBackward: false) + #expect(page.map(\.offset) == [0, 1, 2]) + } + + /// Page two of a tail scan is the page before page one, so it steps further back from the + /// same end rather than further forward from the same start. + @Test("Paging a tail scan walks backwards") + func pagingATailScanWalksBackwards() { + let merged = KafkaRecordOrdering.merge((0 ..< 9).map { record(partition: Int32($0 % 3), offset: Int64($0)) }) + #expect(KafkaRecordOrdering.page(merged, skip: 3, limit: 3, readsBackward: true).map(\.offset) == [3, 4, 5]) + #expect(KafkaRecordOrdering.page(merged, skip: 6, limit: 3, readsBackward: true).map(\.offset) == [0, 1, 2]) + } + + @Test("Paging a forward scan walks forwards") + func pagingAForwardScanWalksForwards() { + let merged = KafkaRecordOrdering.merge((0 ..< 9).map { record(partition: Int32($0 % 3), offset: Int64($0)) }) + #expect(KafkaRecordOrdering.page(merged, skip: 3, limit: 3, readsBackward: false).map(\.offset) == [3, 4, 5]) + #expect(KafkaRecordOrdering.page(merged, skip: 6, limit: 3, readsBackward: false).map(\.offset) == [6, 7, 8]) + } + + /// With one partition the two directions select the same rows, which is why every existing + /// assertion passed while the multi-partition case was wrong. + @Test("One partition cannot tell the two directions apart") + func onePartitionHidesTheDirection() { + let merged = KafkaRecordOrdering.merge((0 ..< 5).map { record(partition: 0, offset: Int64($0)) }) + let backward = KafkaRecordOrdering.page(merged, skip: 0, limit: 5, readsBackward: true) + let forward = KafkaRecordOrdering.page(merged, skip: 0, limit: 5, readsBackward: false) + #expect(backward.map(\.offset) == forward.map(\.offset)) + } + + @Test("A page bigger than the window is the whole window") + func aPageBiggerThanTheWindowIsTheWindow() { + let merged = KafkaRecordOrdering.merge((0 ..< 3).map { record(partition: 0, offset: Int64($0)) }) + #expect(KafkaRecordOrdering.page(merged, skip: 0, limit: 10, readsBackward: true).count == 3) + #expect(KafkaRecordOrdering.page(merged, skip: 10, limit: 10, readsBackward: true).isEmpty) + } +} diff --git a/docs/databases/kafka.mdx b/docs/databases/kafka.mdx index 7ebf28de2..91a38e4c9 100644 --- a/docs/databases/kafka.mdx +++ b/docs/databases/kafka.mdx @@ -28,7 +28,7 @@ record format every message has used since. | **Security Protocol** | `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, or `SASL_SSL` | | **SASL Mechanism** | `PLAIN`, `SCRAM-SHA-256`, or `SCRAM-SHA-512`, for the two SASL protocols | | **Username** / **Password** | The SASL credentials | -| **Broker Addresses** | Whether to dial the addresses the cluster advertises, or stay on the bootstrap | +| **Broker Addresses** | Whether to dial the addresses the cluster advertises, or stay on the bootstrap. Leave this on **Use the addresses the cluster advertises** unless those addresses are unreachable from this Mac | | **Connect Timeout** | Seconds to wait for a broker. Default `10` | There is no Database field. A connection is one cluster, and the sidebar shows its topics directly. @@ -94,6 +94,13 @@ DESCRIBE GROUP "order-processor" `DESCRIBE GROUP` is the lag report: committed offset, end offset, and the gap, per partition. A partition the group has never committed shows an empty lag rather than a number counted from zero. +A group the cluster does not hold is reported by name instead of coming back empty. + +`SHOW GROUPS` asks every broker, because a broker knows only the groups it coordinates. When one +broker cannot be reached the list is marked truncated rather than presented as the whole cluster's. + +In `SHOW BROKERS`, **takes_admin_requests** marks the broker a `DROP TOPIC` goes to. On a KRaft +cluster any broker forwards an admin request, and the one marked here moves between runs. ## Producing a message @@ -133,7 +140,8 @@ can still appear in the list for a moment. Truncate is not offered: Kafka remove retention or by an offset per partition, neither of which empties a topic the way Truncate means. Paging jumps are unavailable. Page two continues from where page one stopped, so a topic being -written to while you read it will not repeat or skip a message. +written to while you read it will not repeat or skip a message. A tail browse pages backwards +through the log, one page older each time. Row counts are approximate. The count is the end offset minus the start offset, which overcounts where retention has removed messages from the middle of a compacted topic. @@ -141,8 +149,8 @@ where retention has removed messages from the middle of a compacted topic. Schema Registry is not read. A topic whose values are Avro or Protobuf shows the raw bytes, including the five-byte Confluent wire-format prefix. -Creating and deleting topics, editing topic configuration, and resetting consumer group offsets are -not available. +Creating a topic, editing topic configuration, and resetting consumer group offsets are not +available. Deleting a topic is, as the paragraph above describes. ## Troubleshooting @@ -160,6 +168,25 @@ one protocol per port, so use the port configured for TLS. The mechanism does not match the cluster. Set **SASL Mechanism** to one of the names in the message. +### The broker rejected ListOffsets for partition …: this broker no longer leads the partition + +The partition moved to another broker while the request was in flight, and it moved again before +the retry. Run `DESCRIBE TOPIC "…"` to see the current leader of each partition. A cluster that is +rebalancing settles on its own; one that keeps moving has a broker leaving and rejoining. + +### Broker … advertises …, which could not be reached: … + +The address in the message is what the cluster told this Mac to use, and nothing is listening there. +Check that the broker's `advertised.listeners` names an address reachable from here, not a +container hostname or a private IP. Where it cannot be changed, set **Broker Addresses** to only use +the bootstrap address. + +### Partition … of … is led by another broker + +Only the leader of a partition can answer for it, and this connection is pinned to one broker. +Set **Broker Addresses** back to the advertised addresses, or forward every broker's port through +the tunnel so the advertised addresses resolve. + ### Requests time out, or a topic loads on a tunnel but its messages do not The cluster is advertising addresses that are not reachable from here. Behind an diff --git a/project.yml b/project.yml index cbd237234..e29b67b18 100644 --- a/project.yml +++ b/project.yml @@ -596,9 +596,11 @@ targets: - Plugins/KafkaDriverPlugin/KafkaApiKey.swift - Plugins/KafkaDriverPlugin/KafkaBrowseEngine.swift - Plugins/KafkaDriverPlugin/KafkaCluster.swift + - Plugins/KafkaDriverPlugin/KafkaCluster+Routing.swift - Plugins/KafkaDriverPlugin/KafkaConnection.swift - Plugins/KafkaDriverPlugin/KafkaDeleteTopicsRequest.swift - Plugins/KafkaDriverPlugin/KafkaFetchRequest.swift + - Plugins/KafkaDriverPlugin/KafkaFindCoordinatorRequest.swift - Plugins/KafkaDriverPlugin/KafkaGroupsRequest.swift - Plugins/KafkaDriverPlugin/KafkaMetadataRequest.swift - Plugins/KafkaDriverPlugin/KafkaOffsetsRequest.swift diff --git a/scripts/ci/kafka-protocol-probe.py b/scripts/ci/kafka-protocol-probe.py index 0e3ad1d14..93e404aa5 100755 --- a/scripts/ci/kafka-protocol-probe.py +++ b/scripts/ci/kafka-protocol-probe.py @@ -128,22 +128,30 @@ def check_api_versions(broker): def check_version_floor(apis): """The plugin negotiates against the broker's advertised range instead of hardcoding a - version, because Kafka 4.x removed the oldest version of several APIs.""" - # api key -> (name, the version the plugin tops out at) - ceilings = { - 0: ("Produce", 9), 1: ("Fetch", 12), 2: ("ListOffsets", 7), 3: ("Metadata", 12), - 9: ("OffsetFetch", 8), 10: ("FindCoordinator", 4), 15: ("DescribeGroups", 5), - 16: ("ListGroups", 4), 17: ("SaslHandshake", 1), 36: ("SaslAuthenticate", 2), + version, because Kafka 4.x removed the oldest version of several APIs. + + So the check is that a version can be AGREED, not that the plugin's ceiling is one the + broker offers. KafkaApiVersionTable.negotiated takes min(broker high, our ceiling) and only + fails when that lands below our floor, which is the whole point of negotiating down. Testing + the ceiling for membership instead reported five failures against any broker older than the + newest, and a real regression is then indistinguishable from that noise.""" + # api key -> (name, the plugin's floor, the plugin's ceiling) + bounds = { + 0: ("Produce", 3, 9), 1: ("Fetch", 4, 12), 2: ("ListOffsets", 1, 7), + 3: ("Metadata", 1, 12), 9: ("OffsetFetch", 1, 8), 10: ("FindCoordinator", 0, 4), + 15: ("DescribeGroups", 0, 5), 16: ("ListGroups", 0, 4), 17: ("SaslHandshake", 0, 1), + 36: ("SaslAuthenticate", 0, 2), 20: ("DeleteTopics", 1, 5), } - for key, (name, ceiling) in ceilings.items(): + for key, (name, floor, ceiling) in bounds.items(): if key not in apis: check(f"{name} is offered by the broker", False, "the broker does not advertise it") continue low, high = apis[key] + agreed = min(high, ceiling) check( - f"{name} v{ceiling} is inside the broker's v{low}..v{high}", - low <= ceiling <= high, - f"the plugin's ceiling v{ceiling} is outside what this broker accepts", + f"{name} negotiates to v{agreed} inside the broker's v{low}..v{high}", + agreed >= low and agreed >= floor, + f"the plugin speaks v{floor}..v{ceiling} and this broker speaks v{low}..v{high}", ) @@ -293,6 +301,189 @@ def check_sasl_handshake_is_never_flexible(apis): ) +def check_find_coordinator_field_order(broker, apis, brokers): + """FindCoordinator moved its error code at v4 and the two orders are not distinguishable + from a successful parse alone. + + Up to v3 the body opens with the error and then names the node. v4 (KIP-699) made the + request batched and put the error at the END of each coordinator entry, after the address. + Reading v4 in the v3 order takes the key's length prefix for an error code and a slice of + the host for a node id, so it yields a plausible broker rather than a parse failure. Both + orders are asserted here because both are hand-coded in KafkaFindCoordinatorRequest. + + What is asserted is the FIELD ORDER, not that a coordinator exists. A cluster where no group + has ever committed has no __consumer_offsets topic yet and answers 15 + COORDINATOR_NOT_AVAILABLE with node -1, which is correct and which the driver retries. The + discriminating evidence is that the key echoes back and the reply is consumed exactly.""" + available = (0, 15) + known = {node for node, _, _ in brokers} if brokers else set() + high = apis.get(10, (0, 0))[1] + + if high >= 4: + body = struct.pack(">b", 0) + b"\x02" + write_compact_string("tp-probe-group") + b"\x00" + resp = broker.send(10, 4, body) + i = skip_tags(resp, 4) + 4 # header tags, throttle_time_ms + count, i = uvarint(resp, i) + key, i = compact_string(resp, i) + node_id = struct.unpack(">i", resp[i:i + 4])[0] + i += 4 + host, i = compact_string(resp, i) + i += 4 # port + error_code = struct.unpack(">h", resp[i:i + 2])[0] + i += 2 + _, i = compact_string(resp, i) # error_message + i = skip_tags(resp, i) + i = skip_tags(resp, i) + addressed = error_code != 0 or not known or node_id in known + check( + "FindCoordinator v4 names the coordinator before its error code", + count == 2 and key == "tp-probe-group" and error_code in available + and addressed and i == len(resp), + f"key={key} node={node_id} host={host} error={error_code}, " + f"consumed {i} of {len(resp)} bytes", + ) + + if high >= 3: + body = write_compact_string("tp-probe-group") + struct.pack(">b", 0) + b"\x00" + resp = broker.send(10, 3, body) + i = skip_tags(resp, 4) + 4 + error_code = struct.unpack(">h", resp[i:i + 2])[0] + i += 2 + _, i = compact_string(resp, i) # error_message + node_id = struct.unpack(">i", resp[i:i + 4])[0] + i += 4 + _, i = compact_string(resp, i) # host + i += 4 # port + i = skip_tags(resp, i) + addressed = error_code != 0 or not known or node_id in known + check( + "FindCoordinator v3 answers with its error code first", + error_code in available and addressed and i == len(resp), + f"node={node_id} error={error_code}, consumed {i} of {len(resp)} bytes", + ) + + +def check_group_request_shapes(broker, apis): + """ListGroups v4, DescribeGroups v5 and OffsetFetch v8 are each hand-encoded and none of + them was covered here before. + + The assertion is that the reply parses to exactly its own length. A version gate written at + the wrong number does not raise: it leaves the reader a field ahead or behind, which reads + as plausible values and a buffer that ends in the wrong place.""" + if apis.get(16, (0, 0))[1] >= 4: + resp = broker.send(16, 4, b"\x01" + b"\x00") # empty states filter, tags + i = skip_tags(resp, 4) + 4 + 2 # header tags, throttle, error_code + count, i = uvarint(resp, i) + for _ in range(max(0, count - 1)): + _, i = compact_string(resp, i) # group_id + _, i = compact_string(resp, i) # protocol_type + _, i = compact_string(resp, i) # group_state, v4 only + i = skip_tags(resp, i) + i = skip_tags(resp, i) + check("ListGroups v4 carries a group state per group", i == len(resp), + f"consumed {i} of {len(resp)} bytes") + + if apis.get(15, (0, 0))[1] >= 5: + body = b"\x02" + write_compact_string("tp-probe-group") + b"\x00" + b"\x00" + resp = broker.send(15, 5, body) + i = skip_tags(resp, 4) + 4 + count, i = uvarint(resp, i) + for _ in range(max(0, count - 1)): + i += 2 # error_code + for _ in range(4): # group_id, state, protocol_type, protocol + _, i = compact_string(resp, i) + members, i = uvarint(resp, i) + for _ in range(max(0, members - 1)): + for _ in range(4): # member_id, instance_id, client_id, host + _, i = compact_string(resp, i) + for _ in range(2): # metadata, assignment + size, i = uvarint(resp, i) + i += max(0, size - 1) + i = skip_tags(resp, i) + i += 4 # authorized_operations + i = skip_tags(resp, i) + i = skip_tags(resp, i) + check("DescribeGroups v5 carries an instance id and authorized operations", + i == len(resp), f"consumed {i} of {len(resp)} bytes") + + if apis.get(9, (0, 0))[1] >= 8: + body = (b"\x02" + write_compact_string("tp-probe-group") + b"\x00" + b"\x00" + + b"\x00" + b"\x00") + resp = broker.send(9, 8, body) + i = skip_tags(resp, 4) + 4 + groups, i = uvarint(resp, i) + for _ in range(max(0, groups - 1)): + _, i = compact_string(resp, i) # group_id + topics, i = uvarint(resp, i) + for _ in range(max(0, topics - 1)): + _, i = compact_string(resp, i) + parts, i = uvarint(resp, i) + for _ in range(max(0, parts - 1)): + i += 4 + 8 + 4 # index, offset, leader_epoch + _, i = compact_string(resp, i) # metadata + i += 2 # error_code + i = skip_tags(resp, i) + i = skip_tags(resp, i) + i += 2 # group-level error_code + i = skip_tags(resp, i) + i = skip_tags(resp, i) + check("OffsetFetch v8 groups its topics under a group and ends with a group error", + i == len(resp), f"consumed {i} of {len(resp)} bytes") + + +def check_list_offsets_is_per_partition(broker, apis, topic): + """ListOffsets answers per partition, which is why the driver splits one request per leader. + + The check is both that the encoding stays in step and that a partition's error arrives + INSIDE a successful response. A broker that is not the leader of a partition reports it + here, not as a request-level failure, and reading it as one is issue #2993.""" + if apis.get(2, (0, 0))[1] < 7: + return + body = struct.pack(">i", -1) + struct.pack(">b", 1) + body += b"\x02" + write_compact_string(topic) + b"\x02" + body += struct.pack(">i", 0) + struct.pack(">i", -1) + struct.pack(">q", -1) + b"\x00" + body += b"\x00" + b"\x00" + resp = broker.send(2, 7, body) + i = skip_tags(resp, 4) + 4 + topics, i = uvarint(resp, i) + saw_partition = False + for _ in range(max(0, topics - 1)): + _, i = compact_string(resp, i) + parts, i = uvarint(resp, i) + for _ in range(max(0, parts - 1)): + i += 4 + 2 + 8 + 8 + 4 # index, error_code, timestamp, offset, leader_epoch + i = skip_tags(resp, i) + saw_partition = True + i = skip_tags(resp, i) + i = skip_tags(resp, i) + check("ListOffsets v7 answers with an error code per partition", + saw_partition and i == len(resp), f"consumed {i} of {len(resp)} bytes") + + +def check_delete_topics_shape(broker, apis): + """DeleteTopics v5 still names topics; v6 switched to a 16-byte topic UUID, which is a + different request rather than a bigger one. v5 also added a broker-supplied message the + reader has to consume even though the app does not show it.""" + if apis.get(20, (0, 0))[1] < 5: + return + body = b"\x02" + write_compact_string("tp-probe-no-such-topic-9e3f") + body += struct.pack(">i", 5000) + b"\x00" + resp = broker.send(20, 5, body) + i = skip_tags(resp, 4) + 4 + count, i = uvarint(resp, i) + codes = [] + for _ in range(max(0, count - 1)): + _, i = compact_string(resp, i) # name, nullable from v6 on + codes.append(struct.unpack(">h", resp[i:i + 2])[0]) + i += 2 + _, i = compact_string(resp, i) # error_message + i = skip_tags(resp, i) + i = skip_tags(resp, i) + check("DeleteTopics v5 names the topic and carries an error message", + codes == [3] and i == len(resp), + f"codes={codes}, consumed {i} of {len(resp)} bytes; 3 is UNKNOWN_TOPIC_OR_PARTITION") + + def main(): host = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1" port = int(sys.argv[2]) if len(sys.argv) > 2 else 9092 @@ -314,6 +505,10 @@ def main(): print(f" broker(s): {', '.join(f'{n}@{h}:{p}' for n, h, p in brokers)}") check_fetch_is_name_based_through_v12(broker, "tp-probe-crc") check_produce_crc(broker, apis) + check_find_coordinator_field_order(broker, apis, brokers) + check_group_request_shapes(broker, apis) + check_list_offsets_is_per_partition(broker, apis, "tp-probe-crc") + check_delete_topics_shape(broker, apis) finally: broker.close() diff --git a/scripts/kafka-test-broker.sh b/scripts/kafka-test-broker.sh index ccb4ae277..e4fbe749a 100755 --- a/scripts/kafka-test-broker.sh +++ b/scripts/kafka-test-broker.sh @@ -3,62 +3,124 @@ # kafka-test-broker.sh: start or stop the broker KafkaIntegrationTests runs against. # # The Kafka unit suites are pure logic and never open a socket. The integration suite drives -# the real driver end to end, so it needs a real broker, and it skips itself unless -# TABLEPRO_KAFKA_TEST_BOOTSTRAP names one. +# the real driver end to end, so it needs a real broker, and it skips itself unless one +# answers on TABLEPRO_KAFKA_TEST_BOOTSTRAP. # # Usage: -# scripts/kafka-test-broker.sh up # start it and print the exports -# scripts/kafka-test-broker.sh down # remove it -# scripts/kafka-test-broker.sh env # print the exports for a broker already running +# scripts/kafka-test-broker.sh up [--brokers N] # start it and print the exports +# scripts/kafka-test-broker.sh down # remove it +# scripts/kafka-test-broker.sh env # print the exports for one already running # # Then: # eval "$(scripts/kafka-test-broker.sh env)" # .claude/skills/fix-issue/scripts/verify.sh test KafkaIntegrationTests # -# Two listeners, not one. The CLI runs inside the container and must reach the broker at its +# Two listeners, not one. The CLI runs inside a container and must reach the cluster at its # INTERNAL advertised address, while the driver connects from the host to the EXTERNAL one. A # single listener advertising "localhost" satisfies exactly one of those and fails the other # with a node-assignment timeout. +# +# --brokers takes a count because one broker cannot reproduce a routing bug. On a single-node +# cluster that node leads every partition and coordinates every group, so a client that sends +# every request to whichever broker it happens to hold is indistinguishable from a correct one. +# That is why the suite passed for the whole life of issue #2993, and why three is the default +# the routing tests ask for: enough that a topic's partitions land on brokers the client did not +# connect to. set -euo pipefail CONTAINER="${TABLEPRO_KAFKA_TEST_CONTAINER:-tp-kafka-it}" PORT="${TABLEPRO_KAFKA_TEST_PORT:-19092}" IMAGE="${TABLEPRO_KAFKA_TEST_IMAGE:-apache/kafka:latest}" +NETWORK="${CONTAINER}-net" +# Fixed so every node of one cluster formats the same storage id. Generated ids differ per +# container and the nodes then refuse to form a quorum. +CLUSTER_ID="${TABLEPRO_KAFKA_TEST_CLUSTER_ID:-5L6g3nShT-eMCtK--X86sw}" + +BROKERS=1 +ACTION="${1:-up}" +shift || true +while [ $# -gt 0 ]; do + case "$1" in + --brokers) + BROKERS="${2:-1}" + shift 2 + ;; + *) + echo "Unknown option: $1" >&2 + exit 2 + ;; + esac +done + +if ! [ "$BROKERS" -ge 1 ] 2>/dev/null; then + echo "--brokers needs a count of 1 or more." >&2 + exit 2 +fi + +container_name() { + if [ "$1" -eq 1 ]; then echo "$CONTAINER"; else echo "${CONTAINER}-$1"; fi +} + +external_port() { + echo $((PORT + $1 - 1)) +} print_env() { echo "export TABLEPRO_KAFKA_TEST_BOOTSTRAP=127.0.0.1:${PORT}" echo "export TABLEPRO_KAFKA_TEST_CONTAINER=${CONTAINER}" } -case "${1:-up}" in +remove_all() { + # A bounded sweep rather than a name glob, so this never reaches a container the script + # did not start. + for node in $(seq 1 16); do + docker rm -f "$(container_name "$node")" >/dev/null 2>&1 || true + done + docker network rm "$NETWORK" >/dev/null 2>&1 || true +} + +case "$ACTION" in up) if ! docker info >/dev/null 2>&1; then echo "Docker is not running." >&2 exit 1 fi - docker rm -f "$CONTAINER" >/dev/null 2>&1 || true - docker run -d --name "$CONTAINER" -p "${PORT}:${PORT}" \ - -e KAFKA_NODE_ID=1 \ - -e KAFKA_PROCESS_ROLES=broker,controller \ - -e "KAFKA_LISTENERS=INTERNAL://:9092,EXTERNAL://:${PORT},CONTROLLER://:9093" \ - -e "KAFKA_ADVERTISED_LISTENERS=INTERNAL://localhost:9092,EXTERNAL://localhost:${PORT}" \ - -e KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL \ - -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \ - -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT \ - -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \ - -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \ - -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \ - -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \ - -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \ - -e KAFKA_AUTO_CREATE_TOPICS_ENABLE=false \ - "$IMAGE" >/dev/null + remove_all + docker network create "$NETWORK" >/dev/null 2>&1 || true + + voters="" + for node in $(seq 1 "$BROKERS"); do + voters="${voters:+$voters,}${node}@$(container_name "$node"):9093" + done + + for node in $(seq 1 "$BROKERS"); do + name="$(container_name "$node")" + port="$(external_port "$node")" + docker run -d --name "$name" --network "$NETWORK" -p "${port}:${port}" \ + -e CLUSTER_ID="$CLUSTER_ID" \ + -e KAFKA_NODE_ID="$node" \ + -e KAFKA_PROCESS_ROLES=broker,controller \ + -e "KAFKA_LISTENERS=INTERNAL://:9092,EXTERNAL://:${port},CONTROLLER://:9093" \ + -e "KAFKA_ADVERTISED_LISTENERS=INTERNAL://${name}:9092,EXTERNAL://127.0.0.1:${port}" \ + -e KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL \ + -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \ + -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT \ + -e "KAFKA_CONTROLLER_QUORUM_VOTERS=$voters" \ + -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR="$BROKERS" \ + -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR="$BROKERS" \ + -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \ + -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \ + -e KAFKA_AUTO_CREATE_TOPICS_ENABLE=false \ + "$IMAGE" >/dev/null + done - printf 'Waiting for the broker' >&2 - for _ in $(seq 1 60); do - if docker exec "$CONTAINER" /opt/kafka/bin/kafka-topics.sh \ - --bootstrap-server localhost:9092 --list >/dev/null 2>&1; then + printf 'Waiting for %s broker(s)' "$BROKERS" >&2 + for _ in $(seq 1 90); do + ready=$(docker exec "$CONTAINER" /opt/kafka/bin/kafka-broker-api-versions.sh \ + --bootstrap-server localhost:9092 2>/dev/null | grep -c 'id:' || true) + if [ "${ready:-0}" -ge "$BROKERS" ]; then echo " ready." >&2 print_env exit 0 @@ -72,14 +134,14 @@ up) exit 1 ;; down) - docker rm -f "$CONTAINER" >/dev/null 2>&1 || true + remove_all echo "Removed $CONTAINER." >&2 ;; env) print_env ;; *) - echo "Usage: $0 [up|down|env]" >&2 + echo "Usage: $0 [up [--brokers N]|down|env]" >&2 exit 2 ;; esac