Merge branch 'main' into issue-165-nonblocking-poll-task-executor

Author: ktoso

Package.swift

@@ -24,6 +24,12 @@ let rdkafkaExclude = [
     "./librdkafka/src/rdkafka_sasl_win32.c",
     "./librdkafka/src/rdwin32.h",
     "./librdkafka/src/win32_config.h",
+    // Remove dependency on cURL. Disabling `ENABLE_CURL` and `WITH_CURL` does
+    // not appear to prevent processing of the below files, so we have to exclude
+    // them explicitly.
+    "./librdkafka/src/rdkafka_sasl_oauthbearer.c",
+    "./librdkafka/src/rdkafka_sasl_oauthbearer_oidc.c",
+    "./librdkafka/src/rdhttp.c",
 ]
 
 let package = Package(
@@ -72,7 +78,6 @@ let package = Package(
                 .define("_GNU_SOURCE", to: "1"),  // Fix build error for Swift 5.9 onwards
             ],
             linkerSettings: [
-                .linkedLibrary("curl"),
                 .linkedLibrary("sasl2"),
                 .linkedLibrary("z"),  // zlib
             ]

Sources/Crdkafka/custom/config/config.h

@@ -23,7 +23,7 @@
 #define ENABLE_ZSTD 1
 #define ENABLE_SSL 1
 #define ENABLE_GSSAPI 1
-#define ENABLE_CURL 1
+#define ENABLE_CURL 0
 #define ENABLE_DEVEL 0
 #define ENABLE_VALGRIND 0
 #define ENABLE_REFCNT_DEBUG 0
@@ -35,7 +35,6 @@
 #define ENABLE_ZSTD 1
 #define ENABLE_SSL 1
 #define ENABLE_GSSAPI 1
-#define ENABLE_CURL 1
 #define ENABLE_LZ4_EXT 1
 #define WITH_STATIC_LINKING 1
 #define MKL_APP_NAME "librdkafka"
@@ -136,7 +135,7 @@
 // libzstd
 #define WITH_ZSTD 1
 // libcurl
-#define WITH_CURL 1
+#define WITH_CURL 0
 // WITH_HDRHISTOGRAM
 #define WITH_HDRHISTOGRAM 1
 // WITH_SNAPPY
@@ -146,9 +145,9 @@
 // WITH_SASL_SCRAM
 #define WITH_SASL_SCRAM 1
 // WITH_SASL_OAUTHBEARER
-#define WITH_SASL_OAUTHBEARER 1
+#define WITH_SASL_OAUTHBEARER 0
 // WITH_OAUTHBEARER_OIDC
-#define WITH_OAUTHBEARER_OIDC 1
+#define WITH_OAUTHBEARER_OIDC 0
 // regex
 #define HAVE_REGEX 1
 // strndup

Sources/Kafka/Configuration/KafkaConfiguration+Security.swift

@@ -309,7 +309,7 @@ extension KafkaConfiguration {
             }
         }
 
