Contents

swift-server/swift-kafka-client

The Swift Kafka Client library provides a convenient way to interact with Apache Kafka by leveraging Swift's new concurrency features. This package wraps the native [librdkafka](http

Features

  • Async/await producer with awaitable delivery acknowledgments (sendAndAwait)
  • High-throughput fire-and-forget producer with batched delivery reports
  • AsyncSequence-based consumer with automatic rebalancing
  • At-least-once delivery semantics via manual offset storage
  • Consumer group support with dynamic subscription management
  • Pause/resume partition consumption
  • Typed error handling with retriable and fatal error classification
  • SASL and TLS authentication
  • Fully static Linux builds via the Swift Static Linux (musl) SDK — TLS uses a vendored BoringSSL, so no system OpenSSL is required
  • Integration with swift-service-lifecycle, swift-log, and swift-metrics
  • Full librdkafka configuration exposed as typed Swift properties

Adding Kafka as a dependency

To use the Kafka library in a SwiftPM project, add the following line to the dependencies in your Package.swift file:

.package(url: "https://github.com/swift-server/swift-kafka-client", branch: "main")

Include "Kafka" as a dependency for your executable target:

.target(name: "<target>", dependencies: [
    .product(name: "Kafka", package: "swift-kafka-client"),
]),

Finally, add import Kafka to your source code.

Usage

Run Kafka within a Swift Service Lifecycle ServiceGroup for proper startup and shutdown handling. Both the KafkaProducer and the KafkaConsumer implement the Service protocol.

Producer API

The sendAndAwait(_:) method produces a message and asynchronously awaits broker acknowledgment, confirming the partition and offset where your message landed without blocking any threads:

var config = KafkaProducerConfig()
config.bootstrapServers = ["localhost:9092"]

let logger = Logger(label: "kafka-example")

// Clients read the logger from the task-local `Logger.current`, set with `withLogger(_:_:)`.
try await withLogger(logger) { _ in
    let (producer, _) = try KafkaProducer.makeProducer(config: config)

    let serviceGroup = ServiceGroup(
        services: [producer],
        gracefulShutdownSignals: [.sigterm]
    )

    await withThrowingTaskGroup(of: Void.self) { group in
        group.addTask { try await serviceGroup.run() }

        group.addTask {
            let message = KafkaProducer.Message(topic: "topic-name", value: "Hello, World!")
            let report = try await producer.sendAndAwait(message)
            switch report.status {
            case .acknowledged(let ack):
                print("Delivered to partition \(ack.partition) at offset \(ack.offset)")
            case .failure(let error):
                print("Delivery failed: \(error)")
            }
        }
    }
}

For high-throughput pipelines that need to maximize send rate, use the fire-and-forget send(_:) method and process delivery reports in batches through the events sequence:

var config = KafkaProducerConfig()
config.bootstrapServers = ["localhost:9092"]

let logger = Logger(label: "kafka-example")

try await withLogger(logger) { _ in
    let (producer, events) = try KafkaProducer.makeProducer(config: config)

    let serviceGroup = ServiceGroup(
        services: [producer],
        gracefulShutdownSignals: [.sigterm]
    )

    await withThrowingTaskGroup(of: Void.self) { group in
        group.addTask { try await serviceGroup.run() }

        group.addTask {
            let message = KafkaProducer.Message(topic: "topic-name", value: "Hello, World!")
            try producer.send(message)

            for await event in events {
                switch event {
                case .deliveryReports(let reports):
                    for report in reports {
                        // Handle delivery acknowledgment
                    }
                default:
                    break
                }
            }
        }
    }
}

For simple cases, withProducer(config:_:) runs the producer for the duration of a closure and shuts it down automatically when the closure returns — no ServiceGroup wiring required:

var config = KafkaProducerConfig()
config.bootstrapServers = ["localhost:9092"]

let logger = Logger(label: "kafka-example")

