Review Franz

Modifications:

* introduce new `KafkaConsumer.StateMachine.State` `.finishing` to avoid
  retaining `client` in state `.finished`
* rename `KafkaConsumer.shutdownGracefully` to
  `KafkaConsumer.triggerGracefulShutdown`
* add note that `KafkaConsumer.commitSync` does not support `Task`
  cancellation

Author: felixschlegel

Sources/SwiftKafka/KafkaConsumer.swift

@@ -40,10 +40,10 @@ extension ShutdownOnTerminate: NIOAsyncSequenceProducerDelegate {
     }
 
     func didTerminate() {
-        // Duplicate of _shutdownGracefully
+        // Duplicate of _triggerGracefulShutdown
         let action = self.stateMachine.withLockedValue { $0.finish() }
         switch action {
-        case .shutdownGracefullyAndFinishSource(let client, let source):
+        case .triggerGracefulShutdownAndFinishSource(let client, let source):
             source.finish()
 
             do {
@@ -151,7 +151,7 @@ public final class KafkaConsumer {
     }
 
     deinit {
-        self.shutdownGracefully()
+        self.triggerGracefulShutdown()
     }
 
     /// Subscribe to the given list of `topics`.
@@ -226,6 +226,7 @@ public final class KafkaConsumer {
     /// - Parameter message: Last received message that shall be marked as read.
     /// - Throws: A ``KafkaError`` if committing failed.
     /// - Warning: This method fails if the `enable.auto.commit` configuration property is set to `true`.
+    /// - Important: This method does not support `Task` cancellation.
     public func commitSync(_ message: KafkaConsumerMessage) async throws {
         let action = self.stateMachine.withLockedValue { $0.commitSync() }
         switch action {
@@ -244,11 +245,11 @@ public final class KafkaConsumer {
     ///
     /// - Note: Invoking this function is not always needed as the ``KafkaConsumer``
     /// will already shut down when consumption of the ``KafkaConsumerMessages`` has ended.
-    private func shutdownGracefully() {
+    private func triggerGracefulShutdown() {
         let action = self.stateMachine.withLockedValue { $0.finish() }
         switch action {
-        case .shutdownGracefullyAndFinishSource(let client, let source):
-            self._shutdownGracefullyAndFinishSource(
+        case .triggerGracefulShutdownAndFinishSource(let client, let source):
+            self._triggerGracefulShutdownAndFinishSource(
                 client: client,
                 source: source,
                 logger: self.logger
@@ -258,7 +259,7 @@ public final class KafkaConsumer {
         }
     }
 
-    private func _shutdownGracefullyAndFinishSource(
+    private func _triggerGracefulShutdownAndFinishSource(
         client: KafkaClient,
         source: Producer.Source,
         logger: Logger
@@ -307,10 +308,13 @@ extension KafkaConsumer {
                 client: KafkaClient,
                 source: Producer.Source
             )
-            /// The ``KafkaConsumer`` has been closed.
+            /// The ``KafkaConsumer/triggerGracefulShutdown()`` has been invoked.
+            /// We are now in the process of commiting our last state to the broker.
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
-            case finished(client: KafkaClient)
+            case finishing(client: KafkaClient)
+            /// The ``KafkaConsumer`` is closed.
+            case finished
         }
 
         /// The current state of the StateMachine.
@@ -354,20 +358,23 @@ extension KafkaConsumer {
         /// - Returns: The next action to be taken when wanting to poll, or `nil` if there is no action to be taken.
         ///
         /// - Important: This function throws a `fatalError` if called while in the `.initializing` state.
-        func nextPollLoopAction() -> PollLoopAction {
+        mutating func nextPollLoopAction() -> PollLoopAction {
             switch self.state {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
             case .initializing:
                 fatalError("Subscribe to consumer group / assign to topic partition pair before reading messages")
             case .consuming(let client, let source):
                 return .pollForAndYieldMessage(client: client, source: source)
-            case .finished(let client):
+            case .finishing(let client):
                 if client.isConsumerClosed {
+                    self.state = .finished
                     return .terminatePollLoop
                 } else {
                     return .pollUntilClosed(client: client)
                 }
+            case .finished:
+                return .terminatePollLoop
             }
         }
 
@@ -391,7 +398,7 @@ extension KafkaConsumer {
                     source: source
                 )
                 return .setUpConnection(client: client)
-            case .consuming, .finished:
+            case .consuming, .finishing, .finished:
                 fatalError("\(#function) should only be invoked upon initialization of KafkaConsumer")
             }
         }
@@ -420,7 +427,7 @@ extension KafkaConsumer {
                 fatalError("Subscribe to consumer group / assign to topic partition pair before committing offsets")
             case .consuming(let client, _):
                 return .commitSync(client: client)
-            case .finished:
+            case .finishing, .finished:
                 return .throwClosedError
             }
         }
@@ -431,7 +438,7 @@ extension KafkaConsumer {
             ///
             /// - Parameter client: Client used for handling the connection to the Kafka cluster.
             /// - Parameter source: ``NIOAsyncSequenceProducer/Source`` used for yielding new elements.
-            case shutdownGracefullyAndFinishSource(
+            case triggerGracefulShutdownAndFinishSource(
                 client: KafkaClient,
                 source: Producer.Source
             )
@@ -448,12 +455,12 @@ extension KafkaConsumer {
             case .initializing:
                 fatalError("subscribe() / assign() should have been invoked before \(#function)")
             case .consuming(let client, let source):
-                self.state = .finished(client: client)
-                return .shutdownGracefullyAndFinishSource(
+                self.state = .finishing(client: client)
+                return .triggerGracefulShutdownAndFinishSource(
                     client: client,
                     source: source
                 )
-            case .finished:
+            case .finishing, .finished:
                 return nil
             }
         }