Ensure streams are cleaned up on authentication failure to support re…

…connect attempts (#1191)
aklivity · Aug 13, 2024 · 11d82e0 · 11d82e0
1 parent c36b448
commit 11d82e0
Showing 1 changed file with 2 additions and 5 deletions.
diff --git a/...va/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientConnectionPool.java b/...va/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientConnectionPool.java
@@ -1651,6 +1651,7 @@ private void cleanupStreams(
                 streamsByInitialId.remove(s);
             });
             streams.clear();
+            signalerCorrelations.clear();
         }
 
         private void onConnectionSignal(
@@ -1808,11 +1809,7 @@ private void cleanupConnection(
             doConnectionAbort(traceId);
             doConnectionReset(traceId);
 
-            streams.clear();
-            requests.clear();
-            responses.clear();
-            responseAcks.clear();
-            signalerCorrelations.clear();
+            cleanupStreams(traceId);
         }
 
         private void cleanupBudgetCreditorIfNecessary()