RustToMetal commented on code in PR #4553: URL: https://github.com/apache/iggy/pull/4553#discussion_r4286268228
########## foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift: ########## @@ -0,0 +1,403 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// MARK: - Messages + +extension IggyClient { + /// One sync and one retry: a poll the coordinator fences for a stale + /// assignment syncs it again and polls once more. + private static let groupPollMaxAttempts = 2 + + /// Sends a batch of messages to one partition. Delivery is at least once. + /// + /// The server routes explicit partitions only, so `balanced` and + /// `messagesKey` partitioning are resolved on the client: balanced + /// round-robins over the topic's partitions, a key is hashed with XXH32. + /// A zero message id is replaced with a random one for this send only; + /// `messages` is not changed. A caller that retries a failed send and + /// wants each attempt to carry the same ids sets them first, for example + /// with ``UInt128Value/random()``. + /// + /// The returned confirmations may be empty: the server can commit a batch + /// it has no offsets to describe. + @discardableResult + public func sendMessages( + streamID: Identifier, topicID: Identifier, partitioning: Partitioning = .balanced, messages: [IggyMessage] + ) async throws -> SendMessagesResponse { + try await requireAuthenticated() + // Checked before a balanced or keyed send reads the topic. + guard !messages.isEmpty else { + throw IggyError(.invalidMessagesCount, context: "a batch needs at least one message") + } + // The key is hashed to a partition before the batch is encoded, so its + // length is checked first. + try partitioning.validate() + let partitionID = try await resolvePartition(partitioning, streamID: streamID, topicID: topicID) + let outgoing = messages.map { message in + Batch.OutgoingMessage( + id: message.id.isZero ? .random() : message.id, originTimestamp: message.originTimestamp.microseconds, payload: message.payload, + userHeaders: message.rawUserHeaders) + } + let payload = try Batch.encodeSend(streamID: streamID, topicID: topicID, partitioning: .partition(partitionID), messages: outgoing) + let body = try await sendWithPartitionContext(streamID: streamID, topicID: topicID, partitionID: partitionID, payload: payload) + // The batch committed once the reply arrived. An unreadable + // confirmation must not make a retrying caller write it twice. + return (try? Responses.sendConfirmations(body[...])) ?? SendMessagesResponse() + } + + /// Polls up to `count` messages. + /// + /// A consumer-group poll with `partitionID` nil takes the next of the + /// member's assigned partitions, round-robin, syncing the assignment from + /// the coordinator when needed. A member holding no partitions gets an + /// empty result whose partition id is ``PolledMessages/noAssignedPartition``, + /// and a client that is not a member gets + /// ``IggyErrorCode/consumerGroupMemberNotFound``, so it can join again. + /// + /// With `autoCommit` the server stores the offset of the last polled + /// message as consumed. A `strategy` with a ``PollingStrategy/context`` + /// is refused once its partition was recreated. + public func pollMessages( + streamID: Identifier, topicID: Identifier, partitionID: UInt32?, consumer: Consumer = .default, strategy: PollingStrategy, count: UInt32, + autoCommit: Bool + ) async throws -> PolledMessages { + try await pollMessages(streamID: streamID, topicID: topicID, partitionID: partitionID, consumer: consumer, count: count, autoCommit: autoCommit) { + _ in strategy + } + } + + /// ``pollMessages(streamID:topicID:partitionID:consumer:strategy:count:autoCommit:)`` + /// with the strategy chosen once the partition is known, so a caller can + /// continue each partition from its own position. + public func pollMessages( + streamID: Identifier, topicID: Identifier, partitionID: UInt32?, consumer: Consumer = .default, count: UInt32, autoCommit: Bool, + strategyFor: @Sendable (UInt32) -> PollingStrategy + ) async throws -> PolledMessages { + try await requireAuthenticated() + if consumer.kind == .consumerGroup, partitionID == nil { + return try await pollGroup(streamID: streamID, topicID: topicID, consumer: consumer, count: count, autoCommit: autoCommit, strategyFor: strategyFor) + } + let strategy = strategyFor(partitionID ?? 0) + let payload = try Requests.pollMessages( + consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, strategy: strategy, count: count, autoCommit: autoCommit) + return try Responses.polledMessages(try await routedPoll(payload, context: strategy.context)[...]) + } + + // MARK: Partition contexts + + /// Sends under the partition's cached context, or one read from its + /// route. A history refusal means the context is stale: the topic's + /// contexts are dropped and one fresh route is tried. + private func sendWithPartitionContext(streamID: Identifier, topicID: Identifier, partitionID: UInt32, payload: [UInt8]) async throws -> [UInt8] { + let key = ConsumerGroupState.TopicKey(stream: streamID, topic: topicID) + let state = core.groupState + var context = state.partitionContext(key, partitionID: partitionID) + var refreshAvailable = true + while true { + let captured: PartitionContext + if let context { + captured = context + } else { + let body = try await send(.getPartitionRoute, Requests.partitionRoute(streamID: streamID, topicID: topicID, partitionID: partitionID)) + captured = try Responses.partitionRoute(body[...]) + state.setPartitionContext(key, partitionID: partitionID, captured) + } + do { + return try await core.exchange(code: CommandCode.sendMessages.rawValue, payload: payload, context: captured) Review Comment: Done in d5c1dbbfa: `RetryOutcome.observe` maps `requestTooOld` to `transientNotCommitted` and marks the outcome uncertain. Test: `retryOutcomeReportsARequestTooOldAsUnknown`. ########## foreign/swift/Sources/Iggy/Client/IggyClient.swift: ########## @@ -167,10 +172,23 @@ public final class IggyClient: Sendable { /// raw reply body. For commands without a typed method; the caller owns /// the wire format. Session-control codes are rejected because they must /// go through ``login(username:password:)`` and ``logout()``. + /// + /// Partition requests carry the partition context their typed call + /// sends: `SendMessages` with an explicit partition id, `PollMessages`, + /// `StoreConsumerOffset`, and `DeleteConsumerOffset` take the route of + /// ``sendMessages(streamID:topicID:partitioning:messages:)``, + /// ``pollMessages(streamID:topicID:partitionID:consumer:strategy:count:autoCommit:)``, + /// and the offset calls. A raw send with balanced or key partitioning + /// fails with ``IggyErrorCode/featureUnavailable``, and a partition + /// payload that does not decode fails with ``IggyErrorCode/invalidCommand``, + /// before anything is sent. public func sendRawRequest(code: UInt32, payload: [UInt8]) async throws -> [UInt8] { guard !CommandCode.isSessionControl(code) else { Review Comment: Fixed in d5c1dbbfa. `BindSession` is in `isSessionControl`, so a raw one is refused. Raw `GetPollRouting`, `PollMessagesOnPrimary`, and `GetConsumerOffsetRouting` are sent once with `exchangeOnce`, with no reconnect replay, as `is_poll_routing_code` routes them in Rust (a dropped client reconnects first, like a typed call). Tests: `rawPartitionRequestsAreRoutedLikeTypedOnes` for the refused bind and `rawRoutingRequestsAreSentOnce`. ########## foreign/swift/Sources/Iggy/Client/PollRouter.swift: ########## @@ -0,0 +1,370 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import NIOConcurrencyHelpers + +/// Routes polls and consumer-offset writes to the primary of their +/// partition, a port of `PollRouter` in `core/sdk/src/driver/poll_router.rs`. +/// +/// A request first asks the coordinator, the connection the client signed in +/// on, for its route (`GetPollRouting` or `GetConsumerOffsetRouting`): the +/// partition primary and the partition context the request carries. On a +/// single node the request then goes over the coordinator connection. In a +/// cluster it goes over a data connection to the primary, bound to the +/// coordinator's session with `BindSession`, so consumer-group membership +/// stays with the coordinator. Routes are cached until a request on them +/// fails or the session is replaced. +final class PollRouter: Sendable { + /// Bounds one routed request across its lookups, attempts, and pauses. + static let pollTimeout: Duration = .seconds(30) + static let retryInterval: Duration = .milliseconds(50) + static let maxRetryInterval: Duration = .seconds(1) + static let rosterReadTimeout: Duration = .seconds(5) + static let maxCachedRoutes = 4096 + static let maxDataConnections = 256 + + /// A route is cached per lookup code and the request prefix that names + /// the partition and the consumer. + struct RouteKey: Hashable, Sendable { + let code: CommandCode + let prefix: ArraySlice<UInt8> + } + + private struct Route: Sendable { + /// The fence of the session the route was read under. + let fence: UInt128Value + let endpoint: String + let consumerSession: Responses.SessionIdentity + let context: PartitionContext + } + + private struct DataConnection { + let connection: any IggyConnection + var consumerSession: Responses.SessionIdentity? + var usable: Bool + } + + /// One data connection per primary. Its lock keeps the lockstep stream to + /// one request at a time. + private final class Slot: Sendable { + let lock = AsyncMutex() + let connection = NIOLockedValueBox<DataConnection?>(nil) + } + + private struct State { + /// The client id of the current session. A reset replaces it, so a + /// route or a connection of a replaced session is never published + /// or used. + var fence: UInt128Value + /// Nodes in the roster; zero until a roster read succeeded. + var rosterSize = 0 + var routes: [RouteKey: Route] = [:] + var slots: [String: Slot] = [:] + var nextHeartbeat: ContinuousClock.Instant? + } + + private let state: NIOLockedValueBox<State> + + init(fence: UInt128Value) { + state = NIOLockedValueBox(State(fence: fence)) + } + + /// Records the roster size a leader check or a roster read reported. + func learnRosterSize(_ size: Int) { + state.withLockedValue { $0.rosterSize = size } + } + + /// Drops every route and data connection of the session that ended and + /// moves to the fence of the one that replaced it. A route or a + /// connection is published only under the lock after a fence check, so + /// nothing of the replaced session survives. + func clearSession(fence: UInt128Value) { + state.withLockedValue { state in + state.fence = fence + state.routes.removeAll() + state.slots.removeAll() + state.nextHeartbeat = nil + } + } + + /// Sends a poll or an offset write on its route. `context` is the + /// partition context the caller captured, else the route's is taken and + /// kept for every retry. + func send( + _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload: [UInt8], context: PartitionContext?, coordinator: ClientCore + ) async throws -> [UInt8] { + let clustered = try await isClustered(coordinator) + let clock = ContinuousClock() + let now = clock.now + let deadline = now + Self.pollTimeout + let routed: [UInt8]? = try await withDeadline(deadline) { [self] in + if clustered, heartbeatDue(at: now, interval: coordinator.configuration.heartbeatInterval) { + // The requests leave the coordinator connection idle, so it is + // pinged here once per heartbeat interval too. + _ = try await control(.ping, [], coordinator: coordinator) + } + var captured = context + var interval = Self.retryInterval + while true { + do { + if clustered { + return try await sendOnPrimary( + code, payload: payload, key: key, routePayload: routePayload, context: &captured, coordinator: coordinator) + } + return try await sendOnCoordinator( + code, payload: payload, key: key, routePayload: routePayload, context: &captured, coordinator: coordinator) + } catch let error as IggyError where error.code == .transientNotAccepted { + removeRoute(key) + guard clock.now + interval < deadline else { + throw error + } + try await Task.sleep(for: interval) + interval = min(interval * 2, Self.maxRetryInterval) + } + } + } + guard let routed else { + removeRoute(key) + throw IggyError(.transientNotCommitted, context: "command \(code.rawValue) did not complete within \(Self.pollTimeout)") + } + return routed + } + + /// Whether the roster has more than one node, reading it once when no + /// leader check reported it yet. + private func isClustered(_ coordinator: ClientCore) async throws -> Bool { + if state.withLockedValue({ $0.rosterSize }) == 0 { + let body: [UInt8]? = try await withDeadline(ContinuousClock.now + Self.rosterReadTimeout) { [self] in + try await control(.getClusterMetadata, [], coordinator: coordinator) + } + guard let body else { + throw IggyError(.transientNotAccepted, context: "reading the cluster roster took longer than \(Self.rosterReadTimeout)") + } + let metadata = try Responses.clusterMetadata(body[...]) + guard !metadata.nodes.isEmpty else { + throw IggyError(.transientNotAccepted, context: "the cluster roster is empty") + } + learnRosterSize(metadata.nodes.count) + } + return state.withLockedValue { $0.rosterSize } > 1 + } + + private func heartbeatDue(at now: ContinuousClock.Instant, interval: Duration) -> Bool { + state.withLockedValue { state in + let due = state.nextHeartbeat.map { now >= $0 } ?? false + if state.nextHeartbeat == nil || due { + state.nextHeartbeat = now + interval + } + return due + } + } + + /// One exchange on the coordinator connection. A connection that is gone + /// is brought back by a ping, which reconnects, and the exchange runs + /// once more. + private func control(_ code: CommandCode, _ payload: [UInt8], coordinator: ClientCore) async throws -> [UInt8] { + do { + return try await coordinator.exchangeOnce(code: code.rawValue, payload: payload, context: .empty) + } catch let error as IggyError where error.isReconnectable { + _ = try await coordinator.exchange(code: CommandCode.ping.rawValue, payload: []) + return try await coordinator.exchangeOnce(code: code.rawValue, payload: payload, context: .empty) + } + } + + /// The single-node path: the request goes over the coordinator connection + /// with the context of its route. + private func sendOnCoordinator( + _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload: [UInt8], context: inout PartitionContext?, coordinator: ClientCore + ) async throws -> [UInt8] { + let route = try await route(key, routePayload: routePayload, coordinator: coordinator) + let captured = context ?? route.context + context = captured + do { + if code == .pollMessagesOnPrimary { Review Comment: Fixed in d5c1dbbfa. Both paths now go through one `gradedLoss`: a request that lost its connection after it was written is `transientNotCommitted`, and one that found no connection or no session is `transientNotAccepted` and routed again. To tell the two apart, `TCPConnection` now fails a request on a connection that was already closed with `notConnected`, since none of it was written. Test: `aSingleNodePollLostAfterItWasWrittenIsAnUnknownOutcome`. ########## foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift: ########## @@ -0,0 +1,403 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// MARK: - Messages + +extension IggyClient { + /// One sync and one retry: a poll the coordinator fences for a stale + /// assignment syncs it again and polls once more. + private static let groupPollMaxAttempts = 2 + + /// Sends a batch of messages to one partition. Delivery is at least once. + /// + /// The server routes explicit partitions only, so `balanced` and + /// `messagesKey` partitioning are resolved on the client: balanced + /// round-robins over the topic's partitions, a key is hashed with XXH32. + /// A zero message id is replaced with a random one for this send only; + /// `messages` is not changed. A caller that retries a failed send and + /// wants each attempt to carry the same ids sets them first, for example + /// with ``UInt128Value/random()``. + /// + /// The returned confirmations may be empty: the server can commit a batch + /// it has no offsets to describe. + @discardableResult + public func sendMessages( + streamID: Identifier, topicID: Identifier, partitioning: Partitioning = .balanced, messages: [IggyMessage] + ) async throws -> SendMessagesResponse { + try await requireAuthenticated() + // Checked before a balanced or keyed send reads the topic. + guard !messages.isEmpty else { + throw IggyError(.invalidMessagesCount, context: "a batch needs at least one message") + } + // The key is hashed to a partition before the batch is encoded, so its + // length is checked first. + try partitioning.validate() + let partitionID = try await resolvePartition(partitioning, streamID: streamID, topicID: topicID) + let outgoing = messages.map { message in + Batch.OutgoingMessage( + id: message.id.isZero ? .random() : message.id, originTimestamp: message.originTimestamp.microseconds, payload: message.payload, + userHeaders: message.rawUserHeaders) + } + let payload = try Batch.encodeSend(streamID: streamID, topicID: topicID, partitioning: .partition(partitionID), messages: outgoing) + let body = try await sendWithPartitionContext(streamID: streamID, topicID: topicID, partitionID: partitionID, payload: payload) + // The batch committed once the reply arrived. An unreadable + // confirmation must not make a retrying caller write it twice. + return (try? Responses.sendConfirmations(body[...])) ?? SendMessagesResponse() + } + + /// Polls up to `count` messages. + /// + /// A consumer-group poll with `partitionID` nil takes the next of the + /// member's assigned partitions, round-robin, syncing the assignment from + /// the coordinator when needed. A member holding no partitions gets an + /// empty result whose partition id is ``PolledMessages/noAssignedPartition``, + /// and a client that is not a member gets + /// ``IggyErrorCode/consumerGroupMemberNotFound``, so it can join again. + /// + /// With `autoCommit` the server stores the offset of the last polled + /// message as consumed. A `strategy` with a ``PollingStrategy/context`` + /// is refused once its partition was recreated. + public func pollMessages( + streamID: Identifier, topicID: Identifier, partitionID: UInt32?, consumer: Consumer = .default, strategy: PollingStrategy, count: UInt32, + autoCommit: Bool + ) async throws -> PolledMessages { + try await pollMessages(streamID: streamID, topicID: topicID, partitionID: partitionID, consumer: consumer, count: count, autoCommit: autoCommit) { + _ in strategy + } + } + + /// ``pollMessages(streamID:topicID:partitionID:consumer:strategy:count:autoCommit:)`` + /// with the strategy chosen once the partition is known, so a caller can + /// continue each partition from its own position. + public func pollMessages( + streamID: Identifier, topicID: Identifier, partitionID: UInt32?, consumer: Consumer = .default, count: UInt32, autoCommit: Bool, + strategyFor: @Sendable (UInt32) -> PollingStrategy + ) async throws -> PolledMessages { + try await requireAuthenticated() + if consumer.kind == .consumerGroup, partitionID == nil { + return try await pollGroup(streamID: streamID, topicID: topicID, consumer: consumer, count: count, autoCommit: autoCommit, strategyFor: strategyFor) + } + let strategy = strategyFor(partitionID ?? 0) + let payload = try Requests.pollMessages( + consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, strategy: strategy, count: count, autoCommit: autoCommit) + return try Responses.polledMessages(try await routedPoll(payload, context: strategy.context)[...]) + } + + // MARK: Partition contexts + + /// Sends under the partition's cached context, or one read from its + /// route. A history refusal means the context is stale: the topic's + /// contexts are dropped and one fresh route is tried. + private func sendWithPartitionContext(streamID: Identifier, topicID: Identifier, partitionID: UInt32, payload: [UInt8]) async throws -> [UInt8] { + let key = ConsumerGroupState.TopicKey(stream: streamID, topic: topicID) + let state = core.groupState + var context = state.partitionContext(key, partitionID: partitionID) + var refreshAvailable = true + while true { + let captured: PartitionContext + if let context { + captured = context + } else { + let body = try await send(.getPartitionRoute, Requests.partitionRoute(streamID: streamID, topicID: topicID, partitionID: partitionID)) + captured = try Responses.partitionRoute(body[...]) + state.setPartitionContext(key, partitionID: partitionID, captured) + } + do { + return try await core.exchange(code: CommandCode.sendMessages.rawValue, payload: payload, context: captured) + } catch let error as IggyError where error.code == .historyUnavailable { + state.invalidateTopic(key) + guard refreshAvailable else { + throw error + } + refreshAvailable = false + context = nil + } + } + } + + /// A poll on its route. The route key is the poll without its strategy, + /// count, and auto-commit flag, which do not change where it goes. + func routedPoll(_ payload: [UInt8], context: PartitionContext?) async throws -> [UInt8] { + let key = PollRouter.RouteKey(code: .getPollRouting, prefix: payload.dropLast(RawRequest.pollParametersSize)) + return try await core.pollRouter.send(.pollMessagesOnPrimary, payload: payload, key: key, routePayload: payload, context: context, coordinator: core) + } + + /// An offset write on its route. `routePayload` is the consumer, stream, + /// topic, and partition the write names. + func routedOffsetWrite(_ code: CommandCode, payload: [UInt8], routePayload: [UInt8], context: PartitionContext?) async throws { + let key = PollRouter.RouteKey(code: .getConsumerOffsetRouting, prefix: routePayload[...]) + _ = try await core.pollRouter.send(code, payload: payload, key: key, routePayload: routePayload, context: context, coordinator: core) + } + + // MARK: Client-side partitioning + + private func resolvePartition(_ partitioning: Partitioning, streamID: Identifier, topicID: Identifier) async throws -> UInt32 { + switch partitioning { + case .partition(let partitionID): + return partitionID + case .balanced: + let count = try await partitionCount(streamID: streamID, topicID: topicID) + return core.groupState.nextBalancedPartition(.init(stream: streamID, topic: topicID), partitionCount: count) + case .messagesKey(let key): + return XXH32.hash(key) % (try await partitionCount(streamID: streamID, topicID: topicID)) + } + } + + /// The topic's partition count, cached by the topic read so a send does + /// not pay a metadata round trip per batch. + private func partitionCount(streamID: Identifier, topicID: Identifier) async throws -> UInt32 { + var count = core.groupState.partitionCount(.init(stream: streamID, topic: topicID)) + if count == nil { + count = try await getTopic(streamID: streamID, topicID: topicID)?.partitionsCount + } + guard let count, count > 0 else { + throw IggyError(.topicIdNotFound, context: "topic \(topicID) in stream \(streamID)") + } + return count + } + + // MARK: Consumer-group polling + + /// Syncs the member's assignment from the coordinator into the cache. An + /// empty reply means the client is not a member. + func syncGroupAssignment(streamID: Identifier, topicID: Identifier, groupID: Identifier) async throws { + let body = try await send(.syncConsumerGroup, Requests.consumerGroup(streamID: streamID, topicID: topicID, groupID: groupID)) + let key = ConsumerGroupState.GroupKey(stream: streamID, topic: topicID, group: groupID) + let state = core.groupState + if body.isEmpty { + // The coordinator answers any member, even one holding no + // partitions, with an assignment, so this observes a removal: a + // deleted group or an evicted member. Without the deregister every + // later poll would come back empty instead of telling the caller. + state.invalidateAssignment(key) + state.deregisterGroup(key) + return + } + let assignment = try Responses.consumerGroupAssignment(body[...]) + state.registerGroup(key) Review Comment: Fixed in d5c1dbbfa. `ConsumerGroupState` keeps a session scope that `clearSessionScoped` advances. `syncGroupAssignment` reads it before the send, and `applySync` drops a reply whose scope moved, under the same lock as the write. Test: `aSyncAnsweredUnderAnEndedSessionIsDropped`. ########## foreign/swift/Sources/Iggy/Client/ConsumerGroupState.swift: ########## @@ -0,0 +1,205 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import NIOConcurrencyHelpers + +/// Client-side partitioning state, a port of `ConsumerGroupClientState` in +/// `core/common`. +/// +/// The server routes explicit partitions only, so the client resolves the +/// rest before it encodes a request: a consumer-group poll takes the next of +/// the member's assigned partitions, round-robin, from the assignment synced +/// from the coordinator; a `balanced` send round-robins per topic, and a +/// messages key is hashed. The cursors outlive single calls, so this lives on +/// the client. The coordinator fences a stale selection after a rebalance, +/// which makes the client sync again and restart the cursor. +final class ConsumerGroupState: Sendable { + struct TopicKey: Hashable, Sendable { + let stream: Identifier + let topic: Identifier + } + + struct GroupKey: Hashable, Sendable { + let stream: Identifier + let topic: Identifier + let group: Identifier + } + + private struct Assignment { + var partitions: [UInt32] = [] + var generation: UInt64 = 0 + var cursor = 0 + } + + private struct TopicPartitions { + var count: UInt32? + var contexts: [UInt32: PartitionContext] = [:] + var cursor = 0 + } + + private struct Storage { + var assignments: [GroupKey: Assignment] = [:] + var topics: [TopicKey: TopicPartitions] = [:] + /// The groups the last sync found this client a member of, so the + /// heartbeat can sync them again. + var joinedGroups: Set<GroupKey> = [] + } + + private let storage = NIOLockedValueBox(Storage()) + + // MARK: Group assignments + + /// Whether a non-empty assignment is cached for the group. + func hasAssignment(_ key: GroupKey) -> Bool { + storage.withLockedValue { $0.assignments[key].map { !$0.partitions.isEmpty } ?? false } + } + + /// Replaces the group's assignment. A new generation, a rebalance, + /// restarts the round-robin. + func setAssignment(_ key: GroupKey, generation: UInt64, partitions: [UInt32]) { + storage.withLockedValue { storage in + var assignment = storage.assignments[key] ?? Assignment() + if assignment.generation != generation { + assignment.cursor = 0 + } + assignment.generation = generation + assignment.partitions = partitions + storage.assignments[key] = assignment + } + } + + /// Drops the group's assignment, so the next poll syncs it again. + func invalidateAssignment(_ key: GroupKey) { + _ = storage.withLockedValue { $0.assignments.removeValue(forKey: key) } + } + + /// The next assigned partition to poll, or nil when nothing is assigned. + func nextGroupPartition(_ key: GroupKey) -> UInt32? { + storage.withLockedValue { storage in + guard var assignment = storage.assignments[key], !assignment.partitions.isEmpty else { + return nil + } + let partition = assignment.partitions[assignment.cursor % assignment.partitions.count] + assignment.cursor &+= 1 Review Comment: Changed both cursors to `UInt` in d5c1dbbfa. ########## foreign/swift/Sources/Iggy/Models/IggyMessage.swift: ########## @@ -196,6 +196,12 @@ public struct PolledMessages: Sendable, Hashable { /// Partition id of an empty poll by a consumer-group member that currently /// holds no partitions: `NO_ASSIGNED_PARTITION` in `core/common`. public static let noAssignedPartition: UInt32 = UInt32.max - 1 + + /// Partition id of an empty consumer-group poll the coordinator fenced + /// because the member's assignment is stale: + /// `RESYNC_REQUIRED_PARTITION_SENTINEL` in `core/common`. The client + /// syncs the assignment and polls again, so callers never see it. + static let resyncRequiredPartition: UInt32 = UInt32.max } Review Comment: Added in d5c1dbbfa. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
