Description
KafkaClientSaslHandshaker.doEncodeSaslHandshakeRequest() and KafkaClientApiFactory.doEncodeSaslHandshakeRequest() both call guard.reauthorize(traceId, routedId, initialId, ...) when a GuardHandler needs to establish a fresh session for the Kafka client's SASL/OAUTHBEARER authentication (e.g. an OAuth guard performing a token exchange). Neither class ever calls the corresponding guard.deauthorize(sessionId).
For reference-counted guards (e.g. guard-inline's InlineGuardHandler, which increments session.refs on every reauthorize() and only releases the session once refs returns to zero via deauthorize()), this means every Kafka network connection that goes through this path leaks one guard session for the lifetime of the process — the session is never released even after the underlying Kafka connection closes.
Steps to reproduce / where it lives
runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientSaslHandshaker.java — guardSession is assigned from guard.reauthorize(...) in doEncodeSaslHandshakeRequest(); deauthorize() is never called anywhere in the file.
runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java — same pattern in its own doEncodeSaslHandshakeRequest().
For comparison, runtime/binding-mcp's McpClientFactory has the same gap (it also calls guard.reauthorize(...) without a matching deauthorize()), so this may be worth fixing consistently across both bindings rather than only in binding-kafka.
Expected behavior
The guard session established for a Kafka network connection's SASL handshake should be released via guard.deauthorize(sessionId) when that connection closes/cleans up, so reference-counted guard implementations don't accumulate sessions indefinitely.
Context
Found while investigating an unrelated identity/credentials resolution bug in the same SASL handshake code path (see #2293's http.kafka.oneway migration off options.sasl). Filing separately since it's an independent resource-lifecycle issue, not a functional correctness bug — most in-process guards (like guard-inline used for a single static Kafka connection) won't visibly misbehave from one leaked session, but long-running engines with many short-lived Kafka connections through a reauthorizing guard would accumulate sessions over time.
Description
KafkaClientSaslHandshaker.doEncodeSaslHandshakeRequest()andKafkaClientApiFactory.doEncodeSaslHandshakeRequest()both callguard.reauthorize(traceId, routedId, initialId, ...)when aGuardHandlerneeds to establish a fresh session for the Kafka client's SASL/OAUTHBEARER authentication (e.g. an OAuth guard performing a token exchange). Neither class ever calls the correspondingguard.deauthorize(sessionId).For reference-counted guards (e.g.
guard-inline'sInlineGuardHandler, which incrementssession.refson everyreauthorize()and only releases the session oncerefsreturns to zero viadeauthorize()), this means every Kafka network connection that goes through this path leaks one guard session for the lifetime of the process — the session is never released even after the underlying Kafka connection closes.Steps to reproduce / where it lives
runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientSaslHandshaker.java—guardSessionis assigned fromguard.reauthorize(...)indoEncodeSaslHandshakeRequest();deauthorize()is never called anywhere in the file.runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java— same pattern in its owndoEncodeSaslHandshakeRequest().For comparison,
runtime/binding-mcp'sMcpClientFactoryhas the same gap (it also callsguard.reauthorize(...)without a matchingdeauthorize()), so this may be worth fixing consistently across both bindings rather than only inbinding-kafka.Expected behavior
The guard session established for a Kafka network connection's SASL handshake should be released via
guard.deauthorize(sessionId)when that connection closes/cleans up, so reference-counted guard implementations don't accumulate sessions indefinitely.Context
Found while investigating an unrelated identity/credentials resolution bug in the same SASL handshake code path (see #2293's
http.kafka.onewaymigration offoptions.sasl). Filing separately since it's an independent resource-lifecycle issue, not a functional correctness bug — most in-process guards (likeguard-inlineused for a single static Kafka connection) won't visibly misbehave from one leaked session, but long-running engines with many short-lived Kafka connections through a reauthorizing guard would accumulate sessions over time.