add approx. transactional api #98

blindspotbounty · 2023-07-24T17:19:07Z

There is an approximate solution for Transactional API for Kafka gsoc.

Some ideas that were expressed in #78 (comment)

There are the following ideas:

We might initTransaction and switch state in machine
Run blocking transactional calls in GlobalQueue
Retry retryable errors
Automatically abort transactions when abort error received

UPD: after review API changed to the following:

To use kafka transactions, it is required to create KafkaTransactionalProducerConfiguration:

let config = KafkaTransactionalProducerConfiguration(transactionalID: "1234")

Similar to KafkaProducer, it is possible to create KafkaTransactionalProducer:

let transactionalProducer = try await KafkaTransactionalProducer.makeTransactionalProducerWithEvents(config: config)

To commit transactions, it is simply required to call withTransaction(...):

try await transactionalProducer.withTransaction { transaction in
    // Produce new messages:
    let newMessage = KafkaProducerMessage(
        topic: "<some topic>",
        value: "<some value>"
    )
    try transaction.send(newMessage)
...
    // commit offsets:
    let partitionlist = RDKafkaTopicPartitionList()
    partitionlist.setOffset(topic: self.uniqueTestTopic, partition: message.partition, offset: Int64(message.offset))
    try await transaction.send(offsets: partitionlist, forConsumer: consumer)
...
}

blindspotbounty · 2023-07-26T10:34:00Z

Sources/SwiftKafka/KafkaError.swift

@@ -162,6 +192,10 @@ extension KafkaError {
            case messageConsumption
            case topicCreation
            case topicDeletion
+            case transactionAborted


I was thinking if we should add subcodes or provide code from Kafka as is..

FranzBusch · 2023-07-28T08:41:32Z

Package.swift

@@ -44,6 +44,7 @@ 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")),


We can't add this dependency.

Sure, I can copy helpers from there.

FranzBusch · 2023-07-28T08:42:15Z

Sources/SwiftKafka/Configuration/KafkaProducerConfiguration.swift

@@ -36,6 +36,8 @@ 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?


We need some comments on this and what it does. If it only makes sense for the transaction producer we should not expose it all in this configuration and maybe have to introduce a new KafkaTransactionalProducerConfiguration

Do you think of some wrapper around producer configuration like the following?

struct KafkaTransactionalProducerConfiguration { var producerConfiguration: KafkaTransactionalProducerConfiguration public var transactionalId: String }

I am not sure that works. Presumably there are some configurations that don't apply to a transactional producer. We probably should introduce a new configuration and just duplicate stuff like we did for the consumer and producer.

Could you suggest which properties do you have in mind, please?

FranzBusch · 2023-07-28T08:45:05Z

Sources/SwiftKafka/KafkaProducer.swift

@@ -261,6 +261,42 @@ public final class KafkaProducer: Service, Sendable {
            return KafkaProducerMessageID(rawValue: newMessageID)
        }
    }
