This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new ed9898d7d33 KAFKA-20831: Cache SASL principal per authentication
(#22880)
ed9898d7d33 is described below
commit ed9898d7d33480fdaa1087a98527cb09f179f422
Author: Nandini Singhal <[email protected]>
AuthorDate: Thu Jul 30 04:37:51 2026 +0200
KAFKA-20831: Cache SASL principal per authentication (#22880)
## Summary
- Cache the `KafkaPrincipal` built by `SaslServerAuthenticator` for the
lifetime of an authenticated connection.
- Continue to build a new principal after successful SASL
re-authentication, which installs a new authenticator.
- Add regression coverage for repeated principal access and
re-authentication.
## Motivation
`KafkaChannel.principal()` is consulted while request contexts are
constructed. The current SASL implementation creates a new
`SaslAuthenticationContext` and invokes the configured
`KafkaPrincipalBuilder` on every call, even though the authenticated
identity is connection-scoped.
Custom principal builders may perform non-trivial work or have side
effects. Reusing the principal produced for the authenticated session
avoids redundant work and ensures repeated access observes the same
principal instance.
## Compatibility
This change introduces no public API or configuration changes. Initial
authentication behavior is unchanged. During re-authentication Kafka
installs a new `SaslServerAuthenticator`, so the new authenticated
session receives a fresh principal.
## Validation
```text
./gradlew :clients:test --tests
'org.apache.kafka.common.security.authenticator.SaslAuthenticatorTest'
./gradlew spotlessCheck :clients:checkstyleMain :clients:checkstyleTest
:clients:spotbugsMain -x test
```
The test suite covers PLAIN, SCRAM, OAuth bearer, SSL, token
authentication, and SASL re-authentication paths.
Reviewers: Gaurav Narula <[email protected]>, Luke Chen
<[email protected]>
---------
Co-authored-by: Nandini Singhal
<[email protected]>
---
.../authenticator/SaslServerAuthenticator.java | 12 +++++-
.../authenticator/SaslAuthenticatorTest.java | 47 ++++++++++++++++++++++
2 files changed, 57 insertions(+), 2 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
index b84b5dc2abc..bb8fddc671d 100644
---
a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
+++
b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
@@ -139,6 +139,10 @@ public class SaslServerAuthenticator implements
Authenticator {
private AuthenticationException pendingException = null;
private SaslServer saslServer;
private String saslMechanism;
+ // The authenticated principal is connection-scoped. Cache it after the
first build since principal() is invoked
+ // for every request. It is only accessed by the Processor thread, so no
synchronization is needed.
+ // Re-authentication installs a new SaslServerAuthenticator, and therefore
a fresh cache.
+ private KafkaPrincipal principal;
// buffers used in `authenticate`
private Integer saslAuthRequestMaxReceiveSize;
@@ -305,14 +309,18 @@ public class SaslServerAuthenticator implements
Authenticator {
@Override
public KafkaPrincipal principal() {
+ if (principal != null)
+ return principal;
+
Optional<SSLSession> sslSession = transportLayer instanceof
SslTransportLayer ?
Optional.of(((SslTransportLayer) transportLayer).sslSession())
: Optional.empty();
SaslAuthenticationContext context = new
SaslAuthenticationContext(saslServer, securityProtocol,
clientAddress(), listenerName.value(), sslSession);
- KafkaPrincipal principal = principalBuilder.build(context);
+ KafkaPrincipal builtPrincipal = principalBuilder.build(context);
if (ScramMechanism.isScram(saslMechanism) &&
Boolean.parseBoolean((String)
saslServer.getNegotiatedProperty(ScramLoginModule.TOKEN_AUTH_CONFIG))) {
- principal.tokenAuthenticated(true);
+ builtPrincipal.tokenAuthenticated(true);
}
+ principal = builtPrincipal;
return principal;
}
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
index 07f1d3ca690..fbb4f378c89 100644
---
a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
@@ -178,6 +178,7 @@ public class SaslAuthenticatorTest {
saslClientConfigs =
clientCertStores.getTrustingConfig(serverCertStores);
credentialCache = new CredentialCache();
TestLogin.loginCount.set(0);
+ CountingKafkaPrincipalBuilder.buildCount = 0;
}
@AfterEach
@@ -217,6 +218,31 @@ public class SaslAuthenticatorTest {
checkAuthenticationAndReauthentication(securityProtocol, node);
}
+ @Test
+ public void testPrincipalBuiltOncePerAuthentication() throws Exception {
+ String node = "0";
+ time = new MockTime();
+ SecurityProtocol securityProtocol = SecurityProtocol.SASL_PLAINTEXT;
+ configureMechanisms("PLAIN", Collections.singletonList("PLAIN"));
+
saslServerConfigs.put(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG,
+ CountingKafkaPrincipalBuilder.class);
+
+ server = createEchoServer(securityProtocol);
+ createClientConnection(securityProtocol, node);
+ checkClientConnection(node);
+ server.selector().channels().get(0).principal();
+ server.selector().channels().get(0).principal();
+ assertEquals(1, CountingKafkaPrincipalBuilder.buildCount,
+ "Principal should be built only once for repeated access on a
connection");
+
+ time.sleep((long) (CONNECTIONS_MAX_REAUTH_MS_VALUE * 1.1));
+ checkClientConnection(node);
+ server.verifyReauthenticationMetrics(1, 0);
+ server.selector().channels().get(0).principal();
+ assertEquals(2, CountingKafkaPrincipalBuilder.buildCount,
+ "A successful re-authentication should build and cache a new
principal");
+ }
+
/**
* Test SASL/PLAIN with sasl.authentication.max.receive.size config
*/
@@ -2773,4 +2799,25 @@ public class SaslAuthenticatorTest {
return null;
}
}
+
+ public static class CountingKafkaPrincipalBuilder implements
KafkaPrincipalBuilder {
+ private static int buildCount;
+
+ @Override
+ public KafkaPrincipal build(AuthenticationContext context) {
+ SaslAuthenticationContext saslContext =
(SaslAuthenticationContext) context;
+ buildCount++;
+ return new KafkaPrincipal(KafkaPrincipal.USER_TYPE,
saslContext.server().getAuthorizationID());
+ }
+
+ @Override
+ public byte[] serialize(KafkaPrincipal principal) {
+ return new byte[0];
+ }
+
+ @Override
+ public KafkaPrincipal deserialize(byte[] bytes) {
+ return null;
+ }
+ }
}
