Fix rd_kafka_last_error() race in headerless produce path

Motivation:
The headerless produce path called rd_kafka_produce() then
rd_kafka_last_error() to check the result. rd_kafka_last_error()
returns a thread-local error code. In Swift Concurrency, multiple
Tasks share cooperative pool threads, so a concurrent librdkafka
call from another Task can overwrite the error before the first
Task reads it. This can cause silent message loss (failed send
reads another task's success) or spurious errors.

The with-headers path already used rd_kafka_produceva() which
returns errors directly via rd_kafka_error_t*, avoiding the race.

Modifications:
- Remove the if/else branch in produce() that split headerless
  (rd_kafka_produce) from with-headers (rd_kafka_produceva)
- Always use _produceVariadic() -> rd_kafka_produceva() for both
  paths. The helper already handles empty headers correctly.
- Add concurrentSendsDoNotLoseMessages integration test that sends
  100 messages from concurrent tasks on a 4-partition topic

Result:
All produce paths now use rd_kafka_produceva() which returns errors
directly, eliminating the thread-local race condition.

Author: ssikka100

Sources/Kafka/RDKafka/RDKafkaClient.swift

@@ -135,34 +135,18 @@ public final class RDKafkaClient: Sendable {
             topic: message.topic
         ) { topicHandle in
             try Self.withMessageKeyAndValueBuffer(for: message) { keyBuffer, valueBuffer in
-                if message.headers.isEmpty {
-                    // No message headers set, normal produce method can be used.
-                    rd_kafka_produce(
-                        topicHandle,
-                        Int32(message.partition.rawValue),
-                        RD_KAFKA_MSG_F_COPY,
-                        UnsafeMutableRawPointer(mutating: valueBuffer.baseAddress),
-                        valueBuffer.count,
-                        keyBuffer?.baseAddress,
-                        keyBuffer?.count ?? 0,
-                        UnsafeMutableRawPointer(bitPattern: newMessageID)
+                let errorPointer = try Self.withKafkaCHeaders(for: message.headers) { cHeaders in
+                    try self._produceVariadic(
+                        topicHandle: topicHandle,
+                        partition: Int32(message.partition.rawValue),
+                        messageFlags: RD_KAFKA_MSG_F_COPY,
+                        key: keyBuffer,
+                        value: valueBuffer,
+                        opaque: UnsafeMutableRawPointer(bitPattern: newMessageID),
+                        cHeaders: cHeaders
                     )
-                    return rd_kafka_last_error()
-                } else {
-                    let errorPointer = try Self.withKafkaCHeaders(for: message.headers) { cHeaders in
-                        // Setting message headers only works with `rd_kafka_produceva` (variadic arguments).
-                        try self._produceVariadic(
-                            topicHandle: topicHandle,
-                            partition: Int32(message.partition.rawValue),
-                            messageFlags: RD_KAFKA_MSG_F_COPY,
-                            key: keyBuffer,
-                            value: valueBuffer,
-                            opaque: UnsafeMutableRawPointer(bitPattern: newMessageID),
-                            cHeaders: cHeaders
-                        )
-                    }
-                    return rd_kafka_error_code(errorPointer)
                 }
+                return rd_kafka_error_code(errorPointer)
             }
         }
 

Tests/IntegrationTests/KafkaTests.swift

@@ -647,6 +647,78 @@ func withTestTopic(partitions: Int32 = 1, _ body: (_ testTopic: String) async th
         }
     }
 
+    // MARK: - Concurrent Produce Tests
+
+    @Test func concurrentSendsDoNotLoseMessages() async throws {
+        try await withTestTopic(partitions: 4) { testTopic in
+            let (producer, events) = try KafkaProducer.makeProducerWithEvents(
+                config: self.producerConfig,
+                logger: .kafkaTest
+            )
+
+            let serviceGroupConfiguration = ServiceGroupConfiguration(
+                services: [producer],
+                logger: .kafkaTest
+            )
+            let serviceGroup = ServiceGroup(configuration: serviceGroupConfiguration)
+
+            try await withThrowingTaskGroup(of: Void.self) { group in
+                group.addTask {
+                    try await serviceGroup.run()
+                }
+
+                // Drain events to collect delivery reports
+                let acknowledgedCount = ManagedAtomic<Int>(0)
+                group.addTask {
+                    for await event in events {
+                        switch event {
+                        case .deliveryReports(let reports):
+                            for report in reports {
+                                if case .acknowledged = report.status {
+                                    acknowledgedCount.wrappingIncrement(ordering: .relaxed)
+                                }
+                            }
+                        default:
+                            break
+                        }
+                    }
+                }
+
+                // Send 100 messages concurrently from multiple tasks — exercises the
+                // rd_kafka_produceva path under contention. Before the fix, the old
+                // rd_kafka_produce + rd_kafka_last_error() path could return wrong
+                // error codes when tasks shared a cooperative thread pool thread.
+                let messageCount = 100
+                try await withThrowingTaskGroup(of: Void.self) { sendGroup in
+                    for i in 0..<messageCount {
+                        sendGroup.addTask {
+                            let message = KafkaProducerMessage(
+                                topic: testTopic,
+                                key: "key-\(i)",
+                                value: "value-\(i)"
+                            )
+                            try producer.send(message)
+                        }
+                    }
+                    try await sendGroup.waitForAll()
+                }
+
+                // Wait for delivery reports
+                for _ in 0..<100 {
+                    if acknowledgedCount.load(ordering: .relaxed) >= messageCount { break }
+                    try await Task.sleep(for: .milliseconds(100))
+                }
+
+                #expect(
+                    acknowledgedCount.load(ordering: .relaxed) >= messageCount,
+                    "Expected \(messageCount) acknowledged, got \(acknowledgedCount.load(ordering: .relaxed))"
+                )
+
+                await serviceGroup.triggerGracefulShutdown()
+            }
+        }
+    }
+
     // MARK: - storeOffset Tests
 
     @Test func produceAndConsumeWithStoreOffset() async throws {