You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/17 01:15:30 UTC
(pulsar) branch master updated: [fix][test] Split and fix flaky SASL authentication first-stage auth cache tests (#25949)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new d51c364806ce [fix][test] Split and fix flaky SASL authentication first-stage auth cache tests (#25949)
d51c364806ce is described below
commit d51c364806ced01846727b318a102932bd6f53b6
Author: Oneby Wang <89...@qq.com>
AuthorDate: Sat Oct 17 09:15:23 2026 +0800
[fix][test] Split and fix flaky SASL authentication first-stage auth cache tests (#25949)
Co-authored-by: Matteo Merli <mm...@apache.org>
---
.../authentication/SaslAuthenticateTest.java | 78 +++++++++++++---------
1 file changed, 46 insertions(+), 32 deletions(-)
diff --git a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java
index 24d895dffca4..ea58cc8d6d4b 100644
--- a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java
+++ b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java
@@ -18,8 +18,10 @@
*/
package org.apache.pulsar.broker.authentication;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
@@ -33,7 +35,6 @@ import jakarta.servlet.http.HttpServletResponse;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
-import java.lang.reflect.Field;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
@@ -65,13 +66,13 @@ import org.apache.pulsar.client.impl.auth.AuthenticationSasl;
import org.apache.pulsar.common.api.AuthData;
import org.apache.pulsar.common.sasl.SaslConstants;
import org.apache.pulsar.common.util.ObjectMapperFactory;
-import org.testng.Assert;
+import org.awaitility.Awaitility;
+import org.mockito.ArgumentCaptor;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import org.testng.collections.CollectionUtils;
@CustomLog
public class SaslAuthenticateTest extends ProducerConsumerBase {
@@ -224,7 +225,7 @@ public class SaslAuthenticateTest extends ProducerConsumerBase {
@Override
protected void cleanup() throws Exception {
FileUtils.deleteQuietly(secretKeyFile);
- Assert.assertFalse(secretKeyFile.exists());
+ assertFalse(secretKeyFile.exists());
super.internalCleanup();
}
@@ -313,50 +314,63 @@ public class SaslAuthenticateTest extends ProducerConsumerBase {
}
@Test
- @SuppressWarnings("unchecked")
- public void testSaslOnlyAuthFirstStage() throws Exception {
+ public void testSaslOnlyAuthFirstStageKeepsInflightContextsBeforeExpiry() throws Exception {
@Cleanup
AuthenticationProviderSasl saslServer = new AuthenticationProviderSasl();
- // The cache expiration time is set to 500ms. Residual auth info should be cleaned up
- conf.setInflightSaslContextExpiryMs(500);
+ conf.setInflightSaslContextExpiryMs(Integer.MAX_VALUE);
saslServer.initialize(AuthenticationProvider.Context.builder().config(conf).build());
HttpServletRequest servletRequest = mock(HttpServletRequest.class);
- doReturn("Init").when(servletRequest).getHeader("State");
- // 10 clients only do one-stage verification, resulting in 10 auth info remaining in memory
+ doReturn(SaslConstants.SASL_STATE_CLIENT_INIT).when(servletRequest).getHeader(SaslConstants.SASL_HEADER_STATE);
for (int i = 0; i < 10; i++) {
- AuthenticationDataProvider dataProvider = authSasl.getAuthData("localhost");
+ AuthenticationDataProvider dataProvider = authSasl.getAuthData(localHostname);
AuthData initData1 = dataProvider.authenticate(AuthData.INIT_AUTH_DATA);
doReturn(Base64.getEncoder().encodeToString(initData1.getBytes())).when(
- servletRequest).getHeader("SASL-Token");
- doReturn(String.valueOf(i)).when(servletRequest).getHeader("SASL-Server-ID");
+ servletRequest).getHeader(SaslConstants.SASL_AUTH_TOKEN);
+ doReturn(String.valueOf(i)).when(servletRequest).getHeader(SaslConstants.SASL_STATE_SERVER);
saslServer.authenticateHttpRequest(servletRequest, mock(HttpServletResponse.class));
}
- Field field = AuthenticationProviderSasl.class.getDeclaredField("authStates");
- field.setAccessible(true);
- Cache<Long, AuthenticationState> cache = (Cache<Long, AuthenticationState>) field.get(saslServer);
+
+ Cache<Long, AuthenticationState> cache = saslServer.getAuthStates();
assertEquals(cache.asMap().size(), 10);
- // Add more auth info into memory
+ }
+
+ @Test
+ public void testSaslOnlyAuthFirstStageExpiresResidualContexts() throws Exception {
+ @Cleanup
+ AuthenticationProviderSasl saslServer = new AuthenticationProviderSasl();
+ conf.setInflightSaslContextExpiryMs(50);
+ saslServer.initialize(AuthenticationProvider.Context.builder().config(conf).build());
+
+ HttpServletRequest servletRequest = mock(HttpServletRequest.class);
+ doReturn(SaslConstants.SASL_STATE_CLIENT_INIT).when(servletRequest).getHeader(SaslConstants.SASL_HEADER_STATE);
+ // The cache is keyed by the server-generated state id, which the broker returns in the
+ // SASL-Server-ID response header, not by the SASL-Server-ID request header.
+ Set<Long> stateIds = new HashSet<>();
for (int i = 0; i < 10; i++) {
- AuthenticationDataProvider dataProvider = authSasl.getAuthData("localhost");
+ AuthenticationDataProvider dataProvider = authSasl.getAuthData(localHostname);
AuthData initData1 = dataProvider.authenticate(AuthData.INIT_AUTH_DATA);
doReturn(Base64.getEncoder().encodeToString(initData1.getBytes())).when(
- servletRequest).getHeader("SASL-Token");
- doReturn(String.valueOf(10 + i)).when(servletRequest).getHeader("SASL-Server-ID");
- saslServer.authenticateHttpRequest(servletRequest, mock(HttpServletResponse.class));
+ servletRequest).getHeader(SaslConstants.SASL_AUTH_TOKEN);
+ doReturn(String.valueOf(i)).when(servletRequest).getHeader(SaslConstants.SASL_STATE_SERVER);
+ HttpServletResponse response = mock(HttpServletResponse.class);
+ saslServer.authenticateHttpRequest(servletRequest, response);
+ ArgumentCaptor<String> stateId = ArgumentCaptor.forClass(String.class);
+ verify(response).setHeader(eq(SaslConstants.SASL_STATE_SERVER), stateId.capture());
+ stateIds.add(Long.parseLong(stateId.getValue()));
}
- long start = System.currentTimeMillis();
- while (true) {
- if (System.currentTimeMillis() - start > 5000) {
- fail();
- }
- cache = (Cache<Long, AuthenticationState>) field.get(saslServer);
- // Residual auth info should be cleaned up
- if (CollectionUtils.hasElements(cache.asMap())) {
- break;
+ assertEquals(stateIds.size(), 10);
+
+ Cache<Long, AuthenticationState> cache = saslServer.getAuthStates();
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ for (long stateId : stateIds) {
+ assertNull(cache.getIfPresent(stateId));
}
- Thread.sleep(5);
- }
+ // Expired entries are physically removed only by Caffeine maintenance, so run it before
+ // checking that no residual context is left in memory.
+ cache.cleanUp();
+ assertTrue(cache.asMap().isEmpty());
+ });
}
@Test