From 2b3e5b0ca9d152ee291c66f8bf5276dcd8e8d27c Mon Sep 17 00:00:00 2001 From: ssikka Date: Sat, 11 Jul 2026 11:04:02 -0400 Subject: [PATCH 1/7] Document configuration types via DocC extension files Add documentation-extension files providing abstracts and overviews for KafkaConfig, KafkaConsumerConfig, and KafkaProducerConfig, which are generated from .gyb templates and had no type-level doc comments. Documents them without touching the generated sources. --- Sources/Kafka/Kafka.docc/KafkaConfig.md | 7 +++++++ Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md | 7 +++++++ Sources/Kafka/Kafka.docc/KafkaProducerConfig.md | 7 +++++++ 3 files changed, 21 insertions(+) create mode 100644 Sources/Kafka/Kafka.docc/KafkaConfig.md create mode 100644 Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md create mode 100644 Sources/Kafka/Kafka.docc/KafkaProducerConfig.md diff --git a/Sources/Kafka/Kafka.docc/KafkaConfig.md b/Sources/Kafka/Kafka.docc/KafkaConfig.md new file mode 100644 index 00000000..1734a9f0 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/KafkaConfig.md @@ -0,0 +1,7 @@ +# ``Kafka/KafkaConfig`` + +A namespace of typed option values for Kafka client configuration properties. + +## Overview + +Nested types provide type-safe wrappers for the string values librdkafka accepts for enum-valued options such as ``KafkaConfig/SecurityProtocol``, ``KafkaConfig/CompressionCodec``, and ``KafkaConfig/GroupProtocol``. Assign these values to properties on ``KafkaConsumerConfig`` and ``KafkaProducerConfig``. diff --git a/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md new file mode 100644 index 00000000..0018e54e --- /dev/null +++ b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md @@ -0,0 +1,7 @@ +# ``Kafka/KafkaConsumerConfig`` + +Configures a ``KafkaConsumer``. + +## Overview + +Provide broker addresses through ``KafkaConsumerConfig/bootstrapServers`` and a consumption strategy through ``KafkaConsumerConfig/consumptionStrategy``, then pass the configuration to a ``KafkaConsumer`` initializer. Additional properties mirror librdkafka's consumer configuration surface; consult individual property documentation for defaults and semantics. diff --git a/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md new file mode 100644 index 00000000..9794cd42 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md @@ -0,0 +1,7 @@ +# ``Kafka/KafkaProducerConfig`` + +Configures a ``KafkaProducer``. + +## Overview + +Provide broker addresses through ``KafkaProducerConfig/bootstrapServers``, then pass the configuration to a ``KafkaProducer`` initializer. Additional properties mirror librdkafka's producer configuration surface; consult individual property documentation for defaults and semantics. From 2fe3d7e197bc0dff707409bcacfee71ac2c03947 Mon Sep 17 00:00:00 2001 From: ssikka Date: Sat, 11 Jul 2026 11:04:02 -0400 Subject: [PATCH 2/7] Document consumer rebalances and the events sequence Add an 'Observe rebalances' section to ConsumingMessages demonstrating makeConsumerWithEvents and iterating KafkaConsumerEvents (previously the events sequence was only linked, never shown), and curate the events entry point in Topics. --- Sources/Kafka/Kafka.docc/ConsumingMessages.md | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/Sources/Kafka/Kafka.docc/ConsumingMessages.md b/Sources/Kafka/Kafka.docc/ConsumingMessages.md index a6859690..adc2aceb 100644 --- a/Sources/Kafka/Kafka.docc/ConsumingMessages.md +++ b/Sources/Kafka/Kafka.docc/ConsumingMessages.md @@ -134,6 +134,64 @@ try consumer.resume(topicPartitions: [partition]) While a partition is paused, the consumer stops fetching records for it but continues to participate in the group, including heartbeats and rebalances. +### Observe rebalances + +When the membership of a consumer group changes — a consumer joins, leaves, or fails — Kafka redistributes the group's partitions across the remaining members. This is a *rebalance*. ``KafkaConsumer`` performs the assign and unassign automatically and surfaces a ``KafkaConsumerRebalance`` notification through the ``KafkaConsumerEvents`` sequence, so you can react — for example, by committing offsets for partitions that are moving away. + +Create the consumer with ``KafkaConsumer/makeConsumerWithEvents(config:logger:)`` and iterate the event sequence alongside the messages: + +```swift +config.partitionAssignmentStrategy = "cooperative-sticky" + +let (consumer, events) = try KafkaConsumer.makeConsumerWithEvents(config: config, logger: logger) + +let serviceGroup = ServiceGroup( + services: [consumer], + gracefulShutdownSignals: [.sigterm], + logger: logger +) + +await withThrowingTaskGroup(of: Void.self) { group in + group.addTask { try await serviceGroup.run() } + + // Consume records. + group.addTask { + for try await message in consumer.messages { + // Process the record, then store or commit its offset. + } + } + + // Observe rebalances. + group.addTask { + for await event in events { + switch event { + case .rebalance(let rebalance): + switch rebalance.kind { + case .assign: + // Partitions were assigned; initialize per-partition state. + break + case .revoke: + // Partitions are moving away; commit their offsets first. + break + case .error(let description): + logger.warning("Rebalance error", metadata: ["error": "\(description)"]) + } + case .error(let error): + logger.error("Kafka client error", metadata: ["error": "\(error)"]) + default: + break + } + } + } +} +``` + +Each ``KafkaConsumerRebalance`` reports its ``KafkaConsumerRebalance/kind`` — ``KafkaConsumerRebalance/Kind/assign``, ``KafkaConsumerRebalance/Kind/revoke``, or ``KafkaConsumerRebalance/Kind/error(_:)`` — and the ``KafkaConsumerRebalance/partitions`` involved. Commit offsets on revoke so another consumer resumes from the right position. + +Choose an assignment strategy with ``KafkaConsumerConfig/partitionAssignmentStrategy``: `cooperative-sticky` for incremental cooperative rebalancing, or `range` and `roundrobin` for eager rebalancing. ``KafkaConsumer`` adapts to the negotiated protocol, so your handling code is identical either way. Cooperative and eager strategies must not be mixed within a group. + +> Important: Consume the events sequence. If you create it but never iterate it, events buffer in memory indefinitely. + ## Topics ### Reading records @@ -161,6 +219,7 @@ While a partition is paused, the consumer stops fetching records for it but cont ### Observing rebalances and events +- ``KafkaConsumer/makeConsumerWithEvents(config:logger:)`` - ``KafkaConsumerRebalance`` - ``KafkaConsumerEvent`` - ``KafkaConsumerEvents`` From a35ea3bf37313cf36c5533f97ddd75207eff410c Mon Sep 17 00:00:00 2001 From: ssikka Date: Sat, 11 Jul 2026 11:04:02 -0400 Subject: [PATCH 3/7] Add error-handling, observability, and migration articles Add three standalone articles: HandlingErrors (where KafkaError surfaces and how to classify it), Observability (enabling swift-metrics gauges and the structured logging metadata the client attaches), and Migrating (moving off the deprecated Configuration types and commitSync). Link them from the landing page. --- Sources/Kafka/Kafka.docc/HandlingErrors.md | 62 ++++++++++++++++++++ Sources/Kafka/Kafka.docc/Kafka.md | 3 + Sources/Kafka/Kafka.docc/Migrating.md | 51 +++++++++++++++++ Sources/Kafka/Kafka.docc/Observability.md | 66 ++++++++++++++++++++++ 4 files changed, 182 insertions(+) create mode 100644 Sources/Kafka/Kafka.docc/HandlingErrors.md create mode 100644 Sources/Kafka/Kafka.docc/Migrating.md create mode 100644 Sources/Kafka/Kafka.docc/Observability.md diff --git a/Sources/Kafka/Kafka.docc/HandlingErrors.md b/Sources/Kafka/Kafka.docc/HandlingErrors.md new file mode 100644 index 00000000..4270ac94 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/HandlingErrors.md @@ -0,0 +1,62 @@ +# Handling errors + +Learn where Kafka errors surface and how to classify them to decide whether to retry, recreate the client, or fail. + +## Overview + +Both the producer and the consumer report failures as ``KafkaError`` values. A ``KafkaError`` carries a ``KafkaError/code`` describing the kind of failure and, when the error originates from librdkafka, a ``KafkaError/rdKafkaCode`` with the specific underlying code. The ``KafkaError/isFatal`` and ``KafkaError/isRetriable`` flags tell you how to respond. + +## Where errors surface + +A ``KafkaError`` reaches your code in three ways: + +- **Thrown** from throwing calls such as ``KafkaProducer/send(_:)``, ``KafkaProducer/sendAndAwait(_:)``, ``KafkaConsumer/subscribe(topics:)``, and ``KafkaConsumer/commit(_:)``. +- **On an event sequence**, as ``KafkaConsumerEvent/error(_:)`` or ``KafkaProducerEvent/error(_:)`` — for example, a broker disconnection or an authentication failure. +- **In a delivery report**, when a ``KafkaDeliveryReport``'s status is `failure` because a message could not be delivered. + +## Classify an error + +Use ``KafkaError/isFatal`` and ``KafkaError/isRetriable`` to decide how to respond, and inspect ``KafkaError/rdKafkaCode`` for finer-grained handling: + +```swift +func handle(_ error: KafkaError, logger: Logger) { + // A fatal error means the client instance can no longer be used: + // shut it down and create a new one. + if error.isFatal { + logger.critical("Fatal Kafka error, recreate the client", metadata: ["error": "\(error)"]) + return + } + + // A retriable error may succeed if the operation is attempted again. + if error.isRetriable { + logger.warning("Retriable Kafka error, retrying", metadata: ["error": "\(error)"]) + return + } + + // For finer-grained handling, inspect the underlying librdkafka code. + switch error.rdKafkaCode { + case .allBrokersDown, .transport: + logger.warning("Lost connectivity to the cluster", metadata: ["error": "\(error)"]) + case .authentication, .ssl: + logger.error("Authentication or TLS failure", metadata: ["error": "\(error)"]) + default: + logger.error("Kafka error", metadata: ["kind": "\(error.code)", "error": "\(error)"]) + } +} +``` + +``KafkaError/rdKafkaCode`` is `nil` for errors that don't originate from librdkafka — such as configuration or lifecycle errors — in which case ``KafkaError/isFatal`` and ``KafkaError/isRetriable`` are both `false`. + +## Topics + +### Errors + +- ``KafkaError`` +- ``KafkaError/ErrorCode`` +- ``KafkaError/RDKafkaCode`` + +### Error events + +- ``KafkaConsumerEvent`` +- ``KafkaProducerEvent`` +- ``KafkaDeliveryReport`` diff --git a/Sources/Kafka/Kafka.docc/Kafka.md b/Sources/Kafka/Kafka.docc/Kafka.md index 93f6e0d6..939c5db1 100644 --- a/Sources/Kafka/Kafka.docc/Kafka.md +++ b/Sources/Kafka/Kafka.docc/Kafka.md @@ -15,7 +15,10 @@ Kafka integrates with [swift-log](https://github.com/apple/swift-log) for struct - - - +- +- - +- ### Producing messages diff --git a/Sources/Kafka/Kafka.docc/Migrating.md b/Sources/Kafka/Kafka.docc/Migrating.md new file mode 100644 index 00000000..e5af28e1 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/Migrating.md @@ -0,0 +1,51 @@ +# Migrating to the current API + +Move from the deprecated configuration types and methods to their current replacements. + +## Overview + +Earlier versions of the client configured a ``KafkaConsumer`` or ``KafkaProducer`` with the ``KafkaConsumerConfiguration`` and ``KafkaProducerConfiguration`` types. These are deprecated in favor of ``KafkaConsumerConfig`` and ``KafkaProducerConfig``, which expose librdkafka's full configuration surface as typed Swift properties. + +The deprecated types still compile, but the compiler emits a warning pointing to the replacement. Adopt the current API to silence those warnings and to reach configuration options the older types don't expose. + +## Configuration types + +Replace each `Configuration` type with the corresponding `Config` type: + +| Deprecated | Current | +| --- | --- | +| ``KafkaConsumerConfiguration`` | ``KafkaConsumerConfig`` | +| ``KafkaProducerConfiguration`` | ``KafkaProducerConfig`` | + +## Initializers + +The client initializers now take a `config:` parameter of the `Config` type in place of the `configuration:` parameter of the `Configuration` type: + +```swift +// Deprecated +let consumer = try KafkaConsumer(configuration: consumerConfiguration, logger: logger) + +// Current +var config = KafkaConsumerConfig() +config.bootstrapServers = ["localhost:9092"] +config.consumptionStrategy = .group(id: "example-group", topics: ["topic-name"]) +let consumer = try KafkaConsumer(config: config, logger: logger) +``` + +For the event-carrying variants, use ``KafkaConsumer/makeConsumerWithEvents(config:logger:)`` and ``KafkaProducer/makeProducerWithEvents(config:logger:)``, which likewise take the `Config` type. + +## Renamed methods + +``KafkaConsumer/commitSync(_:)`` was renamed to ``KafkaConsumer/commit(_:)``. The behavior is unchanged. + +## Topics + +### Current configuration types + +- ``KafkaConsumerConfig`` +- ``KafkaProducerConfig`` + +### Deprecated types + +- ``KafkaConsumerConfiguration`` +- ``KafkaProducerConfiguration`` diff --git a/Sources/Kafka/Kafka.docc/Observability.md b/Sources/Kafka/Kafka.docc/Observability.md new file mode 100644 index 00000000..5e8b5008 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/Observability.md @@ -0,0 +1,66 @@ +# Observability + +Emit metrics and structured logs from the producer and consumer. + +## Overview + +The Kafka client integrates with [swift-metrics](https://github.com/apple/swift-metrics) for runtime metrics and [swift-log](https://github.com/apple/swift-log) for structured logging. Both are opt-in through the client configuration and the `Logger` you provide. + +## Emit metrics + +The client periodically samples librdkafka's internal statistics and records them into [swift-metrics](https://github.com/apple/swift-metrics) gauges that you supply. Configure this through the ``KafkaConsumerConfig/metrics`` (or ``KafkaProducerConfig/metrics``) property: set an update interval and assign a `Gauge` to each statistic you want to track. + +```swift +import Kafka +import Metrics + +var config = KafkaConsumerConfig() +config.bootstrapServers = ["localhost:9092"] +config.consumptionStrategy = .group(id: "example-group", topics: ["topic-name"]) + +// Sample statistics once per second into swift-metrics gauges. +config.metrics.updateInterval = .seconds(1) +config.metrics.totalKafkaBrokerRequests = Gauge(label: "kafka_consumer_broker_requests") +config.metrics.totalKafkaBrokerMessagesReceived = Gauge(label: "kafka_consumer_messages_received") +config.metrics.queuedOperation = Gauge(label: "kafka_consumer_queued_operations") + +let consumer = try KafkaConsumer(config: config, logger: logger) +``` + +Metrics are emitted only when ``KafkaConfiguration/ConsumerMetrics/updateInterval`` is set **and** at least one gauge is assigned; otherwise the client skips statistics collection entirely. The producer exposes the same pattern through ``KafkaProducerConfig/metrics`` with producer-specific gauges such as ``KafkaConfiguration/ProducerMetrics/queuedProducerMessages`` and ``KafkaConfiguration/ProducerMetrics/totalKafkaBrokerMessagesSent``. + +The gauges you assign are ordinary [swift-metrics](https://github.com/apple/swift-metrics) types, so the values reach whatever metrics backend you bootstrap through `MetricsSystem`. + +## Structured logging + +Pass a `Logger` when creating a ``KafkaProducer`` or ``KafkaConsumer``. The client logs lifecycle and operational events through it, and enriches every entry with structured metadata so you can filter and correlate logs across many clients: + +| Metadata key | Value | +| --- | --- | +| `kafka.client.id` | the configured `clientId` | +| `kafka.client.type` | `producer` or `consumer` | +| `kafka.group.id` | the consumer group (consumers only) | + +```swift +import Kafka +import Logging + +let logger = Logger(label: "kafka") + +var config = KafkaConsumerConfig() +config.clientId = "orders-consumer" +config.consumptionStrategy = .group(id: "orders", topics: ["orders"]) + +let consumer = try KafkaConsumer(config: config, logger: logger) +// Every log entry from this consumer now carries kafka.client.id, kafka.client.type, +// and kafka.group.id. +``` + +Set the `Logger`'s log level to control verbosity — the client logs routine progress at `debug` and `trace`, and surfaces problems at `info` and above. + +## Topics + +### Metrics + +- ``KafkaConfiguration/ConsumerMetrics`` +- ``KafkaConfiguration/ProducerMetrics`` From 43ac7fa4ad4e3efc88076dda0751928ccedaa23e Mon Sep 17 00:00:00 2001 From: ssikka Date: Tue, 14 Jul 2026 15:22:17 -0400 Subject: [PATCH 4/7] Address review: prose abstracts, active voice, remove Migrating MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - KafkaConsumerConfig.md, KafkaProducerConfig.md: replace symbol link abstracts with narrative prose (heckj) - HandlingErrors.md: make abstract active voice (heckj) - Observability.md: rename to 'Observing Kafka clients' to match sibling gerunds; active voice; drop librdkafka jargon; parallel imperative heading 'Emit structured logs' (heckj) - Remove Migrating.md — belongs in 1.0 release notes, not DocC (FranzBusch) --- Sources/Kafka/Kafka.docc/HandlingErrors.md | 2 +- Sources/Kafka/Kafka.docc/Kafka.md | 1 - .../Kafka/Kafka.docc/KafkaConsumerConfig.md | 2 +- .../Kafka/Kafka.docc/KafkaProducerConfig.md | 2 +- Sources/Kafka/Kafka.docc/Migrating.md | 51 ------------------- Sources/Kafka/Kafka.docc/Observability.md | 8 +-- 6 files changed, 7 insertions(+), 59 deletions(-) delete mode 100644 Sources/Kafka/Kafka.docc/Migrating.md diff --git a/Sources/Kafka/Kafka.docc/HandlingErrors.md b/Sources/Kafka/Kafka.docc/HandlingErrors.md index 4270ac94..bf313b80 100644 --- a/Sources/Kafka/Kafka.docc/HandlingErrors.md +++ b/Sources/Kafka/Kafka.docc/HandlingErrors.md @@ -1,6 +1,6 @@ # Handling errors -Learn where Kafka errors surface and how to classify them to decide whether to retry, recreate the client, or fail. +Find where Kafka errors surface and classify them to decide whether to retry, recreate the client, or fail. ## Overview diff --git a/Sources/Kafka/Kafka.docc/Kafka.md b/Sources/Kafka/Kafka.docc/Kafka.md index 939c5db1..15ea1524 100644 --- a/Sources/Kafka/Kafka.docc/Kafka.md +++ b/Sources/Kafka/Kafka.docc/Kafka.md @@ -18,7 +18,6 @@ Kafka integrates with [swift-log](https://github.com/apple/swift-log) for struct - - - -- ### Producing messages diff --git a/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md index 0018e54e..4831f45b 100644 --- a/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md +++ b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md @@ -1,6 +1,6 @@ # ``Kafka/KafkaConsumerConfig`` -Configures a ``KafkaConsumer``. +Configures a Kafka consumer. ## Overview diff --git a/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md index 9794cd42..952757a0 100644 --- a/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md +++ b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md @@ -1,6 +1,6 @@ # ``Kafka/KafkaProducerConfig`` -Configures a ``KafkaProducer``. +Configures a Kafka producer. ## Overview diff --git a/Sources/Kafka/Kafka.docc/Migrating.md b/Sources/Kafka/Kafka.docc/Migrating.md deleted file mode 100644 index e5af28e1..00000000 --- a/Sources/Kafka/Kafka.docc/Migrating.md +++ /dev/null @@ -1,51 +0,0 @@ -# Migrating to the current API - -Move from the deprecated configuration types and methods to their current replacements. - -## Overview - -Earlier versions of the client configured a ``KafkaConsumer`` or ``KafkaProducer`` with the ``KafkaConsumerConfiguration`` and ``KafkaProducerConfiguration`` types. These are deprecated in favor of ``KafkaConsumerConfig`` and ``KafkaProducerConfig``, which expose librdkafka's full configuration surface as typed Swift properties. - -The deprecated types still compile, but the compiler emits a warning pointing to the replacement. Adopt the current API to silence those warnings and to reach configuration options the older types don't expose. - -## Configuration types - -Replace each `Configuration` type with the corresponding `Config` type: - -| Deprecated | Current | -| --- | --- | -| ``KafkaConsumerConfiguration`` | ``KafkaConsumerConfig`` | -| ``KafkaProducerConfiguration`` | ``KafkaProducerConfig`` | - -## Initializers - -The client initializers now take a `config:` parameter of the `Config` type in place of the `configuration:` parameter of the `Configuration` type: - -```swift -// Deprecated -let consumer = try KafkaConsumer(configuration: consumerConfiguration, logger: logger) - -// Current -var config = KafkaConsumerConfig() -config.bootstrapServers = ["localhost:9092"] -config.consumptionStrategy = .group(id: "example-group", topics: ["topic-name"]) -let consumer = try KafkaConsumer(config: config, logger: logger) -``` - -For the event-carrying variants, use ``KafkaConsumer/makeConsumerWithEvents(config:logger:)`` and ``KafkaProducer/makeProducerWithEvents(config:logger:)``, which likewise take the `Config` type. - -## Renamed methods - -``KafkaConsumer/commitSync(_:)`` was renamed to ``KafkaConsumer/commit(_:)``. The behavior is unchanged. - -## Topics - -### Current configuration types - -- ``KafkaConsumerConfig`` -- ``KafkaProducerConfig`` - -### Deprecated types - -- ``KafkaConsumerConfiguration`` -- ``KafkaProducerConfiguration`` diff --git a/Sources/Kafka/Kafka.docc/Observability.md b/Sources/Kafka/Kafka.docc/Observability.md index 5e8b5008..2200dd02 100644 --- a/Sources/Kafka/Kafka.docc/Observability.md +++ b/Sources/Kafka/Kafka.docc/Observability.md @@ -1,4 +1,4 @@ -# Observability +# Observing Kafka clients Emit metrics and structured logs from the producer and consumer. @@ -8,7 +8,7 @@ The Kafka client integrates with [swift-metrics](https://github.com/apple/swift- ## Emit metrics -The client periodically samples librdkafka's internal statistics and records them into [swift-metrics](https://github.com/apple/swift-metrics) gauges that you supply. Configure this through the ``KafkaConsumerConfig/metrics`` (or ``KafkaProducerConfig/metrics``) property: set an update interval and assign a `Gauge` to each statistic you want to track. +The client periodically samples internal statistics and records them into [swift-metrics](https://github.com/apple/swift-metrics) gauges that you supply. Configure this through the ``KafkaConsumerConfig/metrics`` (or ``KafkaProducerConfig/metrics``) property: set an update interval and assign a `Gauge` to each statistic you want to track. ```swift import Kafka @@ -27,11 +27,11 @@ config.metrics.queuedOperation = Gauge(label: "kafka_consumer_queued_operations" let consumer = try KafkaConsumer(config: config, logger: logger) ``` -Metrics are emitted only when ``KafkaConfiguration/ConsumerMetrics/updateInterval`` is set **and** at least one gauge is assigned; otherwise the client skips statistics collection entirely. The producer exposes the same pattern through ``KafkaProducerConfig/metrics`` with producer-specific gauges such as ``KafkaConfiguration/ProducerMetrics/queuedProducerMessages`` and ``KafkaConfiguration/ProducerMetrics/totalKafkaBrokerMessagesSent``. +The client emits metrics only when ``KafkaConfiguration/ConsumerMetrics/updateInterval`` is set **and** at least one gauge is assigned; otherwise the client skips statistics collection entirely. The producer exposes the same pattern through ``KafkaProducerConfig/metrics`` with producer-specific gauges such as ``KafkaConfiguration/ProducerMetrics/queuedProducerMessages`` and ``KafkaConfiguration/ProducerMetrics/totalKafkaBrokerMessagesSent``. The gauges you assign are ordinary [swift-metrics](https://github.com/apple/swift-metrics) types, so the values reach whatever metrics backend you bootstrap through `MetricsSystem`. -## Structured logging +## Emit structured logs Pass a `Logger` when creating a ``KafkaProducer`` or ``KafkaConsumer``. The client logs lifecycle and operational events through it, and enriches every entry with structured metadata so you can filter and correlate logs across many clients: From bfdd4e88038644f044ccfbd9f5f98399b8d2c3bb Mon Sep 17 00:00:00 2001 From: ssikka Date: Wed, 15 Jul 2026 15:59:31 -0400 Subject: [PATCH 5/7] Cross-link DocC articles and add symbol pages for consumer/producer Adopts the Swift-standard DocC pattern used by swift-nio and swift-metrics: each article groups related articles under `### Articles` before functional symbol groups, and `KafkaConsumer` / `KafkaProducer` symbol extensions add functional topic groups so members render in curated buckets. Config-type extension files keep their prose abstracts and add inline `` links to the security and walkthrough articles. --- Sources/Kafka/Kafka.docc/ConsumingMessages.md | 7 ++ Sources/Kafka/Kafka.docc/HandlingErrors.md | 6 ++ Sources/Kafka/Kafka.docc/KafkaConsumer.md | 64 +++++++++++++++++++ .../Kafka/Kafka.docc/KafkaConsumerConfig.md | 2 + Sources/Kafka/Kafka.docc/KafkaProducer.md | 39 +++++++++++ .../Kafka/Kafka.docc/KafkaProducerConfig.md | 2 + Sources/Kafka/Kafka.docc/Observability.md | 6 ++ Sources/Kafka/Kafka.docc/ProducingMessages.md | 7 ++ .../Kafka/Kafka.docc/SecuringConnections.md | 13 ++++ 9 files changed, 146 insertions(+) create mode 100644 Sources/Kafka/Kafka.docc/KafkaConsumer.md create mode 100644 Sources/Kafka/Kafka.docc/KafkaProducer.md diff --git a/Sources/Kafka/Kafka.docc/ConsumingMessages.md b/Sources/Kafka/Kafka.docc/ConsumingMessages.md index adc2aceb..9bf5f77d 100644 --- a/Sources/Kafka/Kafka.docc/ConsumingMessages.md +++ b/Sources/Kafka/Kafka.docc/ConsumingMessages.md @@ -194,6 +194,13 @@ Choose an assignment strategy with ``KafkaConsumerConfig/partitionAssignmentStra ## Topics +### Articles + +- +- +- +- + ### Reading records - ``KafkaConsumer/messages`` diff --git a/Sources/Kafka/Kafka.docc/HandlingErrors.md b/Sources/Kafka/Kafka.docc/HandlingErrors.md index bf313b80..eb2e3028 100644 --- a/Sources/Kafka/Kafka.docc/HandlingErrors.md +++ b/Sources/Kafka/Kafka.docc/HandlingErrors.md @@ -49,6 +49,12 @@ func handle(_ error: KafkaError, logger: Logger) { ## Topics +### Articles + +- +- +- + ### Errors - ``KafkaError`` diff --git a/Sources/Kafka/Kafka.docc/KafkaConsumer.md b/Sources/Kafka/Kafka.docc/KafkaConsumer.md new file mode 100644 index 00000000..b0789759 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/KafkaConsumer.md @@ -0,0 +1,64 @@ +# ``Kafka/KafkaConsumer`` + +Consumes messages from a Kafka cluster as part of a service lifecycle. + +## Overview + +Create a consumer from a ``KafkaConsumerConfig``, run it inside a `ServiceGroup`, and iterate ``KafkaConsumer/messages`` to receive records. The consumer conforms to `Service`, so the surrounding application controls its lifecycle; the `run()` method drives the underlying poll loop until the calling task is canceled or a graceful shutdown is triggered. + +By default, the consumer stores and commits offsets automatically as iteration proceeds. For at-least-once delivery, disable automatic offset storage and call ``storeOffset(_:)`` after processing each record. For full control over commit timing, disable auto-commit as well and use ``commit(_:)`` or ``commit()`` explicitly. + +For an end-to-end guide including configuration, rebalance handling, and offset patterns, see . + +## Topics + +### Creating a consumer + +- ``init(config:logger:)`` +- ``makeConsumerWithEvents(config:logger:)`` + +### Consuming messages + +- ``messages`` +- ``KafkaConsumerMessages`` +- ``KafkaConsumerMessage`` + +### Running the service + +- ``run()`` +- ``triggerGracefulShutdown()`` + +### Managing subscriptions + +- ``subscribe(topics:)`` +- ``unsubscribe()`` +- ``subscribedTopics()`` + +### Storing and committing offsets + +- ``storeOffset(_:)`` +- ``commit(_:)`` +- ``commit()`` +- ``scheduleCommit(_:)`` +- ``scheduleCommit()`` + +### Querying position + +- ``committed(topicPartitions:timeout:)`` +- ``position(topicPartitions:)`` +- ``isAssignmentLost`` + +### Pausing and resuming partitions + +- ``pause(topicPartitions:)`` +- ``resume(topicPartitions:)`` + +### Seeking + +- ``seek(topicPartitionOffsets:timeout:)`` + +### Observing events + +- ``KafkaConsumerEvents`` +- ``KafkaConsumerEvent`` +- ``KafkaConsumerRebalance`` diff --git a/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md index 4831f45b..d8639882 100644 --- a/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md +++ b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md @@ -5,3 +5,5 @@ Configures a Kafka consumer. ## Overview Provide broker addresses through ``KafkaConsumerConfig/bootstrapServers`` and a consumption strategy through ``KafkaConsumerConfig/consumptionStrategy``, then pass the configuration to a ``KafkaConsumer`` initializer. Additional properties mirror librdkafka's consumer configuration surface; consult individual property documentation for defaults and semantics. + +For security options, see . For an end-to-end example, see . diff --git a/Sources/Kafka/Kafka.docc/KafkaProducer.md b/Sources/Kafka/Kafka.docc/KafkaProducer.md new file mode 100644 index 00000000..af784381 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/KafkaProducer.md @@ -0,0 +1,39 @@ +# ``Kafka/KafkaProducer`` + +Sends messages to a Kafka cluster as part of a service lifecycle. + +## Overview + +Create a producer from a ``KafkaProducerConfig``, run it inside a `ServiceGroup`, and call ``send(_:)`` or ``sendAndAwait(_:)`` to publish records. The producer conforms to `Service`, so the surrounding application controls its lifecycle; the `run()` method drives the internal poll loop, dispatches delivery reports, and flushes outstanding messages on graceful shutdown. + +Two send styles are available. Use ``sendAndAwait(_:)`` when the caller needs the delivery outcome inline — the method suspends until the broker acknowledges (or rejects) the message and returns a ``KafkaDeliveryReport``. Use ``send(_:)`` paired with ``makeProducerWithEvents(config:logger:)`` for maximum throughput, processing delivery reports asynchronously through the events sequence. + +When messages are published to a nonexistent topic, a new topic is created using the default topic configuration (based on ``KafkaProducerConfig`` topic-level properties). + +For an end-to-end guide including configuration, delivery patterns, and event handling, see . + +## Topics + +### Creating a producer + +- ``init(config:logger:)`` +- ``makeProducerWithEvents(config:logger:)`` + +### Sending messages + +- ``send(_:)`` +- ``sendAndAwait(_:)`` +- ``KafkaProducerMessage`` +- ``KafkaProducerMessageID`` + +### Running the service + +- ``run()`` +- ``triggerGracefulShutdown()`` + +### Observing outcomes + +- ``KafkaProducerEvents`` +- ``KafkaProducerEvent`` +- ``KafkaDeliveryReport`` +- ``KafkaAcknowledgedMessage`` diff --git a/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md index 952757a0..267ce2c6 100644 --- a/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md +++ b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md @@ -5,3 +5,5 @@ Configures a Kafka producer. ## Overview Provide broker addresses through ``KafkaProducerConfig/bootstrapServers``, then pass the configuration to a ``KafkaProducer`` initializer. Additional properties mirror librdkafka's producer configuration surface; consult individual property documentation for defaults and semantics. + +For security options, see . For an end-to-end example, see . diff --git a/Sources/Kafka/Kafka.docc/Observability.md b/Sources/Kafka/Kafka.docc/Observability.md index 2200dd02..d3a59ad3 100644 --- a/Sources/Kafka/Kafka.docc/Observability.md +++ b/Sources/Kafka/Kafka.docc/Observability.md @@ -60,6 +60,12 @@ Set the `Logger`'s log level to control verbosity — the client logs routine pr ## Topics +### Articles + +- +- +- + ### Metrics - ``KafkaConfiguration/ConsumerMetrics`` diff --git a/Sources/Kafka/Kafka.docc/ProducingMessages.md b/Sources/Kafka/Kafka.docc/ProducingMessages.md index 8a91658b..59ef922c 100644 --- a/Sources/Kafka/Kafka.docc/ProducingMessages.md +++ b/Sources/Kafka/Kafka.docc/ProducingMessages.md @@ -99,6 +99,13 @@ Beyond delivery reports, the events sequence emits errors and other broker event ## Topics +### Articles + +- +- +- +- + ### Sending records - ``KafkaProducer/send(_:)`` diff --git a/Sources/Kafka/Kafka.docc/SecuringConnections.md b/Sources/Kafka/Kafka.docc/SecuringConnections.md index a61ea459..f3fda27b 100644 --- a/Sources/Kafka/Kafka.docc/SecuringConnections.md +++ b/Sources/Kafka/Kafka.docc/SecuringConnections.md @@ -55,3 +55,16 @@ config.saslPassword = "password" ``` `SCRAM-SHA-256` and `SCRAM-SHA-512` keep the password from traveling in the clear during the handshake; `PLAIN` does not, so reserve `PLAIN` for transports already protected by TLS. + +## Topics + +### Articles + +- +- + +### Configuration + +- ``KafkaProducerConfig`` +- ``KafkaConsumerConfig`` +- ``KafkaConfig`` From 881702640237cafefada054d8ea3989f798daa62 Mon Sep 17 00:00:00 2001 From: ssikka Date: Wed, 15 Jul 2026 16:01:43 -0400 Subject: [PATCH 6/7] Drop librdkafka from config-type overviews Applies heckj's rule about keeping implementation-detail library names out of user-facing conceptual prose, extending it to the config extension files and namespace type. --- Sources/Kafka/Kafka.docc/KafkaConfig.md | 2 +- Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md | 2 +- Sources/Kafka/Kafka.docc/KafkaProducerConfig.md | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Sources/Kafka/Kafka.docc/KafkaConfig.md b/Sources/Kafka/Kafka.docc/KafkaConfig.md index 1734a9f0..fe4a7eda 100644 --- a/Sources/Kafka/Kafka.docc/KafkaConfig.md +++ b/Sources/Kafka/Kafka.docc/KafkaConfig.md @@ -4,4 +4,4 @@ A namespace of typed option values for Kafka client configuration properties. ## Overview -Nested types provide type-safe wrappers for the string values librdkafka accepts for enum-valued options such as ``KafkaConfig/SecurityProtocol``, ``KafkaConfig/CompressionCodec``, and ``KafkaConfig/GroupProtocol``. Assign these values to properties on ``KafkaConsumerConfig`` and ``KafkaProducerConfig``. +Nested types provide type-safe wrappers for enum-valued configuration options such as ``KafkaConfig/SecurityProtocol``, ``KafkaConfig/CompressionCodec``, and ``KafkaConfig/GroupProtocol``. Assign these values to properties on ``KafkaConsumerConfig`` and ``KafkaProducerConfig``. diff --git a/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md index d8639882..514edd31 100644 --- a/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md +++ b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md @@ -4,6 +4,6 @@ Configures a Kafka consumer. ## Overview -Provide broker addresses through ``KafkaConsumerConfig/bootstrapServers`` and a consumption strategy through ``KafkaConsumerConfig/consumptionStrategy``, then pass the configuration to a ``KafkaConsumer`` initializer. Additional properties mirror librdkafka's consumer configuration surface; consult individual property documentation for defaults and semantics. +Provide broker addresses through ``KafkaConsumerConfig/bootstrapServers`` and a consumption strategy through ``KafkaConsumerConfig/consumptionStrategy``, then pass the configuration to a ``KafkaConsumer`` initializer. Additional properties expose the consumer's configuration options; consult individual property documentation for defaults and semantics. For security options, see . For an end-to-end example, see . diff --git a/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md index 267ce2c6..0be4fb7e 100644 --- a/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md +++ b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md @@ -4,6 +4,6 @@ Configures a Kafka producer. ## Overview -Provide broker addresses through ``KafkaProducerConfig/bootstrapServers``, then pass the configuration to a ``KafkaProducer`` initializer. Additional properties mirror librdkafka's producer configuration surface; consult individual property documentation for defaults and semantics. +Provide broker addresses through ``KafkaProducerConfig/bootstrapServers``, then pass the configuration to a ``KafkaProducer`` initializer. Additional properties expose the producer's configuration options; consult individual property documentation for defaults and semantics. For security options, see . For an end-to-end example, see . From 281386b5dd02d786ada8388ab38114bde5a328f8 Mon Sep 17 00:00:00 2001 From: ssikka Date: Thu, 16 Jul 2026 11:19:55 -0400 Subject: [PATCH 7/7] Remove sibling-article Topics groups to break DocC curation cycles Each article listed every sibling under a Topics '### Articles' group, which DocC treats as a curation edge in the documentation hierarchy. The mutual references formed cycles (Observability -> HandlingErrors -> Observability, etc.), producing warnings on the Documentation soundness check. The landing page's Essentials group is now the sole curator of articles; inline links in each article's prose already provide cross-navigation. --- Sources/Kafka/Kafka.docc/ConsumingMessages.md | 7 ------- Sources/Kafka/Kafka.docc/HandlingErrors.md | 6 ------ Sources/Kafka/Kafka.docc/Observability.md | 6 ------ Sources/Kafka/Kafka.docc/ProducingMessages.md | 7 ------- Sources/Kafka/Kafka.docc/SecuringConnections.md | 5 ----- 5 files changed, 31 deletions(-) diff --git a/Sources/Kafka/Kafka.docc/ConsumingMessages.md b/Sources/Kafka/Kafka.docc/ConsumingMessages.md index 9bf5f77d..adc2aceb 100644 --- a/Sources/Kafka/Kafka.docc/ConsumingMessages.md +++ b/Sources/Kafka/Kafka.docc/ConsumingMessages.md @@ -194,13 +194,6 @@ Choose an assignment strategy with ``KafkaConsumerConfig/partitionAssignmentStra ## Topics -### Articles - -- -- -- -- - ### Reading records - ``KafkaConsumer/messages`` diff --git a/Sources/Kafka/Kafka.docc/HandlingErrors.md b/Sources/Kafka/Kafka.docc/HandlingErrors.md index eb2e3028..bf313b80 100644 --- a/Sources/Kafka/Kafka.docc/HandlingErrors.md +++ b/Sources/Kafka/Kafka.docc/HandlingErrors.md @@ -49,12 +49,6 @@ func handle(_ error: KafkaError, logger: Logger) { ## Topics -### Articles - -- -- -- - ### Errors - ``KafkaError`` diff --git a/Sources/Kafka/Kafka.docc/Observability.md b/Sources/Kafka/Kafka.docc/Observability.md index d3a59ad3..2200dd02 100644 --- a/Sources/Kafka/Kafka.docc/Observability.md +++ b/Sources/Kafka/Kafka.docc/Observability.md @@ -60,12 +60,6 @@ Set the `Logger`'s log level to control verbosity — the client logs routine pr ## Topics -### Articles - -- -- -- - ### Metrics - ``KafkaConfiguration/ConsumerMetrics`` diff --git a/Sources/Kafka/Kafka.docc/ProducingMessages.md b/Sources/Kafka/Kafka.docc/ProducingMessages.md index 59ef922c..8a91658b 100644 --- a/Sources/Kafka/Kafka.docc/ProducingMessages.md +++ b/Sources/Kafka/Kafka.docc/ProducingMessages.md @@ -99,13 +99,6 @@ Beyond delivery reports, the events sequence emits errors and other broker event ## Topics -### Articles - -- -- -- -- - ### Sending records - ``KafkaProducer/send(_:)`` diff --git a/Sources/Kafka/Kafka.docc/SecuringConnections.md b/Sources/Kafka/Kafka.docc/SecuringConnections.md index f3fda27b..a01b4073 100644 --- a/Sources/Kafka/Kafka.docc/SecuringConnections.md +++ b/Sources/Kafka/Kafka.docc/SecuringConnections.md @@ -58,11 +58,6 @@ config.saslPassword = "password" ## Topics -### Articles - -- -- - ### Configuration - ``KafkaProducerConfig``