+
+    func initTransactions(timeout: Duration = .seconds(5)) async throws {


I think we definitely want to create a new KafkaTransactionalProducer type so that the APIs stay clear and separated between the normal producer and the transaction one.

Additionally, I think the APIs should look a bit different. I can imagine something like this

public func withTransaction<Result>(_ body: @Sendable (KafkaTransaction) async throws -> Void) async throws -> Result { // init transaction // begin transaction // call out to user code and pass a new `Transaction` struct to them // at the end call abort or finish transaction for them }

That makes perfect sense to have a separate producer for transactions.
We have it at a higher level but that might be done here

…support-for-transactions

blindspotbounty · 2023-07-31T14:28:59Z

Sources/SwiftKafka/Configuration/KafkaProducerConfiguration.swift

@@ -158,6 +117,8 @@ extension KafkaProducerConfiguration: Hashable {}

 extension KafkaProducerConfiguration: Sendable {}

+extension KafkaProducerConfiguration: KafkaProducerSharedProperties {}


I've tried to use at least internal shared protocol to unite mostly all properties in producer/transactional producer instead of copying them

blindspotbounty · 2023-07-31T14:30:50Z

Sources/SwiftKafka/Utilities/BlockingCall.swift

+
+// performs blocking calls outside of cooperative thread pool
+internal func performBlockingCall<T>(queue: DispatchQueue, body: @escaping () -> T) async -> T {
+    await withCheckedContinuation { continuation in


I am not sure that there is a good way to use swift concurrency with blocking calls.
The only thing is to use some separate thread to avoid contract violation

blindspotbounty · 2023-07-31T14:31:39Z

Tests/IntegrationTests/SwiftKafkaTests.swift

@@ -489,4 +494,115 @@ final class SwiftKafkaTests: XCTestCase {
            XCTAssertTrue(acknowledgedMessages.contains(where: { $0.value == ByteBuffer(string: message.value) }))
        }
    }
+
+    func testProduceAndConsumeWithTransaction() async throws {


This is just an example test so far to demonstrate how current API would look like from user perpective

blindspotbounty · 2023-07-31T14:35:41Z

Sources/SwiftKafka/Configuration/KafkaTransactionalProducerConfiguration.swift

+
+    /// 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`
+    internal let enableIdempotence: Bool = true


The difference between KafkaProducerConfiguration and KafkaTransactionalProducerConfiguration is in couple of properties:

enableIdempotence always true for transactions

maxInFlightRequestsPerConnection not greater than 5

transactionsTimeout is set to socket timeout or greater than socket timeout

blindspotbounty · 2023-07-31T14:37:48Z

Sources/SwiftKafka/KafkaProducer.swift

-    /// The configuration object of the producer client.
-    private let config: KafkaProducerConfiguration
+    /// Configured poll interval
+    private let pollInterval: Duration


I was thinking to use private let config: any KafkaProducerSharedProperties but there is only 2 properties that required that might be more lightweight to store

…support-for-transactions

blindspotbounty added 2 commits July 24, 2023 20:14

add approx. transactional api

b42b3f9

add state for initalized transactions

f5bfbe2

blindspotbounty marked this pull request as ready for review July 25, 2023 14:42

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

b2c1619

blindspotbounty commented Jul 26, 2023

View reviewed changes

FranzBusch reviewed Jul 28, 2023

View reviewed changes

blindspotbounty added 3 commits July 28, 2023 13:01

Merge remote-tracking branch 'origin/main' into feature/sc-2764/gsoc-…

4f9520f

…support-for-transactions

address some review comments

7da4901

use shared config with producer

ce702f3

blindspotbounty marked this pull request as draft July 31, 2023 14:08

blindspotbounty commented Jul 31, 2023

View reviewed changes

blindspotbounty requested a review from FranzBusch July 31, 2023 14:45

blindspotbounty force-pushed the feature/sc-2764/gsoc-support-for-transactions branch from 63ffb51 to ce702f3 Compare August 10, 2023 13:47

blindspotbounty added 3 commits August 10, 2023 18:16

Merge remote-tracking branch 'origin/main' into feature/sc-2764/gsoc-…

8118757

…support-for-transactions

remove merge artifact

6501aaa

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

21a3cd7

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add approx. transactional api #98

add approx. transactional api #98

blindspotbounty commented Jul 24, 2023 •

edited

Loading

blindspotbounty Jul 26, 2023

FranzBusch Jul 28, 2023

blindspotbounty Jul 28, 2023

FranzBusch Jul 28, 2023

blindspotbounty Jul 28, 2023

FranzBusch Jul 28, 2023

blindspotbounty Jul 28, 2023

FranzBusch Jul 28, 2023

blindspotbounty Jul 28, 2023

blindspotbounty Jul 31, 2023

blindspotbounty Jul 31, 2023

blindspotbounty Jul 31, 2023

blindspotbounty Jul 31, 2023

blindspotbounty Jul 31, 2023

		@@ -158,6 +117,8 @@ extension KafkaProducerConfiguration: Hashable {}

		extension KafkaProducerConfiguration: Sendable {}

		extension KafkaProducerConfiguration: KafkaProducerSharedProperties {}

add approx. transactional api #98

Are you sure you want to change the base?

add approx. transactional api #98

Conversation

blindspotbounty commented Jul 24, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

blindspotbounty commented Jul 24, 2023 •

edited

Loading