defer source.finish (#157)

Author: blindspotbounty

Sources/Kafka/KafkaProducer.swift

@@ -238,8 +238,10 @@ public final class KafkaProducer: Service, Sendable {
                     0...Int(Int32.max) ~= self.configuration.flushTimeoutMilliseconds,
                     "Flush timeout outside of valid range \(0...Int32.max)"
                 )
+                defer { // we should finish source indefinetely of exception in client.flush()
+                    source?.finish()
+                }
                 try await client.flush(timeoutMilliseconds: Int32(self.configuration.flushTimeoutMilliseconds))
-                source?.finish()
                 return
             case .terminatePollLoop:
                 return