Fix race conditions in ConnectionPool and SubProducer causing ThreadPool starvation

[eugene-polozhenkov]

Fixes #291

Changes

SubProducer.EstablishNewChannel

The Task.Run lambda captured _dispatcherCts and _channel fields rather than local values. Under thread pool saturation a concurrent EstablishNewChannel call could dispose _dispatcherCts before the lambda read .Token, causing ObjectDisposedException -> fatal producer fault -> recreation death spiral.

Fix: materialize token = cts.Token and pin channel to locals before Task.Run.

ConnectionPool.GetConnection

No serialization on connection establishment: all N producers missing the cache simultaneously called EstablishNewConnection in parallel, spawning N(producers) TCP connections and background tasks.

Fix: SemaphoreSlim(1,1) gate per PulsarUrl with double-checked locking.

ConnectionPool.DisposeConnection

TryRemove(key) removed whichever connection was at the key, not specifically the dead one. Under load the ContinueWith delay could cause it to evict a freshly established replacement.

Fix: TryRemove(KeyValuePair<K,V>) / ICollection.Remove(kvp) for atomic
key+value comparison, with #if NETSTANDARD2_0 || NETSTANDARD2_1 following the existing library pattern.

Tests

• SubProducerTests: unit test for the dispatcher field-capture race
• ConnectionPoolTests: integration tests verifying all producers share a single
  connection after simultaneous reconnect (asserted via broker admin stats)
• IntegrationFixture: CreateAdminClient() moved here from ConsumerTests

Same test as for the #291 with 3 sequential brokers ungraceful restarts, no thread spikes and N of connections stays within expected range.

[blankensteiner]

Very nice, thanks! :-)