fix: dispose deferred scale-down connection during DisposeAsync

When DisposeAsync runs while _deferredScaleDownConnection is non-null,
the connection was leaked. This happens when a shrink task completes and
the connection is deferred (waiting for in-flight requests to drain for
partition affinity), but DisposeAsync is called before the drain finishes.
The connection is already removed from _pinnedConnections by the shrink
task, so it isn't reached by any other disposal path.

Author: thomhurst

src/Dekaf/Producer/BrokerSender.cs

@@ -2482,6 +2482,17 @@ public async ValueTask DisposeAsync()
             _drainingConnection = null;
         }
 
+        // Dispose deferred scale-down connection if present. This connection was removed from
+        // _pinnedConnections by the shrink task but hasn't been applied via ApplyScaleDown yet
+        // (waiting for in-flight requests to drain). It's not tracked anywhere else, so we must
+        // dispose it explicitly here to avoid a resource leak.
+        if (_deferredScaleDownConnection is not null)
+        {
+            try { await _deferredScaleDownConnection.DisposeAsync().ConfigureAwait(false); }
+            catch (Exception ex) { LogBatchCleanupStepFailed(ex, _brokerId); }
+            _deferredScaleDownConnection = null;
+        }
+
         var totalPending = _totalPendingResponseCount;
         if (totalPending > 0)
         {