KafkaProducer.makeProducerWithEvents() avoid fatalError (#117)

* `KafkaProducer.makeeProducerWithEvents()` avoid `fatalError`

Motivation:

`KafkaProducer.makeeProducerWithEvents()`: initializing the
`NIOAsyncSequenceProducer

Modifications:

* make designated initializer of `KafkaProducer` non-throwing
* `KafkaProducer.makeProducerWithEvents`:
    * initialize `client` before
      calling `NIOAsyncSequenceProducer.makeSequence`
      -> if initializing `client` fails the appropiate error gets thrown
         instead of triggering a fatalError because
         `KafkaProducerCloseOnTerminate` gets triggered in an invalid state

Result:

When the initialization `KafkaProducer.makeProducerWithEvents` fails an
error is thrown instead of triggering the `fatalError` in
`KafkaProducerCloseOnTerminate`.

* Add order comment to KafkaConsumer.makeConsumerWithEvents

Author: felixschlegel

Sources/Kafka/KafkaConsumer.swift

@@ -256,6 +256,11 @@ public final class KafkaConsumer: Sendable, Service {
             logger: logger
         )
 
+        // Note:
+        // It's crucial to initialize the `sourceAndSequence` variable AFTER `client`.
+        // This order is important to prevent the accidental triggering of `KafkaConsumerCloseOnTerminate.didTerminate()`.
+        // If this order is not met and `RDKafkaClient.makeClient()` fails,
+        // it leads to a call to `stateMachine.messageSequenceTerminated()` while it's still in the `.uninitialized` state.
         let sourceAndSequence = NIOAsyncSequenceProducer.makeSequence(
             elementType: KafkaConsumerEvent.self,
             backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure(),

Sources/Kafka/KafkaProducer.swift

@@ -95,7 +95,7 @@ public final class KafkaProducer: Service, Sendable {
         stateMachine: NIOLockedValueBox<KafkaProducer.StateMachine>,
         configuration: KafkaProducerConfiguration,
         topicConfiguration: KafkaTopicConfiguration
-    ) throws {
+    ) {
         self.stateMachine = stateMachine
         self.configuration = configuration
         self.topicConfiguration = topicConfiguration
@@ -130,7 +130,7 @@ public final class KafkaProducer: Service, Sendable {
             )
         }
 
-        try self.init(
+        self.init(
             stateMachine: stateMachine,
             configuration: configuration,
             topicConfiguration: configuration.topicConfiguration
@@ -156,30 +156,34 @@ public final class KafkaProducer: Service, Sendable {
     ) throws -> (KafkaProducer, KafkaProducerEvents) {
         let stateMachine = NIOLockedValueBox(StateMachine(logger: logger))
 
-        let sourceAndSequence = NIOAsyncSequenceProducer.makeSequence(
-            elementType: KafkaProducerEvent.self,
-            backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure(),
-            delegate: KafkaProducerCloseOnTerminate(stateMachine: stateMachine)
-        )
-        let source = sourceAndSequence.source
-
         let client = try RDKafkaClient.makeClient(
             type: .producer,
             configDictionary: configuration.dictionary,
             events: [.log, .deliveryReport],
             logger: logger
         )
 
-        let producer = try KafkaProducer(
+        let producer = KafkaProducer(
             stateMachine: stateMachine,
             configuration: configuration,
             topicConfiguration: configuration.topicConfiguration
         )
 
+        // Note:
+        // It's crucial to initialize the `sourceAndSequence` variable AFTER `client`.
+        // This order is important to prevent the accidental triggering of `KafkaProducerCloseOnTerminate.didTerminate()`.
+        // If this order is not met and `RDKafkaClient.makeClient()` fails,
+        // it leads to a call to `stateMachine.stopConsuming()` while it's still in the `.uninitialized` state.
+        let sourceAndSequence = NIOAsyncSequenceProducer.makeSequence(
+            elementType: KafkaProducerEvent.self,
+            backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.NoBackPressure(),
+            delegate: KafkaProducerCloseOnTerminate(stateMachine: stateMachine)
+        )
+
         stateMachine.withLockedValue {
             $0.initialize(
                 client: client,
-                source: source
+                source: sourceAndSequence.source
             )
         }