Handle disconnect -> close properly in ChoosePartions

CHANGELOG.md

@@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- Fixed race condition in `Producer` between `Send(...)` and `DisposeAsync()` dispose causing an unintended
+  `DivideByZeroException`. It now correctly throws a `ProducerClosedException`
+
 ## [3.3.0] - 2024-06-10
 
 ### Added

src/DotPulsar/Internal/Producer.cs

@@ -170,6 +170,7 @@ private SubProducer CreateSubProducer(string topic, int partition)
         process.Start();
         return producer;
     }
+
     public bool IsFinalState()
         => _state.IsFinalState();
 
@@ -202,9 +203,11 @@ private async ValueTask<int> ChoosePartitions(MessageMetadata metadata, Cancella
     {
         if (_producerCount == 0)
         {
-            _ = await _state.StateChangedFrom(ProducerState.Disconnected, cancellationToken).ConfigureAwait(false);
+            var newState = await _state.StateChangedFrom(ProducerState.Disconnected, cancellationToken).ConfigureAwait(false);
             if (_faultException is not null)
                 throw new ProducerFaultedException(_faultException);
+            if (newState == ProducerState.Closed)
+                throw new ProducerClosedException();
         }
 
         if (_producerCount == 1)