Merge remote-tracking branch 'origin/main' into feature/sc-2764/gsoc-support-for-transactions

Author: blindspotbounty

Sources/SwiftKafka/Data/Array+KafkaContiguousBytes.swift

@@ -0,0 +1,15 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+extension Array: KafkaContiguousBytes where Array.Element == UInt8 {}

Sources/SwiftKafka/Data/ByteBuffer+KafkaContiguousBytes.swift

@@ -0,0 +1,23 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+
+extension ByteBuffer: KafkaContiguousBytes {
+    public func withUnsafeBytes<R>(_ body: (UnsafeRawBufferPointer) throws -> R) rethrows -> R {
+        try self.withUnsafeReadableBytes {
+            try body($0)
+        }
+    }
+}

Sources/SwiftKafka/Data/KafkaContiguousBytes.swift

@@ -0,0 +1,30 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+/// Conformance to this protocol gives users a way to provide their own "bag of bytes" types
+/// to be used for serialization of Kafka messages.
+/// It provides a general interface for bytes since the Swift Standard Library currently does not
+/// provide such a protocol.
+///
+/// By conforming your own types to this protocol, you will be able to pass instances of said types
+/// directly to ``KafkaProducerMessage`` as key and value.
+public protocol KafkaContiguousBytes {
+    /// Calls the given closure with the contents of underlying storage.
+    ///
+    /// - note: Calling `withUnsafeBytes` multiple times does not guarantee that
+    ///         the same buffer pointer will be passed in every time.
+    /// - warning: The buffer argument to the body should not be stored or used
+    ///            outside of the lifetime of the call to the closure.
+    func withUnsafeBytes<R>(_ body: (UnsafeRawBufferPointer) throws -> R) rethrows -> R
+}

Sources/SwiftKafka/Data/Never+KafkaContiguousBytes.swift

@@ -0,0 +1,19 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+extension Never: KafkaContiguousBytes {
+    public func withUnsafeBytes<R>(_: (UnsafeRawBufferPointer) throws -> R) rethrows -> R {
+        fatalError("This statement should never be reached")
+    }
+}

Sources/SwiftKafka/Data/String+KafkaContiguousBytes.swift

@@ -0,0 +1,30 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+
+extension String: KafkaContiguousBytes {
+    public func withUnsafeBytes<R>(_ body: (UnsafeRawBufferPointer) throws -> R) rethrows -> R {
+        if let read = try self.utf8.withContiguousStorageIfAvailable({ unsafePointer in
+            // Fast Path
+            let unsafeRawBufferPointer = UnsafeRawBufferPointer(start: unsafePointer.baseAddress, count: self.count)
+            return try body(unsafeRawBufferPointer)
+        }) {
+            return read
+        } else {
+            // Slow path
+            return try ByteBuffer(string: self).withUnsafeBytes(body)
+        }
+    }
+}

Sources/SwiftKafka/KafkaConsumer.swift

@@ -35,6 +35,29 @@ extension KafkaConsumerCloseOnTerminate: NIOAsyncSequenceProducerDelegate {
     }
 }
 
+// MARK: - KafkaConsumerEvents
+
+/// `AsyncSequence` implementation for handling ``KafkaConsumerEvent``s emitted by Kafka.
+public struct KafkaConsumerEvents: Sendable, AsyncSequence {
+    public typealias Element = KafkaConsumerEvent
+    typealias BackPressureStrategy = NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure
+    typealias WrappedSequence = NIOAsyncSequenceProducer<Element, BackPressureStrategy, KafkaConsumerCloseOnTerminate>
+    let wrappedSequence: WrappedSequence
+
+    /// `AsynceIteratorProtocol` implementation for handling ``KafkaConsumerEvent``s emitted by Kafka.
+    public struct AsyncIterator: AsyncIteratorProtocol {
+        var wrappedIterator: WrappedSequence.AsyncIterator
+
+        public mutating func next() async -> Element? {
+            await self.wrappedIterator.next()
+        }
+    }
+
+    public func makeAsyncIterator() -> AsyncIterator {
+        return AsyncIterator(wrappedIterator: self.wrappedSequence.makeAsyncIterator())
+    }
+}
+
 // MARK: - KafkaConsumerMessages
 
 /// `AsyncSequence` implementation for handling messages received from the Kafka cluster (``KafkaConsumerMessage``).
