From 07879a21b9ed12b0da2f50d3b6bac280ed5345b0 Mon Sep 17 00:00:00 2001 From: Oneby Wang <891734032@qq.com> Date: Sat, 6 Jun 2026 18:14:06 +0800 Subject: [PATCH 1/3] [fix][test] Split SASL first-stage auth cache tests --- .../authentication/SaslAuthenticateTest.java | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) 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 bb8595dacedd8..09f16b52ab863 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 @@ -61,6 +61,7 @@ import org.apache.pulsar.common.api.AuthData; import org.apache.pulsar.common.sasl.SaslConstants; import org.apache.pulsar.common.util.ObjectMapperFactory; +import org.awaitility.Awaitility; import org.testng.Assert; import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; @@ -355,6 +356,59 @@ public void testSaslOnlyAuthFirstStage() throws Exception { } } + @Test + @SuppressWarnings("unchecked") + public void testSaslOnlyAuthFirstStageKeepsInflightContextsBeforeExpiry() throws Exception { + @Cleanup + AuthenticationProviderSasl saslServer = new AuthenticationProviderSasl(); + conf.setInflightSaslContextExpiryMs(Integer.MAX_VALUE); + 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); + for (int i = 0; i < 10; i++) { + AuthenticationDataProvider dataProvider = authSasl.getAuthData(localHostname); + AuthData initData1 = dataProvider.authenticate(AuthData.INIT_AUTH_DATA); + doReturn(Base64.getEncoder().encodeToString(initData1.getBytes())).when( + 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 cache = (Cache) field.get(saslServer); + assertEquals(cache.asMap().size(), 10); + } + + @Test + @SuppressWarnings("unchecked") + 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); + for (int i = 0; i < 10; i++) { + AuthenticationDataProvider dataProvider = authSasl.getAuthData(localHostname); + AuthData initData1 = dataProvider.authenticate(AuthData.INIT_AUTH_DATA); + doReturn(Base64.getEncoder().encodeToString(initData1.getBytes())).when( + 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 cache = (Cache) field.get(saslServer); + Awaitility.await().atMost(3, TimeUnit.SECONDS).pollDelay(100, TimeUnit.MILLISECONDS).untilAsserted(() -> { + cache.cleanUp(); + assertEquals(cache.asMap().size(), 0); + }); + } + @Test @SuppressWarnings("unchecked") public void testMaxInflightContext() throws Exception { From 6c6057a6959835ef39af1facbb090d9004b31453 Mon Sep 17 00:00:00 2001 From: Oneby Wang <891734032@qq.com> Date: Sat, 6 Jun 2026 18:46:28 +0800 Subject: [PATCH 2/3] Delete old test --- .../authentication/SaslAuthenticateTest.java | 48 ------------------- 1 file changed, 48 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 09f16b52ab863..81abc13c9b7e2 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 @@ -68,7 +68,6 @@ 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 { @@ -309,53 +308,6 @@ public void testSaslServerAndClientAuth() throws Exception { log.info().attr("value", methodName).log("---- end"); } - @Test - @SuppressWarnings("unchecked") - public void testSaslOnlyAuthFirstStage() 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); - 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 - for (int i = 0; i < 10; i++) { - AuthenticationDataProvider dataProvider = authSasl.getAuthData("localhost"); - 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"); - saslServer.authenticateHttpRequest(servletRequest, mock(HttpServletResponse.class)); - } - Field field = AuthenticationProviderSasl.class.getDeclaredField("authStates"); - field.setAccessible(true); - Cache cache = (Cache) field.get(saslServer); - assertEquals(cache.asMap().size(), 10); - // Add more auth info into memory - for (int i = 0; i < 10; i++) { - AuthenticationDataProvider dataProvider = authSasl.getAuthData("localhost"); - 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)); - } - long start = System.currentTimeMillis(); - while (true) { - if (System.currentTimeMillis() - start > 5000) { - fail(); - } - cache = (Cache) field.get(saslServer); - // Residual auth info should be cleaned up - if (CollectionUtils.hasElements(cache.asMap())) { - break; - } - Thread.sleep(5); - } - } - @Test @SuppressWarnings("unchecked") public void testSaslOnlyAuthFirstStageKeepsInflightContextsBeforeExpiry() throws Exception { From 986713ed201c1cbd8e9a5e8394362f8af13e6e72 Mon Sep 17 00:00:00 2001 From: Oneby Wang <891734032@qq.com> Date: Mon, 8 Jun 2026 16:56:38 +0800 Subject: [PATCH 3/3] Address pr comment --- .../broker/authentication/SaslAuthenticateTest.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 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 81abc13c9b7e2..4e111eb3aeee4 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 @@ -62,7 +62,6 @@ import org.apache.pulsar.common.sasl.SaslConstants; import org.apache.pulsar.common.util.ObjectMapperFactory; import org.awaitility.Awaitility; -import org.testng.Assert; import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeClass; @@ -220,7 +219,7 @@ protected void setup() throws Exception { @Override protected void cleanup() throws Exception { FileUtils.deleteQuietly(secretKeyFile); - Assert.assertFalse(secretKeyFile.exists()); + assertFalse(secretKeyFile.exists()); super.internalCleanup(); } @@ -355,9 +354,10 @@ public void testSaslOnlyAuthFirstStageExpiresResidualContexts() throws Exception Field field = AuthenticationProviderSasl.class.getDeclaredField("authStates"); field.setAccessible(true); Cache cache = (Cache) field.get(saslServer); - Awaitility.await().atMost(3, TimeUnit.SECONDS).pollDelay(100, TimeUnit.MILLISECONDS).untilAsserted(() -> { - cache.cleanUp(); - assertEquals(cache.asMap().size(), 0); + Awaitility.await().atMost(5, TimeUnit.SECONDS).pollDelay(100, TimeUnit.MILLISECONDS).untilAsserted(() -> { + for (int i = 0; i < 10; i++) { + assertNull(cache.getIfPresent(((long) i))); + } }); }