Repository navigation
feat(swift): add the consumer - #4641
Merged
Merged
Conversation
An application reading a topic needs more than pollMessages: it has to keep each partition's position, continue after the last message it handled, commit offsets at the right moment, and come back after an outage or a rebalance. The Rust SDK does this in IggyConsumer, and the Swift SDK had no counterpart. IggyConsumer ports it as an actor that is also an AsyncSequence. A standalone consumer reads one partition, a group member joins its group and reads the server's assignment. Each poll continues after the last message handed over under the context that delivered it, a message already handed over from the same partition history is skipped unless replay is allowed, and the auto-commit modes, interval task, queued commits, and shutdown flush follow the Rust consumer. consumeMessages carries the after-handler commits and stops once the client cannot come back on its own. Encryption and resolving a lost group create or join need client features Swift does not have yet.
|
Thanks for the PR. It is labeled If AI was used to create this PR, you must run the Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
# Conflicts: # foreign/swift/README.md
Contributor
|
/skill team-review |
hubcio
requested changes
Oct 26, 2026
github-actions
Bot
added
S-waiting-on-author
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Summary: The new Swift consumer has three shutdown defects: a pending next() can outlive shutdown(), a join can leave a group member behind, and the interval drain can stall past its timeout. The review also finds canRecover wrong when the heartbeat is off, two simplifications, and two documentation nits.
Counts: critical 1, warning 3, nit 2, simplification 2
This review was generated by Claude Code 2.1.310 on deepseek-flash[1m]. Review the output before you act on it.
A consumer outlived its shutdown. A join in flight ended after the final leave. Without a heartbeat, a poll that waited for a lost client never woke. The drain also waited past offsetDrainTimeout. A tick that stored slept one more interval, and a waiter that gives up on task.value does not cancel the task. Joins and the final leave now share a lock. The poll wait now signs the client in and ends at shutdown. A tick ends after its store, and the drain cancels a task at its deadline. Overlapping stores moved the stored offset back, so stores and deletes now run one at a time. Under the next strategy, a poll that did not commit returned the same offset again and stalled the read. The consumer now continues after the last message, unless the poll commits. The iterator ends at its first error, as AsyncIteratorProtocol requires.
numinnex
approved these changes
Oct 26, 2026
mmodzelewski
approved these changes
Oct 26, 2026
spetz
approved these changes
Oct 26, 2026
hubcio
approved these changes
Oct 26, 2026
github-actions
Bot
removed
the
S-waiting-on-author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR address?
Relates to #4001
Rationale
The Swift client can poll messages and store offsets (#4553) and produce with
IggyProducer(#4577), but an application that reads a topic steadily needs what the RustIggyConsumergives it: tracking each partition's position, committing it at the right moment, and coming back after an outage or a rebalance. This is the next step of #4001: the consumer.What changed?
IggyConsumeris a port of the current Rust consumer, as an actor that is also anAsyncSequence.client.consumer(name:streamID:topicID:partitionID:)builds a standalone consumer of one partition andclient.consumerGroup(name:streamID:topicID:)a group member that joins its group, creating it when allowed, and reads the server's assignment.initialize()looks the stream and topic up with the configured retries.next()hands messages over one at a time from a buffer that a poll refills; each poll of a partition continues after the last message handed over, under the partition context that delivered it, and a message already handed over from the same partition history is skipped unlessallowReplayis set. A failed poll is thrown after the retry interval and leaves the consumer usable; a member that lost its membership joins again before the next poll.The auto-commit modes follow Rust: with the poll, on an interval, while messages are handed over (each, every nth offset, end of batch), and after a handler of
consumeMessages(_:)returned. Queued commits collapse to the latest per partition, a position the last store already covers is not sent again, andshutdown()drains the commit tasks, stores the reading position of every partition read unless auto-commit is disabled, and leaves the group. Concurrentinitialize()andshutdown()calls share one run, a dropped consumer ends its tasks, and options out of range are refused when the consumer is built. The client gains an internalcanRecover, whichconsumeMessages(_:)uses to stop once the client cannot come back without the caller.The Swift client has no encryptor and no operation lanes yet, so the consumer does not decrypt and does not resolve a group create or join whose reply was lost; the docs and README say so.
Of the 1,588 added lines, 664 are tests.
Local Execution
Run on macOS (Swift 6.4) on the final commit:
swift format lint --strict,swift build -Xswiftc -warnings-as-errors(sources and tests),swift build -c release,swift test(395 tests, three runs), the golden lane (regenerate, no diff), andprek runover the commit. The same build and tests passed in Docker on Linux with Swift 6.4 and with Swift 6.1, the toolchain of the macOS lane.The end-to-end suite (50 tests, 2 of them for the consumer) passed against
iggy-serverbuilt from current master, started the way the CI lane starts it: plain, with TLS using the certificates incore/certs, and as a two-node cluster.AI Usage
core/sdk/src/clients/consumer.rs,consumer_builder.rs, andconsumer_ext/consumer_message_ext.rs, applying the patterns from the review rounds on the earlier Swift PRs.consumeMessagesfor a client that cannot come back, and the after-handler triggers were each broken on purpose and a test failed or hung for each.