RustToMetal commented on code in PR #4553: URL: https://github.com/apache/iggy/pull/4553#discussion_r4279807188
########## foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift: ########## @@ -0,0 +1,399 @@ +// 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. + /// + /// 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, Review Comment: Documented in f2bb9bc4f: the minted id applies to that send only, `messages` is left as it was, and a caller that retries and wants the same ids on every attempt sets them first, for example with `UInt128Value.random()`. The Rust client takes `&mut [IggyMessage]` and writes the id back. I did not copy that: an `inout` batch would rule out array literals (`messages: ["placed"]`) at every call site, and a caller can already choose its ids up front. ########## foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift: ########## @@ -0,0 +1,399 @@ +// 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. + /// + /// 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: Array(payload.dropLast(RawRequest.pollParametersSize))) Review Comment: Fixed in f2bb9bc4f. `RouteKey.prefix` is now an `ArraySlice<UInt8>`, so the poll key is `payload.dropLast(...)` without a copy, the same as the Rust key built with `payload.slice(..)`. ########## foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift: ########## @@ -0,0 +1,399 @@ +// 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. + /// + /// 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: Array(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) + state.setAssignment(key, generation: assignment.generation, partitions: assignment.partitions) + } + + /// Syncs every joined group, so a member picks up a widened assignment + /// without first hitting an ownership fence. The heartbeat runs it; one + /// group that fails does not stop the others. + func refreshConsumerGroupAssignments() async { + for key in core.groupState.registeredGroups { + do { + try await syncGroupAssignment(streamID: key.stream, topicID: key.topic, groupID: key.group) + } catch { + logger.warning("Failed to refresh the consumer group assignment of \(key.stream)|\(key.topic)|\(key.group): \(error)") + } + } + } + + private func pollGroup( + streamID: Identifier, topicID: Identifier, consumer: Consumer, count: UInt32, autoCommit: Bool, strategyFor: @Sendable (UInt32) -> PollingStrategy + ) async throws -> PolledMessages { + let key = ConsumerGroupState.GroupKey(stream: streamID, topic: topicID, group: consumer.id) + let state = core.groupState + if !state.hasAssignment(key) { + try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id) + } + for _ in 0..<Self.groupPollMaxAttempts { + guard let partitionID = state.nextGroupPartition(key) else { + // A member can hold no partitions, while a client that is not + // a member must learn it, or it polls an empty assignment + // forever instead of joining again. + guard state.isRegistered(key) else { + throw IggyError(.consumerGroupMemberNotFound, context: "group \(consumer.id) in topic \(topicID) of stream \(streamID)") + } + var empty = PolledMessages.empty + empty.partitionID = PolledMessages.noAssignedPartition + return empty + } + // A retry after a fence can land on another partition, so the + // strategy is asked for each attempt. + let strategy = strategyFor(partitionID) + let payload = try Requests.pollMessages( + consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, strategy: strategy, count: count, autoCommit: autoCommit) + do { + let polled = try Responses.polledMessages(try await routedPoll(payload, context: strategy.context)[...]) + if polled.messages.isEmpty, polled.partitionID == PolledMessages.resyncRequiredPartition { + // The coordinator fences a stale assignment through the + // empty poll, for want of a typed error. + state.invalidateAssignment(key) + try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id) + continue + } + return polled + } catch let error as IggyError where error.code == .consumerGroupPartitionNotOwned { + state.invalidateAssignment(key) + try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id) + if strategy.context != nil { + // A continuation of the partition the member lost cannot + // move to another one. + throw error + } + } + } + // Every attempt met a rebalance. The assignment is synced again and + // an empty poll lets the caller poll again, where an error would make + // a normal rebalance look like a failure. + return .empty + } +} + +// MARK: - Consumer offsets + +extension IggyClient { + /// Stores the offset of a consumer on a partition. A consumer-group + /// consumer stores it for the group. + public func storeConsumerOffset( + consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?, offset: UInt64 + ) + async throws + { + try await requireAuthenticated() + try await routedOffsetWrite( + .storeConsumerOffset, + payload: Requests.storeConsumerOffset( + consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, offset: offset, ack: .quorum), + routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID), context: nil) + } + + /// Stores a checkpoint under the partition context of the poll that + /// delivered it. It is refused once the partition was recreated or, for a + /// consumer group, reassigned since. + public func storeConsumerPosition(consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, position: ConsumerPosition) async throws { + try await requireAuthenticated() + try await routedOffsetWrite( + .storeConsumerOffset, + payload: Requests.storeConsumerOffset( + consumer: consumer, streamID: streamID, topicID: topicID, partitionID: position.partitionID, offset: position.offset, ack: .quorum), + routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: position.partitionID), + context: position.context) + } + + /// The stored offset of a consumer on a partition, or nil when none is + /// stored. + public func getConsumerOffset( + consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32? + ) async throws + -> ConsumerOffsetInfo? + { + try await requireAuthenticated() + let body = try await send( + .getConsumerOffset, Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID)) + return body.isEmpty ? nil : try Responses.consumerOffset(body[...]) + } + + /// Deletes the stored offset of a consumer on a partition. + public func deleteConsumerOffset(consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?) async throws { + try await requireAuthenticated() + try await routedOffsetWrite( + .deleteConsumerOffset, + payload: Requests.deleteConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, ack: .quorum), + routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID), context: nil) + } +} + +// MARK: - Raw partition requests + +extension IggyClient { + /// Sends a raw partition request the way its typed call sends it, so it + /// carries the partition context the server requires. Nil for any other + /// code. + func sendRawPartitionRequest(code: UInt32, payload: [UInt8]) async throws -> [UInt8]? { + switch CommandCode(rawValue: code) { + case .sendMessages: + let target = try RawRequest.sendTarget(payload) + return try await sendWithPartitionContext(streamID: target.stream, topicID: target.topic, partitionID: target.partitionID, payload: payload) + case .pollMessages: + try RawRequest.validatePoll(payload) + return try await routedPoll(payload, context: nil) + case .storeConsumerOffset, .deleteConsumerOffset: + let prefix = try RawRequest.consumerTargetPrefix(payload) + let command: CommandCode = code == CommandCode.storeConsumerOffset.rawValue ? .storeConsumerOffset : .deleteConsumerOffset + let key = PollRouter.RouteKey(code: .getConsumerOffsetRouting, prefix: prefix) + return try await core.pollRouter.send(command, payload: payload, key: key, routePayload: prefix, context: nil, coordinator: core) + default: + return nil + } + } +} + +/// Reads the partition a raw request names. A payload that does not decode +/// fails with ``IggyErrorCode/invalidCommand`` before anything is sent. +private enum RawRequest { + /// Strategy kind and value, count, and the auto-commit flag. + static let pollParametersSize = 9 + 4 + 1 + + struct SendTarget { + let stream: Identifier + let topic: Identifier + let partitionID: UInt32 + } + + /// `[metadata_len:u32][stream][topic][partitioning]...`. Balanced and + /// key partitioning are resolved by the typed send from its cache and + /// cursor, so a raw send names no partition to route. + static func sendTarget(_ payload: [UInt8]) throws -> SendTarget { + let target = try decode(payload) { reader in + _ = try reader.readUInt32() Review Comment: Fixed in f2bb9bc4f. The header is now decoded from the `metadata_len` bytes only, the way `send_raw_messages` does it in Rust. The stream, topic, partitioning, and messages count must fit, and the partitioning lengths are checked as the server checks them (balanced 0, partition id 4, key non-empty). A malformed header is `invalidCommand` before any route is read, and balanced or keyed partitioning is still `featureUnavailable`. The raw-routing test covers a count outside the declared length and a 3-byte partition id. I broke each check on purpose and the test failed both times. ########## foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift: ########## @@ -0,0 +1,399 @@ +// 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. + /// + /// 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: Array(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) + state.setAssignment(key, generation: assignment.generation, partitions: assignment.partitions) + } + + /// Syncs every joined group, so a member picks up a widened assignment + /// without first hitting an ownership fence. The heartbeat runs it; one + /// group that fails does not stop the others. + func refreshConsumerGroupAssignments() async { + for key in core.groupState.registeredGroups { + do { + try await syncGroupAssignment(streamID: key.stream, topicID: key.topic, groupID: key.group) + } catch { + logger.warning("Failed to refresh the consumer group assignment of \(key.stream)|\(key.topic)|\(key.group): \(error)") + } + } + } + + private func pollGroup( + streamID: Identifier, topicID: Identifier, consumer: Consumer, count: UInt32, autoCommit: Bool, strategyFor: @Sendable (UInt32) -> PollingStrategy + ) async throws -> PolledMessages { + let key = ConsumerGroupState.GroupKey(stream: streamID, topic: topicID, group: consumer.id) + let state = core.groupState + if !state.hasAssignment(key) { + try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id) + } + for _ in 0..<Self.groupPollMaxAttempts { + guard let partitionID = state.nextGroupPartition(key) else { + // A member can hold no partitions, while a client that is not + // a member must learn it, or it polls an empty assignment + // forever instead of joining again. + guard state.isRegistered(key) else { + throw IggyError(.consumerGroupMemberNotFound, context: "group \(consumer.id) in topic \(topicID) of stream \(streamID)") + } + var empty = PolledMessages.empty + empty.partitionID = PolledMessages.noAssignedPartition + return empty + } + // A retry after a fence can land on another partition, so the + // strategy is asked for each attempt. + let strategy = strategyFor(partitionID) + let payload = try Requests.pollMessages( + consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, strategy: strategy, count: count, autoCommit: autoCommit) + do { + let polled = try Responses.polledMessages(try await routedPoll(payload, context: strategy.context)[...]) + if polled.messages.isEmpty, polled.partitionID == PolledMessages.resyncRequiredPartition { + // The coordinator fences a stale assignment through the + // empty poll, for want of a typed error. + state.invalidateAssignment(key) + try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id) + continue + } + return polled + } catch let error as IggyError where error.code == .consumerGroupPartitionNotOwned { + state.invalidateAssignment(key) + try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id) + if strategy.context != nil { + // A continuation of the partition the member lost cannot + // move to another one. + throw error + } + } + } + // Every attempt met a rebalance. The assignment is synced again and + // an empty poll lets the caller poll again, where an error would make + // a normal rebalance look like a failure. + return .empty + } +} + +// MARK: - Consumer offsets + +extension IggyClient { + /// Stores the offset of a consumer on a partition. A consumer-group + /// consumer stores it for the group. + public func storeConsumerOffset( + consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?, offset: UInt64 + ) + async throws + { + try await requireAuthenticated() + try await routedOffsetWrite( + .storeConsumerOffset, + payload: Requests.storeConsumerOffset( + consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, offset: offset, ack: .quorum), + routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID), context: nil) + } + + /// Stores a checkpoint under the partition context of the poll that + /// delivered it. It is refused once the partition was recreated or, for a + /// consumer group, reassigned since. + public func storeConsumerPosition(consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, position: ConsumerPosition) async throws { + try await requireAuthenticated() + try await routedOffsetWrite( + .storeConsumerOffset, + payload: Requests.storeConsumerOffset( + consumer: consumer, streamID: streamID, topicID: topicID, partitionID: position.partitionID, offset: position.offset, ack: .quorum), + routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: position.partitionID), + context: position.context) + } + + /// The stored offset of a consumer on a partition, or nil when none is + /// stored. + public func getConsumerOffset( + consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32? + ) async throws + -> ConsumerOffsetInfo? + { + try await requireAuthenticated() + let body = try await send( + .getConsumerOffset, Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID)) + return body.isEmpty ? nil : try Responses.consumerOffset(body[...]) + } + + /// Deletes the stored offset of a consumer on a partition. + public func deleteConsumerOffset(consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?) async throws { + try await requireAuthenticated() + try await routedOffsetWrite( + .deleteConsumerOffset, + payload: Requests.deleteConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, ack: .quorum), + routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID), context: nil) + } +} + +// MARK: - Raw partition requests + +extension IggyClient { + /// Sends a raw partition request the way its typed call sends it, so it + /// carries the partition context the server requires. Nil for any other + /// code. + func sendRawPartitionRequest(code: UInt32, payload: [UInt8]) async throws -> [UInt8]? { + switch CommandCode(rawValue: code) { + case .sendMessages: + let target = try RawRequest.sendTarget(payload) + return try await sendWithPartitionContext(streamID: target.stream, topicID: target.topic, partitionID: target.partitionID, payload: payload) + case .pollMessages: + try RawRequest.validatePoll(payload) + return try await routedPoll(payload, context: nil) + case .storeConsumerOffset, .deleteConsumerOffset: + let prefix = try RawRequest.consumerTargetPrefix(payload) + let command: CommandCode = code == CommandCode.storeConsumerOffset.rawValue ? .storeConsumerOffset : .deleteConsumerOffset + let key = PollRouter.RouteKey(code: .getConsumerOffsetRouting, prefix: prefix) + return try await core.pollRouter.send(command, payload: payload, key: key, routePayload: prefix, context: nil, coordinator: core) + default: + return nil + } + } +} + +/// Reads the partition a raw request names. A payload that does not decode +/// fails with ``IggyErrorCode/invalidCommand`` before anything is sent. +private enum RawRequest { + /// Strategy kind and value, count, and the auto-commit flag. + static let pollParametersSize = 9 + 4 + 1 + + struct SendTarget { + let stream: Identifier + let topic: Identifier + let partitionID: UInt32 + } + + /// `[metadata_len:u32][stream][topic][partitioning]...`. Balanced and + /// key partitioning are resolved by the typed send from its cache and + /// cursor, so a raw send names no partition to route. + static func sendTarget(_ payload: [UInt8]) throws -> SendTarget { + let target = try decode(payload) { reader in + _ = try reader.readUInt32() + let stream = try Identifier.decode(from: &reader) + let topic = try Identifier.decode(from: &reader) + let kind = try reader.readUInt8() + let value = try reader.readBytes(Int(try reader.readUInt8())) + return (stream, topic, kind, value) + } + guard target.2 == 2, target.3.count == 4 else { Review Comment: Fixed in f2bb9bc4f. `Partitioning.decode(from:)` sits next to the encoder and uses the same named kind codes and the new `partitionIDSize`, which the encoder and `encodedSize` also use now. `sendTarget` matches on `.partition(let id)` instead of the literals. `PartitioningTests` round-trips all three kinds through it and rejects malformed ones. -- 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]