let report = try await withLogger(logger) { _ in
    try await KafkaProducer.withProducer(config: config) { producer, _ in
        let message = KafkaProducer.Message(topic: "topic-name", value: "Hello, World!")
        return try await producer.sendAndAwait(message)
    }
}

To send bytes you already hold in another type, such as Foundation.Data, pass them as a RawSpan instead of building a KafkaProducer.Message:

let payload = Data("Hello, World!".utf8)
let report = try await producer.sendAndAwait(topic: "topic-name", value: payload.bytes)

Consumer API

The consumer delivers messages as an AsyncSequence. Iterate them with a standard for try await loop that integrates naturally with Swift concurrency, structured tasks, and cancellation:

Consumer groups
var config = KafkaConsumerConfig()
config.bootstrapServers = ["localhost:9092"]
config.consumptionStrategy = .group(id: "example-group", topics: ["topic-name"])

let logger = Logger(label: "kafka-example")

try await withLogger(logger) { _ in
    let (consumer, messages, _) = try KafkaConsumer.makeConsumer(config: config)

    let serviceGroup = ServiceGroup(
        services: [consumer],
        gracefulShutdownSignals: [.sigterm]
    )

    await withThrowingTaskGroup(of: Void.self) { group in
        group.addTask { try await serviceGroup.run() }

        group.addTask {
            for try await message in messages {
                print("Received: \(message.topic)/\(message.partition) at offset \(message.offset)")
            }
        }
    }
}
At-least-once processing

For at-least-once delivery semantics, disable automatic offset storage and manually store offsets after processing:

var config = KafkaConsumerConfig()
config.bootstrapServers = ["localhost:9092"]
config.consumptionStrategy = .group(id: "example-group", topics: ["topic-name"])
config.enableAutoOffsetStore = false

let logger = Logger(label: "kafka-example")

try await withLogger(logger) { _ in
    let (consumer, messages, _) = try KafkaConsumer.makeConsumer(config: config)

    let serviceGroup = ServiceGroup(
        services: [consumer],
        gracefulShutdownSignals: [.sigterm]
    )

    await withThrowingTaskGroup(of: Void.self) { group in
        group.addTask { try await serviceGroup.run() }

        group.addTask {
            for try await message in messages {
                // Process message...
                try consumer.storeOffset(message)
                // The background auto-commit timer commits the offset automatically.
            }
        }
    }
}
Manual commits

To control exactly when offsets are committed to the broker:

var config = KafkaConsumerConfig()
config.bootstrapServers = ["localhost:9092"]
config.consumptionStrategy = .group(id: "example-group", topics: ["topic-name"])
config.enableAutoCommit = false

let logger = Logger(label: "kafka-example")

try await withLogger(logger) { _ in
    let (consumer, messages, _) = try KafkaConsumer.makeConsumer(config: config)

    let serviceGroup = ServiceGroup(
        services: [consumer],
        gracefulShutdownSignals: [.sigterm]
    )

    await withThrowingTaskGroup(of: Void.self) { group in
        group.addTask { try await serviceGroup.run() }

        group.addTask {
            for try await message in messages {
                // Process message...
                try await consumer.commit(message)
            }
        }
    }
}

To commit all previously stored offsets at once:

try await consumer.commitStoredOffsets()
Scoped lifetime

withConsumer(config:_:) runs the consumer for the duration of a closure and shuts it down automatically when the closure returns:

var config = KafkaConsumerConfig()
config.bootstrapServers = ["localhost:9092"]
config.consumptionStrategy = .group(id: "example-group", topics: ["topic-name"])

let logger = Logger(label: "kafka-example")

let processed = try await withLogger(logger) { _ in
    try await KafkaConsumer.withConsumer(config: config) { consumer, messages, _ in
        var count = 0
        for try await message in messages {
            // Process message...
            count += 1
            if count >= 10 { break }
        }
        return count
    }
}
Dynamic subscription management

Change topics at runtime:

// Subscribe to additional topics
try consumer.subscribe(topics: ["topic-a", "topic-b"])

// Query current subscription
let topics = try consumer.subscribedTopics

