Adopt swift-service-lifecycle (#81)

* Adapt `swift-service-lifecycle`

Modifications:

* add `swift-service-lifecycle` aka `ServicLifecycle` dependency to
  `Package.swift`
* add `Sendable` conformance to the following types:
    * `KafkaProducerConfiguration`
    * `KafkaConsumerConfiguration`
    * `KafkaTopicConfiguration`
    * `KafkaPartition`
    * `KafkaClient`
    * `KafkaConsumerMessages`
    * `KafkaConsumerMessage`
    * `RDKafkaConfig.CapturedClosures`
    * `RDKafkaTopicHandles`
* add `@unchecked Sendable` conformance to the following types:
    * `KafkaProducer`
    * `KafkaConsumer`
* add `Service` conformance to the following types:
    * `KafkaProducer`
    * `KafkaConsumer`
* `SwiftKafkaTests`:
    * use `ServiceGroup`s
* `KafkaProducerTests`:
    * use `ServiceGroup`s
    * remove `testFlushQueuedProducerMessages` as it relied on the now
      `private` implementation detail `triggerGracefulShutdown`
    * remove `testProducerNotUsableAfterShutdown` as it relied on the now
      `private` implementation detail `triggerGracefulShutdown`
    * `testNoMemoryLeakAfterShutdown`: remove obsolete wait for timeout
      as flushing in `KafkaProducer` is now non-blocking
* `KafkaConsumer`:
    * remove `triggerGracefulShutdown` on `deinit` -> this is now
      `ServicLifecycle`'s responsibility
* refactor `RDKafkaConfig` to make `RDKafkaConfig.CapturedClosures`
  sendable

* Remove `ShutdownOnTerminate` for `AsyncSequence`s

Motivation:

Exiting the `run()` loop early is an error in `swift-service-lifecycle`.
Therefore we **don't** want to invoke
`KafkaProducer/KafkaConsumer.triggerGracefulShutdown()` when the our
`NIOAsyncSequence`s have stopped being consumed

Modifications:

* remove `Kafka[Producer|Consumer]ShutdownOnTerminate` behaviour
* adjust tests

* Review Franz

Modifications:

* remove `*Service` suffix from `consumerService` and `producerService`
  in tests and README
* remove old comment in `KafkaConsumer`

Author: felixschlegel

Package.swift

@@ -42,6 +42,7 @@ let package = Package(
     ],
     dependencies: [
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.43.1"),
+        .package(url: "https://github.com/swift-server/swift-service-lifecycle.git", from: "2.0.0-alpha.1"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.0.0"),
         // The zstd Swift package produces warnings that we cannot resolve:
         // https://github.com/facebook/zstd/issues/3328
@@ -73,6 +74,7 @@ let package = Package(
             dependencies: [
                 "Crdkafka",
                 .product(name: "NIOCore", package: "swift-nio"),
+                .product(name: "ServiceLifecycle", package: "swift-service-lifecycle"),
                 .product(name: "Logging", package: "swift-log"),
             ]
         ),

README.md

@@ -23,6 +23,10 @@ Finally, add `import SwiftKafka` to your source code.
 
 ## Usage
 
+`SwiftKafka` should be used within a [`Swift Service Lifecycle`](https://github.com/swift-server/swift-service-lifecycle) 
+[`ServiceGroup`](https://swiftpackageindex.com/swift-server/swift-service-lifecycle/main/documentation/servicelifecycle/servicegroup) for proper startup and shutdown handling.
+Both the `KafkaProducer` and the `KafkaConsumer` implement the [`Service`](https://swiftpackageindex.com/swift-server/swift-service-lifecycle/main/documentation/servicelifecycle/service) protocol.
+
 ### Producer API
 
 The `send(_:)` method of `KafkaProducer` returns a message-id that can later be used to identify the corresponding acknowledgement. Acknowledgements are received through the `acknowledgements` [`AsyncSequence`](https://developer.apple.com/documentation/swift/asyncsequence). Each acknowledgement indicates that producing a message was successful or returns an error.
@@ -32,14 +36,19 @@ let config = KafkaProducerConfiguration(bootstrapServers: ["localhost:9092"])
 
 let (producer, acknowledgements) = try KafkaProducer.makeProducerWithAcknowledgements(
     config: config,
-    logger: .kafkaTest // Your logger here
+    logger: logger
 )
 
 await withThrowingTaskGroup(of: Void.self) { group in
 
     // Run Task
     group.addTask {
-        try await producer.run()
+        let serviceGroup = ServiceGroup(
+            services: [producer],
+            configuration: ServiceGroupConfiguration(gracefulShutdownSignals: []),
+            logger: logger
+        )
+        try await serviceGroup.run()
     }
 
     // Task receiving acknowledgements
@@ -54,9 +63,6 @@ await withThrowingTaskGroup(of: Void.self) { group in
         for await acknowledgement in acknowledgements {
             // Check if acknowledgement belongs to the sent message
         }
-
-        // Required
-        producer.triggerGracefulShutdown()
     }
 }
 ```
@@ -76,14 +82,19 @@ let config = KafkaConsumerConfiguration(
 
 let consumer = try KafkaConsumer(
     config: config,
-    logger: .kafkaTest // Your logger here
+    logger: logger
 )
 
 await withThrowingTaskGroup(of: Void.self) { group in
 
     // Run Task
     group.addTask {
-        try await consumer.run()
+        let serviceGroup = ServiceGroup(
+            services: [consumer],
+            configuration: ServiceGroupConfiguration(gracefulShutdownSignals: []),
+            logger: logger
+        )
+        try await serviceGroup.run()
     }
 
     // Task receiving messages
@@ -107,14 +118,19 @@ let config = KafkaConsumerConfiguration(
 
 let consumer = try KafkaConsumer(
     config: config,
-    logger: .kafkaTest // Your logger here
+    logger: logger
 )
 
 await withThrowingTaskGroup(of: Void.self) { group in
 
     // Run Task
     group.addTask {
-        try await consumer.run()
+        let serviceGroup = ServiceGroup(
+            services: [consumer],
+            configuration: ServiceGroupConfiguration(gracefulShutdownSignals: []),
+            logger: logger
+        )
+        try await serviceGroup.run()
     }
 
     // Task receiving messages
@@ -139,14 +155,19 @@ let config = KafkaConsumerConfiguration(
 
 let consumer = try KafkaConsumer(
     config: config,
-    logger: .kafkaTest // Your logger here
+    logger: logger
 )
 
 await withThrowingTaskGroup(of: Void.self) { group in
 
     // Run Task
     group.addTask {
-        try await consumer.run()
+        let serviceGroup = ServiceGroup(
+            services: [consumer],
+            configuration: ServiceGroupConfiguration(gracefulShutdownSignals: []),
+            logger: logger
+        )
+        try await serviceGroup.run()
     }
 
     // Task receiving messages

Sources/SwiftKafka/Configuration/KafkaConsumerConfiguration.swift

@@ -15,7 +15,7 @@
 import Crdkafka
 import struct Foundation.UUID
 
-public struct KafkaConsumerConfiguration: Hashable {
+public struct KafkaConsumerConfiguration {
     // MARK: - SwiftKafka-specific Config properties
 
     /// The time between two consecutive polls.
@@ -468,12 +468,20 @@ public struct KafkaConsumerConfiguration: Hashable {
     }
 }
 
+// MARK: - KafkaConsumerConfiguration + Hashable
+
+extension KafkaConsumerConfiguration: Hashable {}
+
+// MARK: - KafkaConsumerConfiguration + Sendable
+
+extension KafkaConsumerConfiguration: Sendable {}
+
 // MARK: - KafkaSharedConfiguration + Consumer Additions
 
 extension KafkaSharedConfiguration {
     /// A struct representing the different Kafka message consumption strategies.
-    public struct ConsumptionStrategy: Hashable {
-        enum _ConsumptionStrategy: Hashable {
+    public struct ConsumptionStrategy: Sendable, Hashable {
+        enum _ConsumptionStrategy: Sendable, Hashable {
             case partition(topic: String, partition: KafkaPartition, offset: Int)
             case group(groupID: String, topics: [String])
         }

Sources/SwiftKafka/Configuration/KafkaProducerConfiguration.swift

@@ -12,7 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
-public struct KafkaProducerConfiguration: Hashable {
+public struct KafkaProducerConfiguration {
     // MARK: - SwiftKafka-specific Config properties
 
     /// The time between two consecutive polls.
@@ -427,3 +427,11 @@ public struct KafkaProducerConfiguration: Hashable {
         return KafkaSharedConfiguration.SASLMechanism(description: value)
     }
 }
+
+// MARK: - KafkaProducerConfiguration + Hashable
+
+extension KafkaProducerConfiguration: Hashable {}
+
+// MARK: - KafkaProducerConfiguration + Sendable
+
+extension KafkaProducerConfiguration: Sendable {}

Sources/SwiftKafka/Configuration/KafkaTopicConfiguration.swift

@@ -13,7 +13,7 @@
 //===----------------------------------------------------------------------===//
 
 /// Used to configure new topics created by the ``KafkaProducer``.
-public struct KafkaTopicConfiguration: Hashable {
+public struct KafkaTopicConfiguration {
     var dictionary: [String: String] = [:]
 
     /// This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request: 0=Broker does not send any response/ack to client, -1 or all=Broker will block until message is committed by all in sync replicas (ISRs). If there are less than min.insync.replicas (broker configuration) in the ISR set the produce request will fail.
@@ -126,3 +126,11 @@ extension KafkaSharedConfiguration {
         public static let inherit = CompressionCodec(description: "inherit")
     }
 }
+
+// MARK: - KafkaTopicConfiguration + Hashable
+
+extension KafkaTopicConfiguration: Hashable {}
+
+// MARK: - KafkaTopicConfiguration + Sendable
+
+extension KafkaTopicConfiguration: Sendable {}

Sources/SwiftKafka/KafkaClient.swift

@@ -17,7 +17,7 @@ import Logging
 
 /// Base class for ``KafkaProducer`` and ``KafkaConsumer``,
 /// which is used to handle the connection to the Kafka ecosystem.
-final class KafkaClient {
+final class KafkaClient: Sendable {
     // Default size for Strings returned from C API
     static let stringSize = 1024
 

Sources/SwiftKafka/KafkaConsumer.swift

@@ -16,52 +16,15 @@ import Crdkafka
 import Logging
 import NIOConcurrencyHelpers
 import NIOCore
-
-// MARK: - KafkaConsumerShutDownOnTerminate
-
-/// `NIOAsyncSequenceProducerDelegate` that terminates the shuts the consumer down when
-/// `didTerminate()` is invoked.
-internal struct KafkaConsumerShutdownOnTerminate: @unchecked Sendable { // We can do that because our stored propery is protected by a lock
-    let stateMachine: NIOLockedValueBox<KafkaConsumer.StateMachine>
-}
-
-extension KafkaConsumerShutdownOnTerminate: NIOAsyncSequenceProducerDelegate {
-    func produceMore() {
-        // No back pressure
-        return
-    }
-
-    func didTerminate() {
-        // Duplicate of _triggerGracefulShutdown
-        let action = self.stateMachine.withLockedValue { $0.finish() }
-        switch action {
-        case .triggerGracefulShutdownAndFinishSource(let client, let source):
-            source.finish()
-
-            do {
-                try client.consumerClose()
-            } catch {
-                self.stateMachine.withLockedValue {
-                    if let error = error as? KafkaError {
-                        $0.logger.error("Closing KafkaConsumer failed: \(error.description)")
-                    } else {
-                        $0.logger.error("Caught unknown error: \(error)")
-                    }
-                }
-            }
-        case .none:
-            return
-        }
-    }
-}
+import ServiceLifecycle
 
 // MARK: - KafkaConsumerMessages
 
 /// `AsyncSequence` implementation for handling messages received from the Kafka cluster (``KafkaConsumerMessage``).
-public struct KafkaConsumerMessages: AsyncSequence {
+public struct KafkaConsumerMessages: Sendable, AsyncSequence {
     public typealias Element = KafkaConsumerMessage
     typealias BackPressureStrategy = NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure
-    typealias WrappedSequence = NIOAsyncSequenceProducer<Element, BackPressureStrategy, KafkaConsumerShutdownOnTerminate>
+    typealias WrappedSequence = NIOAsyncSequenceProducer<Element, BackPressureStrategy, NoDelegate>
     let wrappedSequence: WrappedSequence
 
     /// `AsynceIteratorProtocol` implementation for handling messages received from the Kafka cluster (``KafkaConsumerMessage``).
@@ -81,14 +44,14 @@ public struct KafkaConsumerMessages: AsyncSequence {
 // MARK: - KafkaConsumer
 
 /// Receive messages from the Kafka cluster.
-public final class KafkaConsumer {
+public final class KafkaConsumer: Sendable, Service {
     typealias Producer = NIOAsyncSequenceProducer<
         KafkaConsumerMessage,
         NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure,
-        KafkaConsumerShutdownOnTerminate
+        NoDelegate
     >
     /// The configuration object of the consumer client.
-    private var config: KafkaConsumerConfiguration
+    private let config: KafkaConsumerConfiguration
     /// A logger.
     private let logger: Logger
     /// State of the `KafkaConsumer`.
@@ -117,7 +80,7 @@ public final class KafkaConsumer {
         let sourceAndSequence = NIOAsyncSequenceProducer.makeSequence(
             elementType: KafkaConsumerMessage.self,
             backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure(),
-            delegate: KafkaConsumerShutdownOnTerminate(stateMachine: self.stateMachine)
+            delegate: NoDelegate()
         )
 
         self.messages = KafkaConsumerMessages(
@@ -143,10 +106,6 @@ public final class KafkaConsumer {
         }
     }
 
-    deinit {
-        self.triggerGracefulShutdown()
-    }
-
     /// Subscribe to the given list of `topics`.
     /// The partition assignment happens automatically using `KafkaConsumer`'s consumer group.
     /// - Parameter topics: An array of topic names to subscribe to.
@@ -189,6 +148,14 @@ public final class KafkaConsumer {
     ///
     /// - Returns: An awaitable task representing the execution of the poll loop.
     public func run() async throws {
+        try await withGracefulShutdownHandler {
+            try await self._run()
+        } onGracefulShutdown: {
+            self.triggerGracefulShutdown()
+        }
+    }
+
+    private func _run() async throws {
         while !Task.isCancelled {
             let nextAction = self.stateMachine.withLockedValue { $0.nextPollLoopAction() }
             switch nextAction {
@@ -275,12 +242,12 @@ public final class KafkaConsumer {
 
 extension KafkaConsumer {
     /// State machine representing the state of the ``KafkaConsumer``.
-    struct StateMachine {
+    struct StateMachine: Sendable {
         /// A logger.
         let logger: Logger
 
         /// The state of the ``StateMachine``.
-        enum State {
+        enum State: @unchecked Sendable { // TODO: remove @unchecked when https://github.com/apple/swift-nio/pull/2459 is available
             /// The state machine has been initialized with init() but is not yet Initialized
             /// using `func initialize()` (required).
             case uninitialized

Sources/SwiftKafka/KafkaConsumerMessage.swift

@@ -16,7 +16,7 @@ import Crdkafka
 import NIOCore
 
 /// A message received from the Kafka cluster.
-public struct KafkaConsumerMessage: Hashable {
+public struct KafkaConsumerMessage {
     /// The topic that the message was received from.
     public var topic: String
     /// The partition that the message was received from.
@@ -72,3 +72,11 @@ public struct KafkaConsumerMessage: Hashable {
         self.offset = Int(rdKafkaMessage.offset)
     }
 }
+
+// MARK: - KafkaConsumerMessage + Hashable
+
+extension KafkaConsumerMessage: Hashable {}
+
+// MARK: - KafkaConsumerMessage + Sendable
+
+extension KafkaConsumerMessage: Sendable {}

Sources/SwiftKafka/KafkaPartition.swift

@@ -15,7 +15,7 @@
 import Crdkafka
 
 /// Type for representing the number of a Kafka Partition.
-public struct KafkaPartition: RawRepresentable, Hashable {
+public struct KafkaPartition: RawRepresentable {
     public var rawValue: Int32
 
     public init(rawValue: Int32) {
@@ -24,3 +24,11 @@ public struct KafkaPartition: RawRepresentable, Hashable {
 
     public static let unassigned = KafkaPartition(rawValue: RD_KAFKA_PARTITION_UA)
 }
+
+// MARK: KafkaPartition + Hashable
+
+extension KafkaPartition: Hashable {}
+
+// MARK: KafkaPartition + Sendable
+
+extension KafkaPartition: Sendable {}