@@ -52,7 +75,7 @@ public struct KafkaConsumerMessages: Sendable, AsyncSequence {
     let wrappedSequence: WrappedSequence
 
     /// `AsynceIteratorProtocol` implementation for handling messages received from the Kafka cluster (``KafkaConsumerMessage``).
-    public struct ConsumerMessagesAsyncIterator: AsyncIteratorProtocol {
+    public struct AsyncIterator: AsyncIteratorProtocol {
         let stateMachine: NIOLockedValueBox<KafkaConsumer.StateMachine>
         var wrappedIterator: WrappedSequence.AsyncIterator?
 
@@ -80,8 +103,8 @@ public struct KafkaConsumerMessages: Sendable, AsyncSequence {
         }
     }
 
-    public func makeAsyncIterator() -> ConsumerMessagesAsyncIterator {
-        return ConsumerMessagesAsyncIterator(
+    public func makeAsyncIterator() -> AsyncIterator {
+        return AsyncIterator(
             stateMachine: self.stateMachine,
             wrappedIterator: self.wrappedSequence.makeAsyncIterator()
         )
@@ -108,28 +131,27 @@ public final class KafkaConsumer: Sendable, Service {
     /// `AsyncSequence` that returns all ``KafkaConsumerMessage`` objects that the consumer receives.
     public let messages: KafkaConsumerMessages
 
+    // Private initializer, use factory method or convenience init to create KafkaConsumer
     /// Initialize a new ``KafkaConsumer``.
     /// To listen to incoming messages, please subscribe to a list of topics using ``subscribe(topics:)``
     /// or assign the consumer to a particular topic + partition pair using ``assign(topic:partition:offset:)``.
-    /// - Parameter config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``.
-    /// - Parameter logger: A logger.
+    ///
+    /// - Parameters:
+    ///     - client: Client used for handling the connection to the Kafka cluster.
+    ///     - stateMachine: The state machine containing the state of the ``KafkaConsumer``.
+    ///     - config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``.
+    ///     - logger: A logger.
     /// - Throws: A ``KafkaError`` if the initialization failed.
-    public init(
+    private init(
+        client: RDKafkaClient,
+        stateMachine: NIOLockedValueBox<StateMachine>,
         config: KafkaConsumerConfiguration,
         logger: Logger
     ) throws {
         self.config = config
+        self.stateMachine = stateMachine
         self.logger = logger
 
-        let client = try RDKafkaClient.makeClient(
-            type: .consumer,
-            configDictionary: config.dictionary,
-            events: [.log, .fetch, .offsetCommit],
-            logger: logger
-        )
-
-        self.stateMachine = NIOLockedValueBox(StateMachine(logger: self.logger))
-
         let sourceAndSequence = NIOThrowingAsyncSequenceProducer.makeSequence(
             elementType: KafkaConsumerMessage.self,
             backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure(),
@@ -160,6 +182,91 @@ public final class KafkaConsumer: Sendable, Service {
         }
     }
 
+    /// Initialize a new ``KafkaConsumer``.
+    ///
+    /// This creates a consumer without that does not listen to any events other than consumer messages.
+    ///
+    /// - Parameters:
+    ///     - config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``.
+    ///     - logger: A logger.
+    /// - Returns: The newly created ``KafkaConsumer``.
+    /// - Throws: A ``KafkaError`` if the initialization failed.
+    public convenience init(
+        config: KafkaConsumerConfiguration,
+        logger: Logger
+    ) throws {
+        let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
+
+        var subscribedEvents: [RDKafkaEvent] = [.log, .fetch]
+        // Only listen to offset commit events when autoCommit is false
+        if config.enableAutoCommit == false {
+            subscribedEvents.append(.offsetCommit)
+        }
+
+        let client = try RDKafkaClient.makeClient(
+            type: .consumer,
+            configDictionary: config.dictionary,
+            events: subscribedEvents,
+            logger: logger
+        )
+
+        try self.init(
+            client: client,
+            stateMachine: stateMachine,
+            config: config,
+            logger: logger
+        )
+    }
+
+    /// Initialize a new ``KafkaConsumer`` and a ``KafkaConsumerEvents`` 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.
+    ///
+    /// - Parameters:
+    ///     - config: The ``KafkaConsumerConfiguration`` for configuring the ``KafkaConsumer``.
+    ///     - logger: A logger.
+    /// - Returns: A tuple containing the created ``KafkaConsumer`` and the ``KafkaConsumerEvents``
+    /// `AsyncSequence` used for receiving message events.
+    /// - Throws: A ``KafkaError`` if the initialization failed.
+    public static func makeConsumerWithEvents(
+        config: KafkaConsumerConfiguration,
+        logger: Logger
+    ) throws -> (KafkaConsumer, KafkaConsumerEvents) {
+        let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
+
+        var subscribedEvents: [RDKafkaEvent] = [.log, .fetch]
+        // Only listen to offset commit events when autoCommit is false
+        if config.enableAutoCommit == false {
+            subscribedEvents.append(.offsetCommit)
+        }
+
+        let client = try RDKafkaClient.makeClient(
+            type: .consumer,
+            configDictionary: config.dictionary,
+            events: subscribedEvents,
+            logger: logger
+        )
+
+        let consumer = try KafkaConsumer(
+            client: client,
+            stateMachine: stateMachine,
+            config: config,
+            logger: logger
+        )
+
+        let sourceAndSequence = NIOAsyncSequenceProducer.makeSequence(
+            elementType: KafkaConsumerEvent.self,
+            backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure(),
+            delegate: KafkaConsumerCloseOnTerminate(stateMachine: stateMachine)
+        )
+
+        let eventsSequence = KafkaConsumerEvents(wrappedSequence: sourceAndSequence.sequence)
+        return (consumer, eventsSequence)
+    }
+
     /// 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.

Sources/SwiftKafka/KafkaConsumerEvent.swift

@@ -0,0 +1,28 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+/// An enumeration representing events that can be received through the ``KafkaConsumerEvents`` asynchronous sequence.
+public enum KafkaConsumerEvent: Sendable, Hashable {
+    /// - Important: Always provide a `default` case when switiching over this `enum`.
+    case DO_NOT_SWITCH_OVER_THIS_EXHAUSITVELY
+
+    internal init(_ event: RDKafkaClient.KafkaEvent) {
+        switch event {
+        case .deliveryReport:
+            fatalError("Cannot cast \(event) to KafkaConsumerEvent")
+        case .consumerMessages:
+            fatalError("Consumer messages should be handled in the KafkaConsumerMessages asynchronous sequence")
+        }
+    }
+}

Sources/SwiftKafka/KafkaError.swift

@@ -64,16 +64,6 @@ public struct KafkaError: Error, CustomStringConvertible {
         )
     }
 
-    static func acknowledgement(
-        reason: String, file: String = #fileID, line: UInt = #line
-    ) -> KafkaError {
-        return KafkaError(
-            backing: .init(
-                code: .acknowledgement, reason: reason, file: file, line: line
-            )
-        )
-    }
-
     static func config(
         reason: String, file: String = #fileID, line: UInt = #line
     ) -> KafkaError {
@@ -184,7 +174,6 @@ extension KafkaError {
     public struct ErrorCode: Hashable, Sendable, CustomStringConvertible {
         fileprivate enum BackingCode {
             case rdKafkaError
-            case acknowledgement
             case config
             case topicConfig
             case connectionClosed
@@ -206,8 +195,6 @@ extension KafkaError {
 
         /// Errors caused by the underlying `librdkafka` library.
         public static let rdKafkaError = ErrorCode(.rdKafkaError)
-        /// A ``KafkaProducerMessage`` could not be acknowledged by the Kafka server.
-        public static let acknowledgement = ErrorCode(.acknowledgement)
         /// There is an error in the Kafka client configuration.
         public static let config = ErrorCode(.config)
         /// There is an error in the Kafka topic configuration.