Expand DocC: config-type reference, rebalances/events, error handling, observability

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.
+
+The following example creates the consumer with ``KafkaConsumer/makeConsumerWithEvents(config:logger:)`` and iterates 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``

Sources/Kafka/Kafka.docc/HandlingErrors.md

@@ -0,0 +1,62 @@
+# Handling errors
+
+Find where Kafka errors surface and 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.
+
+## Find 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``

Sources/Kafka/Kafka.docc/Kafka.md

@@ -15,6 +15,8 @@ Kafka integrates with [swift-log](https://github.com/apple/swift-log) for struct
 - <doc:GettingStarted>
 - <doc:ProducingMessages>
 - <doc:ConsumingMessages>
+- <doc:HandlingErrors>
+- <doc:Observability>
 - <doc:SecuringConnections>
 
 ### Producing messages

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 enum-valued configuration options such as ``KafkaConfig/SecurityProtocol``, ``KafkaConfig/CompressionCodec``, and ``KafkaConfig/GroupProtocol``. Assign these values to properties on ``KafkaConsumerConfig`` and ``KafkaProducerConfig``.

Sources/Kafka/Kafka.docc/KafkaConsumer.md

@@ -0,0 +1,64 @@
+# ``Kafka/KafkaConsumer``
+
+An object that 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 <doc:ConsumingMessages>.
+
+## 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``

Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md

@@ -0,0 +1,9 @@
+# ``Kafka/KafkaConsumerConfig``
+
+Configuration for 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 expose the consumer's configuration options; consult individual property documentation for defaults and semantics.
+
+For security options, see <doc:SecuringConnections>. For an end-to-end example, see <doc:ConsumingMessages>.

Sources/Kafka/Kafka.docc/KafkaProducer.md

@@ -0,0 +1,39 @@
+# ``Kafka/KafkaProducer``
+
+An object that 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 <doc:ProducingMessages>.
+
+## 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``

Sources/Kafka/Kafka.docc/KafkaProducerConfig.md

@@ -0,0 +1,9 @@
+# ``Kafka/KafkaProducerConfig``
+
+Configuration for a Kafka producer.
+
+## Overview
+
+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 <doc:SecuringConnections>. For an end-to-end example, see <doc:ProducingMessages>.

Sources/Kafka/Kafka.docc/Observability.md

@@ -0,0 +1,68 @@
+# Observing Kafka clients
+
+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 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 following example illustrates configuring a Kafka client with metrics:
+
+```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)
+```
+
+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`.
+
+## 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:
+
+| 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``

Sources/Kafka/Kafka.docc/RDKafkaCode.md

@@ -0,0 +1,39 @@
+# ``Kafka/KafkaError/RDKafkaCode``
+
+An error code reported by the underlying C library that backs the client.
+
+## Overview
+
+When a ``KafkaError`` originates from the underlying C library, its ``KafkaError/rdKafkaCode`` carries one of these codes. Match against the static constants to handle specific failures without importing the C module — for example, telling a connectivity problem (``allBrokersDown``, ``transport``) apart from an authentication problem (``authentication``, ``ssl``).
+
+Each code's ``description`` includes the underlying name and numeric value, which is useful to surface in logs.
+
+## Topics
+
+### Connectivity
+
+- ``allBrokersDown``
+- ``transport``
+- ``timedOut``
+
+### Authentication and security
+
+- ``authentication``
+- ``ssl``
+
+### Producing and consuming
+
+- ``messageTimedOut``
+- ``queueFull``
+- ``maxPollExceeded``
+
+### Topics and partitions
+
+- ``unknownTopic``
+- ``unknownPartition``
+
+### Other
+
+- ``fatal``
+- ``invalidArgument``
+- ``noError``

Sources/Kafka/Kafka.docc/SecuringConnections.md

@@ -55,3 +55,11 @@ 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
+
+### Configuration
+
+- ``KafkaProducerConfig``
+- ``KafkaConsumerConfig``
+- ``KafkaConfig``