RustToMetal commented on code in PR #4553:
URL: https://github.com/apache/iggy/pull/4553#discussion_r4286269916
##########
foreign/swift/Sources/Iggy/Models/Messaging.swift:
##########
@@ -144,10 +161,15 @@ public struct PollingStrategy: Sendable, Hashable,
CustomStringConvertible {
public var kind: Kind
public var value: UInt64
+ /// The partition incarnation and owner a continuation offset was read
+ /// under. A poll that carries one is refused once the partition was
+ /// recreated, instead of reading the offset in another incarnation.
+ public var context: PartitionContext?
Review Comment:
Added in d5c1dbbfa: `.offset(_:context:)` and `.timestamp(_:context:)`, with
`context` defaulting to nil. The README and the unit and e2e tests use them now.
##########
foreign/swift/README.md:
##########
@@ -100,6 +100,48 @@ when it has credentials to sign in with. In a cluster the
client reads the roste
signing in, moves to the leader, and moves on when a node answers that it is
not the
primary.
+### Messages
+
+`sendMessages` writes a batch to one partition. The server routes explicit
partitions
+only, so `balanced` (the default) and `messagesKey` are resolved on the client:
+balanced round-robins over the topic's partitions, and a key is hashed with
XXH32 the
+way every other SDK hashes it. `pollMessages` reads from a partition, starting
where the
+strategy says:
+
+```swift
+try await client.sendMessages(streamID: "orders", topicID: "created",
partitioning: .messagesKey("customer-42"), messages: ["placed"])
+let polled = try await client.pollMessages(
+ streamID: "orders", topicID: "created", partitionID: 0, consumer:
.consumer(1), strategy: .next, count: 100, autoCommit: true)
+for message in polled.messages {
+ print(message.offset, message.payloadString)
+}
+```
+
+Every partition request carries the partition context the server requires: the
+partition's incarnation, which changes when a deleted partition id is reused.
The
+client reads it from the partition's route and caches it. A poll reply carries
the
+context it was served under; hand it back in `PollingStrategy.context` to
continue the
+same partition, or in a `ConsumerPosition` to store a checkpoint, and the
server refuses
+either with `historyUnavailable` once the partition was recreated.
+
+### Consumer groups
+
+A member polls without a partition id and the client takes the next partition
of its
+assignment, round-robin, syncing the assignment from the server when it
changes. The
+heartbeat syncs every joined group, so a member picks up new partitions on its
own:
+
+```swift
+try await client.joinConsumerGroup(streamID: "orders", topicID: "created",
groupID: "billing")
+let batch = try await client.pollMessages(
+ streamID: "orders", topicID: "created", partitionID: nil, consumer:
.group("billing"), strategy: .next, count: 100, autoCommit: true)
+```
+
+A member holding no partitions gets an empty poll whose partition id is
+`PolledMessages.noAssignedPartition`, and a client that is not a member gets
+`consumerGroupMemberNotFound`. Membership belongs to the session, so after a
reconnect
Review Comment:
Reworded in d5c1dbbfa. With the session resumed, membership now survives a
reconnect, so the README says that, and that when the server no longer holds
the session the caller must call `joinConsumerGroup` again, with the next group
poll failing with `consumerGroupMemberNotFound` until it does.
##########
foreign/swift/README.md:
##########
@@ -100,6 +100,48 @@ when it has credentials to sign in with. In a cluster the
client reads the roste
signing in, moves to the leader, and moves on when a node answers that it is
not the
primary.
+### Messages
+
+`sendMessages` writes a batch to one partition. The server routes explicit
partitions
+only, so `balanced` (the default) and `messagesKey` are resolved on the client:
+balanced round-robins over the topic's partitions, and a key is hashed with
XXH32 the
+way every other SDK hashes it. `pollMessages` reads from a partition, starting
where the
+strategy says:
+
+```swift
+try await client.sendMessages(streamID: "orders", topicID: "created",
partitioning: .messagesKey("customer-42"), messages: ["placed"])
+let polled = try await client.pollMessages(
+ streamID: "orders", topicID: "created", partitionID: 0, consumer:
.consumer(1), strategy: .next, count: 100, autoCommit: true)
+for message in polled.messages {
+ print(message.offset, message.payloadString)
+}
+```
+
+Every partition request carries the partition context the server requires: the
+partition's incarnation, which changes when a deleted partition id is reused.
The
+client reads it from the partition's route and caches it. A poll reply carries
the
+context it was served under; hand it back in `PollingStrategy.context` to
continue the
+same partition, or in a `ConsumerPosition` to store a checkpoint, and the
server refuses
+either with `historyUnavailable` once the partition was recreated.
Review Comment:
Done in d5c1dbbfa: the paragraph names `historyUnavailable` for a recreated
partition and `consumerGroupPartitionNotOwned` for a group checkpoint after the
partition moved to another member.
##########
foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift:
##########
@@ -0,0 +1,691 @@
+// 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 Foundation
+import Iggy
+import Testing
+
+@Suite("Sessions", requiresServer)
+struct SessionE2ETests {
+ @Test func onlyPingWorksBeforeSigningIn() async throws {
+ try await E2EEnvironment.withSession(login: false) { session in
+ let client = session.client
+ #expect(await client.state == .connected)
+ try await client.ping()
+ await #expect(throws: IggyError(.unauthenticated)) {
+ try await client.getStreams()
+ }
+ // The roster is auth-gated so an unauthenticated reader cannot map
+ // the private network.
+ await #expect(throws: IggyError(.unauthenticated)) {
+ try await client.getClusterMetadata()
+ }
+ }
+ }
+
+ @Test func loginLogoutAndRelogin() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ #expect(await client.state == .authenticated)
+ let me = try await client.getMe()
+ #expect(me.userID == 0)
+ #expect(me.transport == "TCP")
+ _ = try await client.getStreams()
+
+ try await client.logout()
+ #expect(await client.state == .connected)
+ await #expect(throws: IggyError(.unauthenticated)) {
+ try await client.getStreams()
+ }
+
+ let identity = try await client.login(username:
E2EEnvironment.rootUsername, password: E2EEnvironment.rootPassword)
+ #expect(identity.userID == 0)
+ _ = try await client.getStreams()
+ }
+ }
+
+ @Test func autoLoginSignsInOnConnect() async throws {
+ try await E2EEnvironment.withSession(autoLogin:
.usernamePassword(username: E2EEnvironment.rootUsername, password:
E2EEnvironment.rootPassword)) {
+ session in
+ let client = session.client
+ #expect(await client.state == .authenticated)
+ _ = try await client.getStreams()
+ }
+ }
+
+ @Test func invalidCredentialsAreRejected() async throws {
+ try await E2EEnvironment.withSession(login: false) { session in
+ let client = session.client
+ await #expect(throws: IggyError.self) {
+ try await client.login(username: "nobody", password:
"wrong-password")
+ }
+ #expect(await client.state == .connected)
+ try await client.ping()
+ }
+ }
+
+ @Test func personalAccessTokenRoundTrip() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-pat")
+ let token = try await client.createPersonalAccessToken(name: name,
expiry: .after(.seconds(3600)))
+ #expect(!token.token.isEmpty)
+ let listed = try await client.getPersonalAccessTokens()
+ let created = try #require(listed.first { $0.name == name })
+ #expect(created.expiresAt != nil)
+
+ let other = try await session.secondaryClient()
+ let identity = try await other.login(personalAccessToken:
token.token)
+ #expect(identity.userID == 0)
+ _ = try await other.getStreams()
+
+ try await client.deletePersonalAccessToken(name: name)
+ #expect(try await client.getPersonalAccessTokens().contains {
$0.name == name } == false)
+ }
+ }
+
+ @Test func eventsReportTheLifecycle() async throws {
+ let client = IggyClient(
+ configuration: E2EEnvironment.configuration(
+ autoLogin: .usernamePassword(username:
E2EEnvironment.rootUsername, password: E2EEnvironment.rootPassword)))
+ let events = await client.events
+ try await client.connect()
+ try await client.logout()
+ await client.disconnect()
+ await client.shutdown()
+ var received: [DiagnosticEvent] = []
+ for await event in events {
+ received.append(event)
+ }
+ #expect(received == [.connected, .signedIn, .signedOut, .disconnected,
.shutdown])
+ }
+
+ @Test func shutdownIsTerminal() async throws {
+ let client = try await E2EEnvironment.connect()
+ await client.shutdown()
+ #expect(await client.state == .shutdown)
+ await #expect(throws: IggyError(.clientShutdown)) {
+ try await client.connect()
+ }
+ await #expect(throws: IggyError(.clientShutdown)) {
+ try await client.ping()
+ }
+ }
+
+ @Test func disconnectKeepsTheClientUsable() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ await client.disconnect()
+ #expect(await client.state == .disconnected)
+ // No credentials survive an explicit disconnect, so nothing
replays.
+ await #expect(throws: IggyError(.notConnected)) {
+ try await client.ping()
+ }
+ try await client.connect()
+ try await client.ping()
+ _ = try await client.login(username: E2EEnvironment.rootUsername,
password: E2EEnvironment.rootPassword)
+ _ = try await client.getStreams()
+ }
+ }
+}
+
+@Suite("Streams, topics, and partitions", requiresServer)
+struct ResourceE2ETests {
+ @Test func streamCRUD() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-stream")
+ let created = try await client.createStream(name: name)
+ #expect(created.name == name)
+ #expect(created.topicsCount == 0)
+ let id = Identifier(numeric: created.id)
+
+ let byID = try #require(try await client.getStream(id))
+ #expect(byID.name == name)
+ let byName = try #require(try await
client.getStream(Identifier(named: name)))
+ #expect(byName.id == created.id)
+ #expect(try await client.getStreams().contains { $0.id ==
created.id })
+
+ let renamed = E2EEnvironment.uniqueName("swift-renamed")
+ try await client.updateStream(id, name: renamed)
+ #expect(try await client.getStream(id)?.name == renamed)
+
+ try await client.deleteStream(id)
+ // The server hands out the lowest free id, so a parallel test can
+ // already own it again: assert on the name, which is unique.
+ #expect(try await client.getStream(Identifier(named: renamed)) ==
nil)
+ await #expect(throws: IggyError.self) {
+ try await client.deleteStream(Identifier(named: renamed))
+ }
+ }
+ }
+
+ @Test func duplicateStreamNamesAreRejected() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-dup")
+ _ = try await client.createStream(name: name)
+ await session.track(stream: try Identifier(named: name))
+ do {
+ _ = try await client.createStream(name: name)
+ Issue.record("expected the duplicate to be rejected")
+ } catch let error as IggyError {
+ #expect(error.code == .streamNameAlreadyExists)
+ }
+ }
+ }
+
+ @Test func topicCRUDWithOptionsAndCatalog() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-topic")
+ let stream = try await client.createStream(name: name)
+ let streamID = Identifier(numeric: stream.id)
+ await session.track(stream: try Identifier(named: name))
+
+ let specs = try await client.describeOptions(scope: .topic)
+ let keys = Set(specs.map(\.key))
+ #expect(keys.contains(TopicOptionKey.durability))
+ #expect(keys.contains(TopicOptionKey.segmentSize))
+ #expect(try #require(specs.first { $0.key ==
TopicOptionKey.segmentSize }).kind == .uint64)
+ #expect(try await client.describeOptions(scope: .stream).isEmpty)
+
+ let options = TopicCreateOptions(
+ partitionsCount: 3, compressionAlgorithm: .gzip,
messageExpiry: .after(.seconds(3600)),
+ maxTopicSize: .bytes(10_737_418_240), durability: .persisted)
+ let created = try await client.createTopic(streamID: streamID,
name: name, options: options)
+ #expect(created.name == name)
+ #expect(created.partitionsCount == 3)
+ #expect(created.partitions.map(\.id) == [0, 1, 2])
+ #expect(created.compressionAlgorithm == .gzip)
+ #expect(created.messageExpiry == .after(.seconds(3600)))
+ #expect(created.maxTopicSize == .bytes(10_737_418_240))
+ let topicID = Identifier(numeric: created.id)
+
+ let fetched = try #require(try await client.getTopic(streamID:
streamID, topicID: topicID))
+ #expect(fetched.options[TopicOptionKey.durability] ==
.explicit("persisted"))
+ // Keys the client left alone are resolved by admission and
reported
+ // as derived.
+ #expect(fetched.options[TopicOptionKey.segmentSize]?.explicit ==
false)
+ #expect(try await client.getTopics(streamID: streamID).map(\.id)
== [created.id])
+ #expect(try await client.getStream(streamID)?.topics.first?.name
== name)
+
+ let renamed = E2EEnvironment.uniqueName("swift-topic-renamed")
+ try await client.updateTopic(
+ streamID: streamID, topicID: topicID, name: renamed,
+ options: TopicUpdateOptions(compressionAlgorithm:
CompressionAlgorithm.none, messageExpiry: .never))
+ let updated = try #require(try await client.getTopic(streamID:
streamID, topicID: topicID))
+ #expect(updated.name == renamed)
+ #expect(updated.compressionAlgorithm == CompressionAlgorithm.none)
+ #expect(updated.messageExpiry == .never)
+
+ try await client.createPartitions(streamID: streamID, topicID:
topicID, count: 2)
+ #expect(try await client.getTopic(streamID: streamID, topicID:
topicID)?.partitionsCount == 5)
+ try await client.deletePartitions(streamID: streamID, topicID:
topicID, count: 4)
+ #expect(try await client.getTopic(streamID: streamID, topicID:
topicID)?.partitionsCount == 1)
+
+ try await client.deleteTopic(streamID: streamID, topicID: topicID)
+ #expect(try await client.getTopic(streamID: streamID, topicID:
topicID) == nil)
+ }
+ }
+
+ @Test func unknownOptionKeysAreRefused() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-badopt")
+ let stream = try await client.createStream(name: name)
+ let streamID = Identifier(numeric: stream.id)
+ await session.track(stream: try Identifier(named: name))
+ await #expect(throws: IggyError.self) {
+ try await client.createTopic(streamID: streamID, name: name,
options: TopicCreateOptions(raw: ["not_a_real_option": "1"]))
+ }
+ // A string value for a typed key is parsed by admission and
reported
+ // with its canonical kind.
+ let created = try await client.createTopic(
+ streamID: streamID, name: name, options:
TopicCreateOptions(raw: [TopicOptionKey.durability: "persisted"]))
+ let topic = try #require(try await client.getTopic(streamID:
streamID, topicID: Identifier(numeric: created.id)))
+ #expect(topic.options[TopicOptionKey.durability] ==
.explicit("persisted"))
+ }
+ }
+
+ @Test func consumerGroupCRUD() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 2)
+ let name = E2EEnvironment.uniqueName("swift-group")
+ let group = try await client.createConsumerGroup(streamID:
scratch.stream, topicID: scratch.topic, name: name)
+ #expect(group.name == name)
+ #expect(group.partitionsCount == 2)
+ #expect(group.members.isEmpty)
+ let groupID = Identifier(numeric: group.id)
+ #expect(try await client.getConsumerGroups(streamID:
scratch.stream, topicID: scratch.topic).map(\.id) == [group.id])
+ #expect(try await client.getConsumerGroup(streamID:
scratch.stream, topicID: scratch.topic, groupID: Identifier(named: name))?.id
== group.id)
+
+ try await client.joinConsumerGroup(streamID: scratch.stream,
topicID: scratch.topic, groupID: groupID)
+ // The server installs the member's partitions after the join.
+ let owned = try await E2EEnvironment.eventually {
+ try await client.getConsumerGroup(streamID: scratch.stream,
topicID: scratch.topic, groupID: groupID)?.members.first?.partitions.sorted()
+ == [0, 1]
+ }
+ #expect(owned)
+ let joined = try #require(try await
client.getConsumerGroup(streamID: scratch.stream, topicID: scratch.topic,
groupID: groupID))
+ #expect(joined.membersCount == 1)
+ let me = try await client.getMe()
+ #expect(me.consumerGroups.contains { $0.groupID == group.id })
+
+ try await client.leaveConsumerGroup(streamID: scratch.stream,
topicID: scratch.topic, groupID: groupID)
+ #expect(try await client.getConsumerGroup(streamID:
scratch.stream, topicID: scratch.topic, groupID: groupID)?.membersCount == 0)
+ try await client.deleteConsumerGroup(streamID: scratch.stream,
topicID: scratch.topic, groupID: groupID)
+ #expect(try await client.getConsumerGroup(streamID:
scratch.stream, topicID: scratch.topic, groupID: groupID) == nil)
+ }
+ }
+}
+
+@Suite("Users", requiresServer)
+struct UserE2ETests {
+ @Test func userLifecycle() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let username =
String(E2EEnvironment.uniqueName("swift-user").prefix(50))
+ let permissions = Permissions(
+ global: GlobalPermissions(readServers: true, readStreams:
true, readTopics: true, pollMessages: true, sendMessages: true))
+ let created = try await client.createUser(username: username,
password: "secret-1", status: .active, permissions: permissions)
+ #expect(created.username == username)
+ #expect(created.status == .active)
+ #expect(created.permissions?.global.readStreams == true)
+ #expect(created.permissions?.global.manageStreams == false)
+ let userID = Identifier(numeric: created.id)
+ #expect(try await client.getUsers().contains { $0.id == created.id
})
+ #expect(try await client.getUser(Identifier(named: username))?.id
== created.id)
+
+ // The new user can sign in and is bound by its permissions.
+ let other = try await session.secondaryClient()
+ let identity = try await other.login(username: username, password:
"secret-1")
+ #expect(identity.userID == created.id)
+ _ = try await other.getStreams()
+ await #expect(throws: IggyError(.unauthorized)) {
+ try await other.createStream(name:
E2EEnvironment.uniqueName("swift-forbidden"))
+ }
+
+ try await client.changePassword(userID, currentPassword:
"secret-1", newPassword: "secret-2")
+ try await client.updateUser(userID, username: username +
"-renamed", status: .inactive)
+ let updated = try #require(try await client.getUser(userID))
+ #expect(updated.username == username + "-renamed")
+ #expect(updated.status == .inactive)
+ try await client.updatePermissions(userID, permissions: nil)
+ #expect(try await client.getUser(userID)?.permissions == nil)
+ try await client.deleteUser(userID)
+ #expect(try await client.getUser(userID) == nil)
+ }
+ }
+
+ @Test func invalidUsernamesFailBeforeTheWire() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ await #expect(throws: IggyError(.invalidUsername)) {
+ try await client.createUser(username: "ab", password: "secret")
+ }
+ await #expect(throws: IggyError(.invalidPassword)) {
+ try await client.createUser(username: "valid-name", password:
"ab")
+ }
+ }
+ }
+}
+
+@Suite("Messaging", requiresServer)
+struct MessagingE2ETests {
+ @Test func sendAndPollRoundTrip() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 2)
+
+ var baseOffsets: [UInt64] = []
+ for _ in 0..<3 {
+ let response = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .partition(0), messages: try E2EEnvironment.messages(4))
+ let confirmation = try #require(response.confirmations.first)
+ #expect(confirmation.streamID == scratch.streamID)
+ #expect(confirmation.topicID == scratch.topicID)
+ #expect(confirmation.partitionID == 0)
+ baseOffsets.append(confirmation.baseOffset)
+ }
+ #expect(baseOffsets == [0, 4, 8])
+
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
0, consumer: .default, strategy: .offset(0), count: 12, autoCommit: false)
+ #expect(polled.partitionID == 0)
+ #expect(polled.currentOffset == 11)
+ #expect(polled.messages.count == 12)
+ #expect(polled.messages.map(\.offset) == Array(0..<12))
+ #expect(polled.messages[0].payloadString == "message 0")
+ #expect(polled.messages[11].payloadString == "message 3")
+ #expect(polled.messages.allSatisfy { !$0.id.isZero })
+ #expect(polled.messages.allSatisfy { $0.timestamp.microseconds > 0
&& $0.originTimestamp.microseconds > 0 })
+ #expect(polled.messages.allSatisfy { $0.checksum != 0 })
+ let ids = Set(polled.messages.map(\.id))
+ #expect(ids.count == 12)
+
+ let last = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
0, consumer: .default, strategy: .last, count: 1, autoCommit: false)
+ #expect(last.messages.map(\.offset) == [11])
+ let tail = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
0, consumer: .default, strategy: .offset(10), count: 100, autoCommit: false)
+ #expect(tail.messages.map(\.offset) == [10, 11])
+ let empty = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
1, consumer: .default, strategy: .first, count: 10, autoCommit: false)
+ #expect(empty.messages.isEmpty)
+ }
+ }
+
+ @Test func userHeadersAndIdsRoundTrip() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+
+ let id = MessageID(uuid: UUID())
+ let headers: UserHeaders = ["trace-id": "abc-123", "attempt":
.uint32(3), "urgent": true, "ratio": 0.5, "raw": try .raw([1, 2, 3])]
+ let message = try IggyMessage("with headers", id: id, userHeaders:
headers)
+ let binary = try IggyMessage(payload: [0x00, 0xFF, 0x10, 0x80])
+ _ = try await client.sendMessages(streamID: scratch.stream,
topicID: scratch.topic, partitioning: .partition(0), messages: [message,
binary])
+
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
0, consumer: .default, strategy: .first, count: 10, autoCommit: false)
+ #expect(polled.messages.count == 2)
+ let first = polled.messages[0]
+ #expect(first.id == id)
+ #expect(first.payloadString == "with headers")
+ #expect(first.userHeaders == headers)
+ #expect(first[header: "trace-id"]?.stringValue == "abc-123")
+ #expect(first[header: "attempt"]?.uint32Value == 3)
+ #expect(polled.messages[1].payload == [0x00, 0xFF, 0x10, 0x80])
+ #expect(polled.messages[1].userHeaders == nil)
+ }
+ }
+
+ @Test func balancedAndKeyedPartitioning() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 3)
+
+ var balancedPartitions: [UInt32] = []
+ for _ in 0..<3 {
+ let response = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .balanced, messages: try E2EEnvironment.messages(1))
+ balancedPartitions.append(try
#require(response.confirmations.first).partitionID)
+ }
+ #expect(Set(balancedPartitions) == [0, 1, 2])
+
+ let first = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .messagesKey("order-key-1"), messages: try
E2EEnvironment.messages(1))
+ let second = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .messagesKey("order-key-1"), messages: try
E2EEnvironment.messages(1))
+ #expect(try #require(first.confirmations.first).partitionID ==
(try #require(second.confirmations.first).partitionID))
+ }
+ }
+
+ @Test func partitionCountRefreshesAfterPartitionChanges() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+
+ let first = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .balanced, messages: try E2EEnvironment.messages(1))
+ #expect(try #require(first.confirmations.first).partitionID == 0)
+ try await client.createPartitions(streamID: scratch.stream,
topicID: scratch.topic, count: 2)
+ let second = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .balanced, messages: try E2EEnvironment.messages(1))
+ #expect(try #require(second.confirmations.first).partitionID == 1)
+ try await client.deletePartitions(streamID: scratch.stream,
topicID: scratch.topic, count: 2)
+ let third = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .balanced, messages: try E2EEnvironment.messages(1))
+ #expect(try #require(third.confirmations.first).partitionID == 0)
+ }
+ }
+
+ @Test func consumerOffsets() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+ _ = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .partition(0), messages: try E2EEnvironment.messages(5))
+
+ let consumer = Consumer.consumer("swift-offsets")
+ #expect(try await client.getConsumerOffset(consumer: consumer,
streamID: scratch.stream, topicID: scratch.topic, partitionID: 0) == nil)
+ try await client.storeConsumerOffset(consumer: consumer, streamID:
scratch.stream, topicID: scratch.topic, partitionID: 0, offset: 3)
+ let stored = try #require(try await
client.getConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID:
scratch.topic, partitionID: 0))
+ #expect(stored.storedOffset == 3)
+ #expect(stored.currentOffset == 4)
+ #expect(stored.partitionID == 0)
+
+ let next = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
0, consumer: consumer, strategy: .next, count: 10, autoCommit: true)
+ #expect(next.messages.map(\.offset) == [4])
+ #expect(try await client.getConsumerOffset(consumer: consumer,
streamID: scratch.stream, topicID: scratch.topic, partitionID: 0)?.storedOffset
== 4)
+
+ try await client.deleteConsumerOffset(consumer: consumer,
streamID: scratch.stream, topicID: scratch.topic, partitionID: 0)
+ #expect(try await client.getConsumerOffset(consumer: consumer,
streamID: scratch.stream, topicID: scratch.topic, partitionID: 0) == nil)
+ await #expect(throws: IggyError(.consumerOffsetNotFound)) {
+ try await client.deleteConsumerOffset(consumer: consumer,
streamID: scratch.stream, topicID: scratch.topic, partitionID: 0)
+ }
+ }
+ }
+
+ @Test func checkpointsKeepThePartitionContextOfTheirPoll() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 2)
+ let consumer = Consumer.consumer("swift-positions")
+ _ = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .partition(1), messages: try E2EEnvironment.messages(3))
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
1, consumer: consumer, strategy: .first, count: 3, autoCommit: false)
+ #expect(polled.messages.map(\.offset) == [0, 1, 2])
+ let position = ConsumerPosition(partitionID: 1, offset: 2,
context: polled.context)
+ try await client.storeConsumerPosition(consumer: consumer,
streamID: scratch.stream, topicID: scratch.topic, position: position)
+ #expect(try await client.getConsumerOffset(consumer: consumer,
streamID: scratch.stream, topicID: scratch.topic, partitionID: 1)?.storedOffset
== 2)
+ let continued = PollingStrategy(kind: .offset, value: 1, context:
polled.context)
+ let more = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
1, consumer: consumer, strategy: continued, count: 10, autoCommit: false)
+ #expect(more.messages.map(\.offset) == [1, 2])
+
+ // Another client recreates the partition, so its id now names a
+ // new incarnation this client has not seen.
+ let other = try await session.secondaryClient()
+ _ = try await other.login(username: E2EEnvironment.rootUsername,
password: E2EEnvironment.rootPassword)
+ try await other.deletePartitions(streamID: scratch.stream,
topicID: scratch.topic, count: 1)
+ try await other.createPartitions(streamID: scratch.stream,
topicID: scratch.topic, count: 1)
+
+ // The send under the stale cached context reads the new route.
+ _ = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .partition(1), messages: try E2EEnvironment.messages(1))
+ // Neither a checkpoint nor a continuation of the old incarnation
+ // lands on the new one.
+ await expectIggyError(.historyUnavailable) {
+ try await client.storeConsumerPosition(consumer: consumer,
streamID: scratch.stream, topicID: scratch.topic, position: position)
+ }
+ await expectIggyError(.historyUnavailable) {
+ _ = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitionID: 1, consumer: consumer, strategy: continued, count: 10, autoCommit:
false)
+ }
+ let fresh = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
1, consumer: consumer, strategy: .first, count: 10, autoCommit: false)
+ #expect(fresh.messages.count == 1)
+ #expect(fresh.context.incarnation != polled.context.incarnation)
+ }
+ }
+
+ @Test func rawPartitionRequestsCarryTheirContext() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+ _ = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .partition(0), messages: try E2EEnvironment.messages(2))
+ let target = Self.rawConsumerTarget(consumer: 7, stream:
scratch.streamID, topic: scratch.topicID)
+ // Strategy first, count 10, no auto-commit.
+ let poll = target + [3] + Self.littleEndian(UInt64(0)) +
Self.littleEndian(UInt32(10)) + [0]
+ #expect(!(try await client.sendRawRequest(code:
CommandCode.pollMessages.rawValue, payload: poll)).isEmpty)
+ // Offset 1, quorum acknowledgement.
+ let store = target + Self.littleEndian(UInt64(1)) + [1]
+ _ = try await client.sendRawRequest(code:
CommandCode.storeConsumerOffset.rawValue, payload: store)
+ let stored = try await client.getConsumerOffset(consumer:
.consumer(7), streamID: scratch.stream, topicID: scratch.topic, partitionID: 0)
+ #expect(stored?.storedOffset == 1)
+ }
+ }
+
+ /// `[consumer kind:1][consumer][stream][topic][has partition:1][partition
0:4]`
+ /// with numeric identifiers `[kind 1][length 4][id]`.
+ private static func rawConsumerTarget(consumer: UInt32, stream: UInt32,
topic: UInt32) -> [UInt8] {
+ let numeric = { (id: UInt32) in [1, 4] + littleEndian(id) }
+ return [1] + numeric(consumer) + numeric(stream) + numeric(topic) +
[1] + littleEndian(UInt32(0))
+ }
+
+ private static func littleEndian<T: FixedWidthInteger>(_ value: T) ->
[UInt8] {
+ withUnsafeBytes(of: value.littleEndian) { Array($0) }
+ }
+
+ @Test func consumerGroupPolling() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 3)
+ let group = try await client.createConsumerGroup(streamID:
scratch.stream, topicID: scratch.topic, name: "swift-e2e-group")
+ let groupID = Identifier(numeric: group.id)
+
+ // Polling a group this client never joined is an error, not
silence.
+ await expectIggyError(.consumerGroupMemberNotFound) {
+ _ = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitionID: nil, consumer: .group(groupID), strategy: .next, count: 2,
autoCommit: true)
+ }
+
+ try await client.joinConsumerGroup(streamID: scratch.stream,
topicID: scratch.topic, groupID: groupID)
+ for partition: UInt32 in 0..<3 {
+ _ = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .partition(partition), messages: try E2EEnvironment.messages(2))
+ }
+ // The server installs the member's partitions after the join, so
+ // the member polls until it has read every one of them. A group
+ // poll without a partition round-robins the owned partitions.
+ var polledPartitions = Set<UInt32>()
+ let polledAll = try await E2EEnvironment.eventually {
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitionID: nil, consumer: .group(groupID), strategy: .next, count: 2,
autoCommit: true)
+ if !polled.messages.isEmpty {
+ #expect(polled.messages.count == 2)
+ polledPartitions.insert(polled.partitionID)
+ }
+ return polledPartitions.count == 3
+ }
+ #expect(polledAll)
+ #expect(polledPartitions == [0, 1, 2])
+ try await client.leaveConsumerGroup(streamID: scratch.stream,
topicID: scratch.topic, groupID: groupID)
+ try await client.deleteConsumerGroup(streamID: scratch.stream,
topicID: scratch.topic, groupID: groupID)
+ }
+ }
+
+ @Test func segmentDeletion() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+ _ = try await client.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .partition(0), messages: try E2EEnvironment.messages(3))
+ // Only closed segments can be deleted; a fresh partition has one
open
+ // segment, so the request is accepted and removes nothing.
+ try await client.deleteSegments(streamID: scratch.stream, topicID:
scratch.topic, partitionID: 0, count: 1)
+ let topic = try #require(try await client.getTopic(streamID:
scratch.stream, topicID: scratch.topic))
+ #expect(topic.partitions[0].messagesCount == 3)
+ }
+ }
+
+ @Test func largePayloadsAndBigBatches() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+ let big = try IggyMessage(payload: [UInt8](repeating: 0xAB, count:
512 * 1024))
+ let many = try (0..<1000).map { try IggyMessage("message \($0)") }
+ _ = try await client.sendMessages(streamID: scratch.stream,
topicID: scratch.topic, partitioning: .partition(0), messages: [big] + many)
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID:
0, consumer: .default, strategy: .first, count: 1001, autoCommit: false)
+ #expect(polled.messages.count == 1001)
+ #expect(polled.messages[0].payload.count == 512 * 1024)
+ #expect(polled.messages[1000].payloadString == "message 999")
+ }
+ }
+
+ @Test func emptyBatchesAreRejectedLocally() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ await #expect(throws: IggyError(.invalidMessagesCount)) {
+ try await client.sendMessages(streamID: 1, topicID: 1,
partitioning: .balanced, messages: [])
+ }
+ await #expect(throws: IggyError.self) {
Review Comment:
Done in d5c1dbbfa, with one correction: against the server the code is
`topicIdNotFound`, not `streamIdNotFound`. A balanced send reads the topic
first, and Rust raises `TopicIdNotFound` at the same point (`messages.rs:123`).
The test asserts that exact code now.
##########
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:
Follow-up: changed in d5c1dbbfa. `sendMessages` now takes the batch `inout`
and writes minted ids back, as Rust does, so a retry after
`transientNotCommitted` carries the same ids.
--
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]
