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

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``

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.
+
+## 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 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``.

Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md

@@ -0,0 +1,7 @@
+# ``Kafka/KafkaConsumerConfig``
+
+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.

Sources/Kafka/Kafka.docc/KafkaProducerConfig.md

@@ -0,0 +1,7 @@
+# ``Kafka/KafkaProducerConfig``
+
+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.

Sources/Kafka/Kafka.docc/Observability.md

@@ -0,0 +1,66 @@
+# 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.
+
+```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``