// Unsubscribe from all topics
try consumer.unsubscribe()
Pause and resume

Pause and resume partition consumption temporarily, useful for applying backpressure or performing maintenance without leaving the consumer group:

let partition = KafkaTopicPartition(topic: "topic-name", partition: KafkaPartition(rawValue: 0))
try consumer.pause(topicPartitions: [partition])
// ... later
try consumer.resume(topicPartitions: [partition])

Security mechanisms

Configure both the KafkaProducer and the KafkaConsumer to use different security mechanisms via the securityProtocol property.

Plaintext
var config = KafkaProducerConfig()
config.bootstrapServers = ["localhost:9092"]
config.securityProtocol = .plaintext
TLS (SSL)
var config = KafkaProducerConfig()
config.bootstrapServers = ["localhost:9092"]
config.securityProtocol = .ssl
SASL + plaintext
var config = KafkaProducerConfig()
config.bootstrapServers = ["localhost:9092"]
config.securityProtocol = .sasl_plaintext
config.saslMechanisms = "PLAIN"
config.saslUsername = "user"
config.saslPassword = "password"
SASL + TLS
var config = KafkaProducerConfig()
config.bootstrapServers = ["localhost:9092"]
config.securityProtocol = .sasl_ssl
config.saslMechanisms = "SCRAM-SHA-256"
config.saslUsername = "user"
config.saslPassword = "password"

Metrics

Metrics are auto-registered and enabled by default — the client periodically samples its internal librdkafka statistics and delivery/commit events into swift-metrics instruments under a prefix you choose (default: kafka, every 5 seconds). You don't assign individual instruments; the client owns them and routes their values to whatever backend you bootstrap through MetricsSystem. Customize the prefix/interval, or turn metrics off, via the metrics property:

import Kafka

var config = KafkaConsumerConfig()
config.bootstrapServers = ["localhost:9092"]
config.consumptionStrategy = .group(id: "example-group", topics: ["topic-name"])

// Default is .enabled(prefix: "kafka", updateInterval: .seconds(5)). Customize it:
config.metrics = .enabled(prefix: "orders", updateInterval: .seconds(1))
// ...or disable metrics (and statistics collection) entirely:
// config.metrics = .disabled

let (consumer, _, _) = try KafkaConsumer.makeConsumer(config: config)

Error handling

The events sequence surfaces errors from librdkafka with typed error codes:

// Run the consumer inside a `ServiceGroup` as shown above; errors surface on the events sequence.
let (consumer, _, events) = try KafkaConsumer.makeConsumer(config: config)

for await event in events {
    switch event {
    case .error(let error):
        if error.isFatal {
            // The client is irrecoverable — initiate shutdown.
        } else if error.isRetriable {
            // Transient error — likely resolves on its own.
        }
        print("Error: \(error)")
    default:
        break
    }
}

librdkafka

The Package depends on the librdkafka library, which is included as a git submodule. Its source files are excluded in Package.swift.

Dependencies

The default build is self-contained — TLS is provided by a vendored BoringSSL (via swift-nio-ssl), so no system OpenSSL is required on any platform.

GSSAPI/Kerberos authentication is opt-in and off by default. To enable it, build with SWIFT_KAFKA_ENABLE_GSSAPI=1; on Linux this additionally needs the Cyrus SASL development headers:

  • Linux (GSSAPI only): apt-get install libsasl2-dev

Development setup

Running tests locally

Integration tests require a running Kafka broker. Start one with Docker:

docker run -d -p 9092:9092 apache/kafka:3.9.1

# After starting the container run the tests
swift test

Running tests in Docker

The package provides a Docker environment that automatically starts a local Kafka server and runs the tests:

docker compose -f docker/docker-compose.yaml run client swift test

Alternatively, use a Makefile target:

make docker-test

Specify the Swift compiler version using the SWIFT_VERSION environment variable:

SWIFT_VERSION=6.2 make docker-test

Package Metadata

Repository: swift-server/swift-kafka-client

Default branch: main

README: README.md