-        public struct OAuthBearerMethod: Sendable, Hashable {
+        struct OAuthBearerMethod: Sendable, Hashable {
             internal enum _OAuthBearerMethod: Sendable, Hashable {
                 case `default`(
                     configuration: String?
@@ -337,7 +337,7 @@ extension KafkaConfiguration {
             ///     For example: `principalClaimName=azp principal=admin scopeClaimName=roles scope=role1,role2 lifeSeconds=600`.
             ///     In addition, SASL extensions can be communicated to the broker via `extension_NAME=value`.
             ///     For example: `principal=admin extension_traceId=123`
-            public static func `default`(configuration: String? = nil) -> OAuthBearerMethod {
+            static func `default`(configuration: String? = nil) -> OAuthBearerMethod {
                 OAuthBearerMethod(_internal: .default(configuration: configuration))
             }
 
@@ -359,7 +359,7 @@ extension KafkaConfiguration {
             ///     - scope: The client uses this to specify the scope of the access request to the broker.
             ///     - extensions: Allow additional information to be provided to the broker.
             ///     Comma-separated list of key=value pairs. E.g., "supportFeatureX=true,organizationId=sales-emea".
-            public static func oidc(
+            static func oidc(
                 configuration: String? = nil,
                 clientID: String,
                 clientSecret: String,
@@ -419,7 +419,8 @@ extension KafkaConfiguration {
         }
 
         /// Use the OAUTHBEARER mechanism.
-        public static func oAuthBearer(method: OAuthBearerMethod) -> SASLMechanism {
+        // This is currently disabled since it requires a curl dependency otherwise.
+        static func oAuthBearer(method: OAuthBearerMethod) -> SASLMechanism {
             SASLMechanism(
                 _internal: .oAuthBearer(method: method)
             )

Sources/Kafka/KafkaAcknowledgedMessage.swift

@@ -27,6 +27,8 @@ public struct KafkaAcknowledgedMessage {
     public var value: ByteBuffer
     /// The offset of the message in its partition.
     public var offset: KafkaOffset
+    /// The headers of the message.
+    public var headers: [KafkaHeader]
 
     /// Initialize ``KafkaAcknowledgedMessage`` from `rd_kafka_message_t` pointer.
     /// - Throws: A ``KafkaAcknowledgedMessageError`` for failed acknowledgements or malformed messages.
@@ -53,7 +55,7 @@ public struct KafkaAcknowledgedMessage {
         self.topic = topic
 
         self.partition = KafkaPartition(rawValue: Int(rdKafkaMessage.partition))
-
+        self.headers = try RDKafkaClient.getHeaders(for: messagePointer)
         if let keyPointer = rdKafkaMessage.key {
             let keyBufferPointer = UnsafeRawBufferPointer(
                 start: keyPointer,

Sources/Kafka/KafkaConsumerMessage.swift

@@ -66,7 +66,7 @@ public struct KafkaConsumerMessage {
 
         self.partition = KafkaPartition(rawValue: Int(rdKafkaMessage.partition))
 
-        self.headers = try Self.getHeaders(for: messagePointer)
+        self.headers = try RDKafkaClient.getHeaders(for: messagePointer)
 
         if let keyPointer = rdKafkaMessage.key {
             let keyBufferPointer = UnsafeRawBufferPointer(
@@ -91,82 +91,3 @@ extension KafkaConsumerMessage: Hashable {}
 // MARK: - KafkaConsumerMessage + Sendable
 
 extension KafkaConsumerMessage: Sendable {}
-
-// MARK: - Helpers
-
-extension KafkaConsumerMessage {
-    /// Extract ``KafkaHeader``s from a `rd_kafka_message_t` pointer.
-    ///
-    /// - Parameters:
-    ///    - for: Pointer to the `rd_kafka_message_t` object to extract the headers from.
-    private static func getHeaders(
-        for messagePointer: UnsafePointer<rd_kafka_message_t>
-    ) throws -> [KafkaHeader] {
-        var result: [KafkaHeader] = []
-        var headers: OpaquePointer?
-
-        var readStatus = rd_kafka_message_headers(messagePointer, &headers)
-
-        if readStatus == RD_KAFKA_RESP_ERR__NOENT {
-            // No Header Entries
-            return result
-        }
-
-        guard readStatus == RD_KAFKA_RESP_ERR_NO_ERROR else {
-            throw KafkaError.rdKafkaError(wrapping: readStatus)
-        }
-
-        guard let headers else {
-            return result
-        }
-
-        let headerCount = rd_kafka_header_cnt(headers)
-        result.reserveCapacity(headerCount)
-
-        var headerIndex = 0
-
-        while readStatus != RD_KAFKA_RESP_ERR__NOENT && headerIndex < headerCount {
-            var headerKeyPointer: UnsafePointer<CChar>?
-            var headerValuePointer: UnsafeRawPointer?
-            var headerValueSize = 0
-
-            readStatus = rd_kafka_header_get_all(
-                headers,
-                headerIndex,
-                &headerKeyPointer,
-                &headerValuePointer,
-                &headerValueSize
-            )
-
-            if readStatus == RD_KAFKA_RESP_ERR__NOENT {
-                // No Header Entries
-                return result
-            }
-
-            guard readStatus == RD_KAFKA_RESP_ERR_NO_ERROR else {
-                throw KafkaError.rdKafkaError(wrapping: readStatus)
-            }
-
-            guard let headerKeyPointer else {
-                fatalError("Found null pointer when reading KafkaConsumerMessage header key")
-            }
-            let headerKey = String(cString: headerKeyPointer)
-
-            var headerValue: ByteBuffer?
-            if let headerValuePointer, headerValueSize > 0 {
-                let headerValueBufferPointer = UnsafeRawBufferPointer(
-                    start: headerValuePointer,
-                    count: headerValueSize
-                )
-                headerValue = ByteBuffer(bytes: headerValueBufferPointer)
-            }
-
-            let newHeader = KafkaHeader(key: headerKey, value: headerValue)
-            result.append(newHeader)
-
-            headerIndex += 1
-        }
-
-        return result
-    }
-}

Sources/Kafka/RDKafka/RDKafkaClient.swift

@@ -15,6 +15,7 @@
 import Crdkafka
 import Dispatch
 import Logging
+import NIOCore
 
 import class Foundation.JSONDecoder
 
@@ -616,4 +617,79 @@ public final class RDKafkaClient: Sendable {
     func withKafkaHandlePointer<T>(_ body: (OpaquePointer) throws -> T) rethrows -> T {
         try body(self.kafkaHandle.pointer)
     }
+
+    /// Extract ``KafkaHeader``s from a `rd_kafka_message_t` pointer.
+    ///
+    /// - Parameters:
+    ///    - for: Pointer to the `rd_kafka_message_t` object to extract the headers from.
+    internal static func getHeaders(
+        for messagePointer: UnsafePointer<rd_kafka_message_t>
+    ) throws -> [KafkaHeader] {
+        var result: [KafkaHeader] = []
+        var headers: OpaquePointer?
+
+        var readStatus = rd_kafka_message_headers(messagePointer, &headers)
+
+        if readStatus == RD_KAFKA_RESP_ERR__NOENT {
+            // No Header Entries
+            return result
+        }
+
+        guard readStatus == RD_KAFKA_RESP_ERR_NO_ERROR else {
+            throw KafkaError.rdKafkaError(wrapping: readStatus)
+        }
+
+        guard let headers else {
+            return result
+        }
+
+        let headerCount = rd_kafka_header_cnt(headers)
+        result.reserveCapacity(headerCount)
+
+        var headerIndex = 0
+
+        while readStatus != RD_KAFKA_RESP_ERR__NOENT && headerIndex < headerCount {
+            var headerKeyPointer: UnsafePointer<CChar>?
+            var headerValuePointer: UnsafeRawPointer?
+            var headerValueSize = 0
+
+            readStatus = rd_kafka_header_get_all(
+                headers,
+                headerIndex,
+                &headerKeyPointer,
+                &headerValuePointer,
+                &headerValueSize
+            )
+
+            if readStatus == RD_KAFKA_RESP_ERR__NOENT {
+                // No Header Entries
+                return result
+            }
+
+            guard readStatus == RD_KAFKA_RESP_ERR_NO_ERROR else {
+                throw KafkaError.rdKafkaError(wrapping: readStatus)
+            }
+
+            guard let headerKeyPointer else {
+                fatalError("Found null pointer when reading KafkaConsumerMessage header key")
+            }
+            let headerKey = String(cString: headerKeyPointer)
+
+            var headerValue: ByteBuffer?
+            if let headerValuePointer, headerValueSize > 0 {
+                let headerValueBufferPointer = UnsafeRawBufferPointer(
+                    start: headerValuePointer,
+                    count: headerValueSize
+                )
+                headerValue = ByteBuffer(bytes: headerValueBufferPointer)
+            }
+
+            let newHeader = KafkaHeader(key: headerKey, value: headerValue)
+            result.append(newHeader)
+
+            headerIndex += 1
+        }
+
+        return result
+    }
 }

Tests/KafkaTests/KafkaProducerTests.swift

@@ -82,8 +82,10 @@ final class KafkaProducerTests: XCTestCase {
             }
 
             let expectedTopic = "test-topic"
+            let headers = [KafkaHeader(key: "some", value: ByteBuffer.init(string: "test"))]
             let message = KafkaProducerMessage(
                 topic: expectedTopic,
+                headers: headers,
                 key: "key",
                 value: "Hello, World!"
             )
@@ -118,6 +120,7 @@ final class KafkaProducerTests: XCTestCase {
             XCTAssertEqual(expectedTopic, receivedMessage.topic)
             XCTAssertEqual(ByteBuffer(string: message.key!), receivedMessage.key)
             XCTAssertEqual(ByteBuffer(string: message.value), receivedMessage.value)
+            XCTAssertEqual(headers, receivedMessage.headers)
 
             // Shutdown the serviceGroup
             await serviceGroup.triggerGracefulShutdown()