address some review comments

Author: blindspotbounty

Package.swift

@@ -44,7 +44,6 @@ let package = Package(
         .package(url: "https://github.com/apple/swift-nio.git", from: "2.55.0"),
         .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"),
-        .package(url: "https://github.com/ordo-one/package-concurrency-helpers", .upToNextMajor(from: "1.0.0")),
         // The zstd Swift package produces warnings that we cannot resolve:
         // https://github.com/facebook/zstd/issues/3328
         .package(url: "https://github.com/facebook/zstd.git", from: "1.5.0"),
@@ -74,7 +73,6 @@ let package = Package(
             name: "SwiftKafka",
             dependencies: [
                 "Crdkafka",
-                .product(name: "ConcurrencyHelpers", package: "package-concurrency-helpers", moduleAliases: ["ConcurrencyHelpers" : "BlockingCallWrapper"]),
                 .product(name: "NIOCore", package: "swift-nio"),
                 .product(name: "ServiceLifecycle", package: "swift-service-lifecycle"),
                 .product(name: "Logging", package: "swift-log"),

Sources/SwiftKafka/Configuration/KafkaProducerConfiguration.swift

@@ -36,8 +36,6 @@ public struct KafkaProducerConfiguration {
     /// When set to true, the producer will ensure that messages are successfully produced exactly once and in the original produce order. The following configuration properties are adjusted automatically (if not modified by the user) when idempotence is enabled: max.in.flight.requests.per.connection=5 (must be less than or equal to 5), retries=INT32_MAX (must be greater than 0), acks=all, queuing.strategy=fifo. Producer instantation will fail if user-supplied configuration is incompatible.
     /// Default: `false`
     public var enableIdempotence: Bool = false
-    
-    public var transactionalId: String?
 
     /// Producer queue options.
     public var queue: KafkaConfiguration.QueueOptions = .init()
@@ -99,6 +97,8 @@ public struct KafkaProducerConfiguration {
     /// Security protocol to use (plaintext, ssl, sasl_plaintext, sasl_ssl).
     /// Default: `.plaintext`
     public var securityProtocol: KafkaConfiguration.SecurityProtocol = .plaintext
+    
+    internal var transactionalId: String?
 
     public init() {}
 }
@@ -112,6 +112,9 @@ extension KafkaProducerConfiguration {
         resultDict["enable.idempotence"] = String(self.enableIdempotence)
         if let transactionalId {
             resultDict["transactional.id"] = transactionalId
+            resultDict["transaction.timeout.ms"] = "60000"
+            resultDict["message.timeout.ms"] = "60000"
+            
         }
         resultDict["queue.buffering.max.messages"] = String(self.queue.bufferingMaxMessages)
         resultDict["queue.buffering.max.kbytes"] = String(self.queue.bufferingMaxKBytes)
@@ -191,3 +194,38 @@ extension KafkaConfiguration {
         }
     }
 }
+
+// FIXME: should we really duplicate `KafkaProducerConfiguration`
+// FIXME: after public api updated?
+public struct KafkaTransactionalProducerConfiguration {
+    var transactionalId: String
+    var transactionsTimeout: Duration
+    
+    var producerConfiguration: KafkaProducerConfiguration {
+        set {
+            self.producerConfiguration_ = newValue
+        }
+        get {
+            var conf = self.producerConfiguration_
+            conf.transactionalId = self.transactionalId
+            conf.enableIdempotence = true
+            conf.maxInFlightRequestsPerConnection = min(conf.maxInFlightRequestsPerConnection, 5)
+            return conf
+        }
+    }
+    
+    private var producerConfiguration_: KafkaProducerConfiguration = .init()
+    
+    public init(transactionalId: String, transactionsTimeout: Duration = .kafkaUntilEndOfTransactionTimeout, producerConfiguration: KafkaProducerConfiguration = .init()) {
+        self.transactionalId = transactionalId
+        self.transactionsTimeout = transactionsTimeout
+        self.producerConfiguration = producerConfiguration
+    }
+}
+// MARK: - KafkaProducerConfiguration + Hashable
+
+extension KafkaTransactionalProducerConfiguration: Hashable {}
+
+// MARK: - KafkaProducerConfiguration + Sendable
+
+extension KafkaTransactionalProducerConfiguration: Sendable {}

Sources/SwiftKafka/KafkaProducer.swift

@@ -259,42 +259,10 @@ public final class KafkaProducer: Service, Sendable {
             return KafkaProducerMessageID(rawValue: newMessageID)
         }
     }
-    
-    func initTransactions(timeout: Duration = .seconds(5)) async throws {
-        guard config.transactionalId != nil else {
-            throw KafkaError.config(
-                reason: "Could not initialize transactions because transactionalId is not set in config")
-        }
-        let client = try self.stateMachine.withLockedValue { try $0.initTransactions() }
-        try await client.initTransactions(timeout: timeout)
-    }
-    
-    func beginTransaction() throws {
-        let client = try self.stateMachine.withLockedValue { try $0.transactionsClient() }
-        try client.beginTransaction()
-    }
-    
-    func send(
-        offsets: RDKafkaTopicPartitionList,
-        forConsumer consumer: KafkaConsumer,
-        timeout: Duration = .kafkaUntilEndOfTransactionTimeout,
-        attempts: UInt64 = .max
-    ) async throws {
-        let client = try self.stateMachine.withLockedValue { try $0.transactionsClient() }
-        let consumerClient = try consumer.client()
-        try await consumerClient.withKafkaHandlePointer {
-            try await client.send(attempts: attempts, offsets: offsets, forConsumerKafkaHandle: $0, timeout: timeout)
-        }
-    }
-    
-    func abortTransaction(
-        timeout: Duration = .kafkaUntilEndOfTransactionTimeout,
-        attempts: UInt64) async throws {
-        let client = try self.stateMachine.withLockedValue { try $0.transactionsClient() }
-            try await client.abortTransaction(attempts: attempts, timeout: timeout)
+
+    func client() throws -> RDKafkaClient {
+        try self.stateMachine.withLockedValue { try $0.client() }
     }
-    
-    
 }
 
 // MARK: - KafkaProducer + StateMachine
@@ -322,18 +290,6 @@ extension KafkaProducer {
                 source: Producer.Source?,
                 topicHandles: RDKafkaTopicHandles
             )
-            /// The ``KafkaProducer`` has started and is ready to use, transactions were initialized.
-            ///
-            /// - Parameter messageIDCounter:Used to incrementally assign unique IDs to messages.
-            /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
-            /// - Parameter topicHandles: Class containing all topic names with their respective `rd_kafka_topic_t` pointer.
-            case startedWithTransactions(
-                client: RDKafkaClient,
-                messageIDCounter: UInt,
-                source: Producer.Source?,
-                topicHandles: RDKafkaTopicHandles
-            )
             /// Producer is still running but the event asynchronous sequence was terminated.
             /// All incoming events will be dropped.
             ///
@@ -401,7 +357,7 @@ extension KafkaProducer {
             switch self.state {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
-            case .started(let client, _, let source, _), .startedWithTransactions(let client, _, let source, _):
+            case .started(let client, _, let source, _):
                 return .pollAndYield(client: client, source: source)
             case .consumptionStopped(let client):
                 return .pollWithoutYield(client: client)
@@ -444,19 +400,6 @@ extension KafkaProducer {
                     newMessageID: newMessageID,
                     topicHandles: topicHandles
                 )
-            case .startedWithTransactions(let client, let messageIDCounter, let source, let topicHandles):
-                let newMessageID = messageIDCounter + 1
-                self.state = .startedWithTransactions(
-                    client: client,
-                    messageIDCounter: newMessageID,
-                    source: source,
-                    topicHandles: topicHandles
-                )
-                return .send(
-                    client: client,
-                    newMessageID: newMessageID,
-                    topicHandles: topicHandles
-                )
             case .consumptionStopped:
                 throw KafkaError.connectionClosed(reason: "Sequence consuming events was abruptly terminated, producer closed")
             case .finishing:
@@ -482,7 +425,7 @@ extension KafkaProducer {
                 fatalError("\(#function) invoked while still in state \(self.state)")
             case .consumptionStopped:
                 fatalError("messageSequenceTerminated() must not be invoked more than once")
-            case .started(let client, _, let source, _), .startedWithTransactions(let client, _, let source, _):
+            case .started(let client, _, let source, _):
                 self.state = .consumptionStopped(client: client)
                 return .finishSource(source: source)
             case .finishing(let client, let source):
@@ -502,34 +445,28 @@ extension KafkaProducer {
             switch self.state {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
-            case .started(let client, _, let source, _), .startedWithTransactions(let client, _, let source, _):
+            case .started(let client, _, let source, _):
                 self.state = .finishing(client: client, source: source)
             case .consumptionStopped(let client):
                 self.state = .finishing(client: client, source: nil)
             case .finishing, .finished:
                 break
             }
         }
-
-        mutating func initTransactions() throws -> RDKafkaClient {
+        
+        func client() throws -> RDKafkaClient {
             switch self.state {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
-            case .started(let client, let messageIDCounter, let source, let topicHandles):
-                self.state = .startedWithTransactions(client: client, messageIDCounter: messageIDCounter, source: source, topicHandles: topicHandles)
+            case .started(let client, _, _, _):
                 return client
-            case .startedWithTransactions:
-                throw KafkaError.config(reason: "Transactions were already initialized")
-            case .consumptionStopped, .finishing, .finished:
-                throw KafkaError.connectionClosed(reason: "Producer is stopping or finished")
-            }
-        }
-        
-        func transactionsClient() throws -> RDKafkaClient {
-            guard case let .startedWithTransactions(client, _, _, _) = self.state else {
-                throw KafkaError.transactionAborted(reason: "Transactions were not initialized or producer is being stopped")
+            case .consumptionStopped(let client):
+                return client
+            case .finishing(let client, _):
+                return client
+            case .finished:
+                throw KafkaError.connectionClosed(reason: "Client stopped")
             }
-            return client
         }
     }
 }

Sources/SwiftKafka/KafkaTransaction.swift

@@ -0,0 +1,43 @@
+
+
+public final class KafkaTransaction {
+    let client: RDKafkaClient
+    let producer: KafkaProducer
+    let config: KafkaTransactionalProducerConfiguration
+    
+    init(client: RDKafkaClient, producer: KafkaProducer, config: KafkaTransactionalProducerConfiguration) throws {
+        self.client = client
+        self.producer = producer
+        self.config = config
+
+        try client.beginTransaction()
+    }
+    
+    deinit {
+    }
+
+    public func send(
+        offsets: RDKafkaTopicPartitionList,
+        forConsumer consumer: KafkaConsumer,
+        timeout: Duration = .kafkaUntilEndOfTransactionTimeout,
+        attempts: UInt64 = .max
+    ) async throws {
+        let consumerClient = try consumer.client()
+        try await consumerClient.withKafkaHandlePointer {
+            try await client.send(attempts: attempts, offsets: offsets, forConsumerKafkaHandle: $0, timeout: timeout)
+        }
+    }
+    
+    @discardableResult
+    public func send<Key, Value>(_ message: KafkaProducerMessage<Key, Value>) throws -> KafkaProducerMessageID {
+        try self.producer.send(message)
+    }
+    
+    func commit() async throws {
+        try await client.commitTransaction(attempts: .max, timeout: .kafkaUntilEndOfTransactionTimeout)
+    }
+    
+    func abort() async throws {
+        try await client.abortTransaction(attempts: .max, timeout: .kafkaUntilEndOfTransactionTimeout)
+    }
+}

Sources/SwiftKafka/KafkaTransactionalProducer.swift

@@ -0,0 +1,83 @@
+import Logging
+import ServiceLifecycle
+
+public final class KafkaTransactionalProducer: Service, Sendable {
+    let producer: KafkaProducer
+    let config: KafkaTransactionalProducerConfiguration
+    
+    private init(producer: KafkaProducer, config: KafkaTransactionalProducerConfiguration) async throws {
+        self.producer = producer
+        self.config = config
+        let client = try producer.client()
+        try await client.initTransactions(timeout: config.transactionsTimeout)
+    }
+
+    /// Initialize a new ``KafkaTransactionalProducer``.
+    ///
+    /// This creates a producer without listening for events.
+    ///
+    /// - Parameter config: The ``KafkaProducerConfiguration`` for configuring the ``KafkaProducer``.
+    /// - Parameter topicConfig: The ``KafkaTopicConfiguration`` used for newly created topics.
+    /// - Parameter logger: A logger.
+    /// - Returns: The newly created ``KafkaProducer``.
+    /// - Throws: A ``KafkaError`` if initializing the producer failed.
+    public convenience init(
+        config: KafkaTransactionalProducerConfiguration,
+        topicConfig: KafkaTopicConfiguration = KafkaTopicConfiguration(),
+        logger: Logger
+    ) async throws {
+        let producer = try KafkaProducer(config: config.producerConfiguration, topicConfig: topicConfig, logger: logger)
+        try await self.init(producer: producer, config: config)
+    }
+    
+    /// Initialize a new ``KafkaTransactionalProducer`` and a ``KafkaProducerEvents`` asynchronous sequence.
+    ///
+    /// Use the asynchronous sequence to consume events.
+    ///
+    /// - Important: When the asynchronous sequence is deinited the producer will be shutdown and disallow sending more messages.
+    /// Additionally, make sure to consume the asynchronous sequence otherwise the events will be buffered in memory indefinitely.
+    ///
+    /// - Parameter config: The ``KafkaProducerConfiguration`` for configuring the ``KafkaProducer``.
+    /// - Parameter topicConfig: The ``KafkaTopicConfiguration`` used for newly created topics.
+    /// - Parameter logger: A logger.
+    /// - Returns: A tuple containing the created ``KafkaProducer`` and the ``KafkaProducerEvents``
+    /// `AsyncSequence` used for receiving message events.
+    /// - Throws: A ``KafkaError`` if initializing the producer failed.
+    public static func makeTransactionalProducerWithEvents(
+        config: KafkaTransactionalProducerConfiguration,
+        topicConfig: KafkaTopicConfiguration = KafkaTopicConfiguration(),
+        logger: Logger
+    ) async throws -> (KafkaTransactionalProducer, KafkaProducerEvents) {
+        let (producer, events) = try KafkaProducer.makeProducerWithEvents(config: config.producerConfiguration, topicConfig: topicConfig, logger: logger)
+        
+        let transactionalProducer = try await KafkaTransactionalProducer(producer: producer, config: config)
+        
+        return (transactionalProducer, events)
+    }
+
+    //
+    public func withTransaction(_ body: @Sendable (KafkaTransaction) async throws -> Void) async throws {
+        let transaction = try KafkaTransaction(
+            client: try producer.client(),
+            producer: producer,
+            config: config)
+        
+        do { // need to think here a little bit how to abort transaction
+            try await body(transaction)
+            try await transaction.commit()
+        } catch { // FIXME: maybe catch AbortTransaction?
+            do {
+                try await transaction.abort()
+            } catch {
+                // FIXME: that some inconsistent state
+                // should we emit fatalError(..)
+                // or propagate error as exception with isFatal flag?
+            }
+            throw error
+        }
+    }
+    
+    public func run() async throws {
+        try await producer.run()
+    }
+}