diff --git a/Sources/Kafka/Kafka.docc/ConsumingMessages.md b/Sources/Kafka/Kafka.docc/ConsumingMessages.md index a6859690..9bf5f77d 100644 --- a/Sources/Kafka/Kafka.docc/ConsumingMessages.md +++ b/Sources/Kafka/Kafka.docc/ConsumingMessages.md @@ -134,8 +134,73 @@ 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 +### Articles + +- +- +- +- + ### Reading records - ``KafkaConsumer/messages`` @@ -161,6 +226,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`` diff --git a/Sources/Kafka/Kafka.docc/HandlingErrors.md b/Sources/Kafka/Kafka.docc/HandlingErrors.md new file mode 100644 index 00000000..eb2e3028 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/HandlingErrors.md @@ -0,0 +1,68 @@ +# 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 + +### Articles + +- +- +- + +### 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..15ea1524 100644 --- a/Sources/Kafka/Kafka.docc/Kafka.md +++ b/Sources/Kafka/Kafka.docc/Kafka.md @@ -15,6 +15,8 @@ Kafka integrates with [swift-log](https://github.com/apple/swift-log) for struct - - - +- +- - ### Producing messages diff --git a/Sources/Kafka/Kafka.docc/KafkaConfig.md b/Sources/Kafka/Kafka.docc/KafkaConfig.md new file mode 100644 index 00000000..fe4a7eda --- /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 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/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 new file mode 100644 index 00000000..514edd31 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/KafkaConsumerConfig.md @@ -0,0 +1,9 @@ +# ``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 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/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 new file mode 100644 index 00000000..0be4fb7e --- /dev/null +++ b/Sources/Kafka/Kafka.docc/KafkaProducerConfig.md @@ -0,0 +1,9 @@ +# ``Kafka/KafkaProducerConfig`` + +Configures 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 . For an end-to-end example, see . diff --git a/Sources/Kafka/Kafka.docc/Observability.md b/Sources/Kafka/Kafka.docc/Observability.md new file mode 100644 index 00000000..d3a59ad3 --- /dev/null +++ b/Sources/Kafka/Kafka.docc/Observability.md @@ -0,0 +1,72 @@ +# 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 + +### Articles + +- +- +- + +### Metrics + +- ``KafkaConfiguration/ConsumerMetrics`` +- ``KafkaConfiguration/ProducerMetrics`` 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``