This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new aeb2508a55f Stop basic auth cache manager immediately on service
shutdown (#18052)
aeb2508a55f is described below
commit aeb2508a55f83d8c33cdfd3d76f3de6263b14730
Author: Kashif Faraz <[email protected]>
AuthorDate: Wed Jun 11 19:41:19 2025 +0530
Stop basic auth cache manager immediately on service shutdown (#18052)
Bug:
When a peon is being shutdown,
`CoordinatorPollingBasicAuthorizerCacheManager.stop()` is invoked.
But this doesn't interrupt currently executing poll.
If at this point, the Coordinator happens to be unreachable, the peon gets
stuck in retries for upto 10 minutes
and might eventually be killed off by the Overlord.
Fix:
- Use `shutdownNow` instead of `shutdown` in
`CoordinatorPollingBasicAuthorizerCacheManager.stop()`
- Do not HTTP retry request if it was interrupted
- Do not raise an alert if the request failed due to an
`InterruptedException`
---
...natorPollingBasicAuthenticatorCacheManager.java | 17 +-
...rdinatorPollingBasicAuthorizerCacheManager.java | 36 ++--
...rPollingBasicAuthenticatorCacheManagerTest.java | 110 ++++++++++++
...atorPollingBasicAuthorizerCacheManagerTest.java | 199 +++++++++++++++++++++
4 files changed, 343 insertions(+), 19 deletions(-)
diff --git
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManager.java
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManager.java
index 33450a3efcc..51758b9ec6e 100644
---
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManager.java
+++
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManager.java
@@ -126,6 +126,9 @@ public class
CoordinatorPollingBasicAuthenticatorCacheManager implements BasicAu
}
LOG.debug("Scheduled user cache poll is done");
}
+ catch (InterruptedException e) {
+ LOG.noStackTrace().info(e, "Interrupted while polling
Coordinator for cachedUserMaps.");
+ }
catch (Throwable t) {
LOG.makeAlert(t, "Error occurred while polling for
cachedUserMaps.").emit();
}
@@ -148,7 +151,7 @@ public class
CoordinatorPollingBasicAuthenticatorCacheManager implements BasicAu
}
LOG.info("CoordinatorPollingBasicAuthenticatorCacheManager is stopping.");
- exec.shutdown();
+ exec.shutdownNow();
LOG.info("CoordinatorPollingBasicAuthenticatorCacheManager is stopped.");
}
@@ -188,15 +191,17 @@ public class
CoordinatorPollingBasicAuthenticatorCacheManager implements BasicAu
{
try {
return RetryUtils.retry(
- () -> {
- return tryFetchUserMapFromCoordinator(prefix);
- },
- e -> true,
+ () -> tryFetchUserMapFromCoordinator(prefix),
+ e -> !(e instanceof InterruptedException),
commonCacheConfig.getMaxSyncRetries()
);
}
+ catch (InterruptedException e) {
+ LOG.noStackTrace().info(e, "Interrupted while fetching user map for
authenticator[%s].", prefix);
+ return null;
+ }
catch (Exception e) {
- LOG.makeAlert(e, "Encountered exception while fetching user map for
authenticator [%s]", prefix).emit();
+ LOG.makeAlert(e, "Encountered exception while fetching user map for
authenticator[%s]", prefix).emit();
if (isInit) {
if (commonCacheConfig.getCacheDirectory() != null) {
try {
diff --git
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManager.java
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManager.java
index 3d192e89d0d..4bebc38b210 100644
---
a/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManager.java
+++
b/extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManager.java
@@ -135,6 +135,9 @@ public class CoordinatorPollingBasicAuthorizerCacheManager
implements BasicAutho
}
LOG.debug("Scheduled userMap cache poll is done");
}
+ catch (InterruptedException e) {
+ LOG.noStackTrace().info(e, "Interrupted while polling
Coordinator for cachedUserMaps.");
+ }
catch (Throwable t) {
LOG.makeAlert(t, "Error occurred while polling for
cachedUserMaps.").emit();
}
@@ -161,6 +164,9 @@ public class CoordinatorPollingBasicAuthorizerCacheManager
implements BasicAutho
}
LOG.debug("Scheduled groupMappingMap cache poll is done");
}
+ catch (InterruptedException e) {
+ LOG.noStackTrace().info(e, "Interrupted while polling
Coordinator for cachedGroupMappingMaps.");
+ }
catch (Throwable t) {
LOG.makeAlert(t, "Error occurred while polling for
cachedGroupMappingMaps.").emit();
}
@@ -183,7 +189,7 @@ public class CoordinatorPollingBasicAuthorizerCacheManager
implements BasicAutho
}
LOG.info("CoordinatorPollingBasicAuthorizerCacheManager is stopping.");
- exec.shutdown();
+ exec.shutdownNow();
LOG.info("CoordinatorPollingBasicAuthorizerCacheManager is stopped.");
}
@@ -326,15 +332,17 @@ public class
CoordinatorPollingBasicAuthorizerCacheManager implements BasicAutho
{
try {
return RetryUtils.retry(
- () -> {
- return tryFetchUserMapsFromCoordinator(prefix);
- },
- e -> true,
+ () -> tryFetchUserMapsFromCoordinator(prefix),
+ e -> !(e instanceof InterruptedException),
commonCacheConfig.getMaxSyncRetries()
);
}
+ catch (InterruptedException e) {
+ LOG.noStackTrace().info(e, "Interrupted while fetching user and role map
for authorizer[%s].", prefix);
+ return null;
+ }
catch (Exception e) {
- LOG.makeAlert(e, "Encountered exception while fetching user and role map
for authorizer [%s]", prefix).emit();
+ LOG.makeAlert(e, "Encountered exception while fetching user and role map
for authorizer[%s]", prefix).emit();
if (isInit) {
if (commonCacheConfig.getCacheDirectory() != null) {
try {
@@ -343,7 +351,7 @@ public class CoordinatorPollingBasicAuthorizerCacheManager
implements BasicAutho
}
catch (Exception e2) {
e2.addSuppressed(e);
- LOG.makeAlert(e2, "Encountered exception while loading user-role
map snapshot for authorizer [%s]", prefix)
+ LOG.makeAlert(e2, "Encountered exception while loading user-role
map snapshot for authorizer[%s]", prefix)
.emit();
}
}
@@ -357,15 +365,17 @@ public class
CoordinatorPollingBasicAuthorizerCacheManager implements BasicAutho
{
try {
return RetryUtils.retry(
- () -> {
- return tryFetchGroupMappingMapsFromCoordinator(prefix);
- },
- e -> true,
+ () -> tryFetchGroupMappingMapsFromCoordinator(prefix),
+ e -> !(e instanceof InterruptedException),
commonCacheConfig.getMaxSyncRetries()
);
}
+ catch (InterruptedException e) {
+ LOG.noStackTrace().info(e, "Interrupted while fetching group and role
map for authorizer[%s].", prefix);
+ return null;
+ }
catch (Exception e) {
- LOG.makeAlert(e, "Encountered exception while fetching group and role
map for authorizer [%s]", prefix).emit();
+ LOG.makeAlert(e, "Encountered exception while fetching group and role
map for authorizer[%s]", prefix).emit();
if (isInit) {
if (commonCacheConfig.getCacheDirectory() != null) {
try {
@@ -374,7 +384,7 @@ public class CoordinatorPollingBasicAuthorizerCacheManager
implements BasicAutho
}
catch (Exception e2) {
e2.addSuppressed(e);
- LOG.makeAlert(e2, "Encountered exception while loading group-role
map snapshot for authorizer [%s]", prefix)
+ LOG.makeAlert(e2, "Encountered exception while loading group-role
map snapshot for authorizer[%s]", prefix)
.emit();
}
}
diff --git
a/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManagerTest.java
b/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManagerTest.java
new file mode 100644
index 00000000000..b122c0b5a6e
--- /dev/null
+++
b/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authentication/db/cache/CoordinatorPollingBasicAuthenticatorCacheManagerTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.security.basic.authentication.db.cache;
+
+import com.google.inject.Injector;
+import org.apache.druid.discovery.DruidLeaderClient;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.http.client.Request;
+import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
+import org.apache.druid.java.util.http.client.response.BytesFullResponseHolder;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.apache.druid.security.basic.BasicAuthCommonCacheConfig;
+import org.apache.druid.security.basic.authentication.BasicHTTPAuthenticator;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.server.security.AuthenticatorMapper;
+import org.easymock.EasyMock;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class CoordinatorPollingBasicAuthenticatorCacheManagerTest
+{
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void test_stop_interruptsPollingThread() throws InterruptedException,
IOException
+ {
+ EmittingLogger.registerEmitter(new StubServiceEmitter());
+
+ final BasicHTTPAuthenticator authenticator =
EasyMock.createStrictMock(BasicHTTPAuthenticator.class);
+ final Injector injector = EasyMock.createStrictMock(Injector.class);
+ EasyMock.expect(injector.getInstance(AuthenticatorMapper.class))
+ .andReturn(new AuthenticatorMapper(Map.of("test-basic-auth",
authenticator))).once();
+
+ // Create a mock leader client and request
+ final DruidLeaderClient leaderClient =
EasyMock.createStrictMock(DruidLeaderClient.class);
+ final Request request = EasyMock.createStrictMock(Request.class);
+
+ // Return the first request immediately
+ EasyMock.expect(leaderClient.makeRequest(EasyMock.anyObject(),
EasyMock.anyString()))
+ .andReturn(request).once();
+ EasyMock.expect(
+ leaderClient.go(EasyMock.anyObject(),
EasyMock.anyObject(BytesFullResponseHandler.class))
+ ).andReturn(
+ new BytesFullResponseHolder(null)
+ ).once();
+
+ // Block the second request so that it can be interrupted by stop()
+ final AtomicBoolean isInterrupted = new AtomicBoolean(false);
+
+ EasyMock.expect(leaderClient.makeRequest(EasyMock.anyObject(),
EasyMock.anyString()))
+ .andReturn(request).once();
+ EasyMock.expect(
+ leaderClient.go(EasyMock.anyObject(),
EasyMock.anyObject(BytesFullResponseHandler.class))
+ ).andAnswer(() -> {
+ try {
+ Thread.sleep(10_000);
+ return null;
+ }
+ catch (InterruptedException e) {
+ isInterrupted.set(true);
+ throw e;
+ }
+ }).once();
+
+ EasyMock.replay(injector, leaderClient);
+
+ final int numRetries = 10;
+ final CoordinatorPollingBasicAuthenticatorCacheManager manager = new
CoordinatorPollingBasicAuthenticatorCacheManager(
+ injector,
+ new BasicAuthCommonCacheConfig(0L, 1L,
temporaryFolder.newFolder().getAbsolutePath(), numRetries),
+ TestHelper.JSON_MAPPER,
+ leaderClient
+ );
+
+ // Start the manager and wait for a while to ensure that polling has
started
+ manager.start();
+ Thread.sleep(10);
+
+ // Stop the manager and verify that the polling thread has been interrupted
+ manager.stop();
+ Assert.assertTrue(isInterrupted.get());
+
+ EasyMock.verify(injector, leaderClient);
+ }
+
+}
diff --git
a/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManagerTest.java
b/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManagerTest.java
new file mode 100644
index 00000000000..d02de5e7f70
--- /dev/null
+++
b/extensions-core/druid-basic-security/src/test/java/org/apache/druid/security/basic/authorization/db/cache/CoordinatorPollingBasicAuthorizerCacheManagerTest.java
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.security.basic.authorization.db.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Injector;
+import org.apache.druid.discovery.DruidLeaderClient;
+import org.apache.druid.java.util.common.jackson.JacksonUtils;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.http.client.Request;
+import
org.apache.druid.java.util.http.client.response.BytesFullResponseHandler;
+import org.apache.druid.java.util.http.client.response.BytesFullResponseHolder;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.apache.druid.security.basic.BasicAuthCommonCacheConfig;
+import org.apache.druid.security.basic.authorization.BasicRoleBasedAuthorizer;
+import
org.apache.druid.security.basic.authorization.entity.GroupMappingAndRoleMap;
+import org.apache.druid.security.basic.authorization.entity.UserAndRoleMap;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.easymock.EasyMock;
+import org.easymock.IAnswer;
+import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
+import org.jboss.netty.handler.codec.http.HttpResponse;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.jboss.netty.handler.codec.http.HttpVersion;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class CoordinatorPollingBasicAuthorizerCacheManagerTest
+{
+ private static final ObjectMapper MAPPER = TestHelper.JSON_MAPPER;
+
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ // Mocks
+ private Request request;
+ private Injector injector;
+ private DruidLeaderClient leaderClient;
+
+ private CoordinatorPollingBasicAuthorizerCacheManager manager;
+
+ @Before
+ public void setup() throws IOException
+ {
+ EmittingLogger.registerEmitter(new StubServiceEmitter());
+
+ final BasicRoleBasedAuthorizer authorizer =
EasyMock.createStrictMock(BasicRoleBasedAuthorizer.class);
+ injector = EasyMock.createStrictMock(Injector.class);
+ EasyMock.expect(injector.getInstance(AuthorizerMapper.class))
+ .andReturn(new AuthorizerMapper(Map.of("test-basic-auth",
authorizer))).once();
+
+ request = EasyMock.createStrictMock(Request.class);
+ leaderClient = EasyMock.createStrictMock(DruidLeaderClient.class);
+
+ final int numRetries = 10;
+ manager = new CoordinatorPollingBasicAuthorizerCacheManager(
+ injector,
+ new BasicAuthCommonCacheConfig(0L, 1L,
temporaryFolder.newFolder().getAbsolutePath(), numRetries),
+ MAPPER,
+ leaderClient
+ );
+ }
+
+ private void replayAll()
+ {
+ EasyMock.replay(injector, leaderClient);
+ }
+
+ private void verifyAll()
+ {
+ EasyMock.verify(injector, leaderClient);
+ }
+
+ @Test
+ public void test_stop_interruptsPollingThread_whileFetchingUserRoleMap()
throws InterruptedException
+ {
+ final HttpResponse response = new
DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+ final BytesFullResponseHolder userResponseHolder = new
BytesFullResponseHolder(response);
+ userResponseHolder.addChunk(JacksonUtils.toBytes(MAPPER, new
UserAndRoleMap(Map.of(), Map.of())));
+
+ final BytesFullResponseHolder groupResponseHolder = new
BytesFullResponseHolder(response);
+ groupResponseHolder.addChunk(JacksonUtils.toBytes(MAPPER, new
GroupMappingAndRoleMap(Map.of(), Map.of())));
+
+ // Return the first set of requests immediately
+ expectHttpRequestAndAnswer(() -> userResponseHolder);
+ expectHttpRequestAndAnswer(() -> groupResponseHolder);
+
+ // Block the second user request so that it can be interrupted by stop()
+ final AtomicBoolean isInterrupted = new AtomicBoolean(false);
+ expectHttpRequestAndAnswer(() -> {
+ try {
+ Thread.sleep(10_000);
+ return userResponseHolder;
+ }
+ catch (InterruptedException e) {
+ isInterrupted.set(true);
+ throw e;
+ }
+ });
+
+ replayAll();
+
+ // Start the manager and wait for a while to ensure that polling has
started
+ manager.start();
+ Thread.sleep(10);
+
+ // Stop the manager and verify that the polling thread has been interrupted
+ manager.stop();
+ Assert.assertTrue(isInterrupted.get());
+
+ verifyAll();
+ }
+
+ @Test
+ public void test_stop_interruptsPollingThread_whileFetchingGroupRoleMap()
throws InterruptedException
+ {
+ final HttpResponse response = new
DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+ final BytesFullResponseHolder userResponseHolder = new
BytesFullResponseHolder(response);
+ userResponseHolder.addChunk(JacksonUtils.toBytes(MAPPER, new
UserAndRoleMap(Map.of(), Map.of())));
+
+ final BytesFullResponseHolder groupResponseHolder = new
BytesFullResponseHolder(response);
+ groupResponseHolder.addChunk(JacksonUtils.toBytes(MAPPER, new
GroupMappingAndRoleMap(Map.of(), Map.of())));
+
+ // Return the first set of requests immediately
+ expectHttpRequestAndAnswer(() -> userResponseHolder);
+ expectHttpRequestAndAnswer(() -> groupResponseHolder);
+
+ // Return the second user request immediately
+ expectHttpRequestAndAnswer(() -> userResponseHolder);
+
+ // Block the second group request so that it can be interrupted by stop()
+ final AtomicBoolean isInterrupted = new AtomicBoolean(false);
+ expectHttpRequestAndAnswer(() -> {
+ try {
+ Thread.sleep(10_000);
+ return groupResponseHolder;
+ }
+ catch (InterruptedException e) {
+ isInterrupted.set(true);
+ throw e;
+ }
+ });
+
+ replayAll();
+
+ // Start the manager and wait for a while to ensure that polling has
started
+ manager.start();
+ Thread.sleep(10);
+
+ // Stop the manager and verify that the polling thread has been interrupted
+ manager.stop();
+ Assert.assertTrue(isInterrupted.get());
+
+ verifyAll();
+ }
+
+ private void expectHttpRequestAndAnswer(IAnswer<BytesFullResponseHolder>
responseHolder)
+ {
+ try {
+ EasyMock.expect(
+ leaderClient.makeRequest(EasyMock.anyObject(), EasyMock.anyString())
+ ).andReturn(request).once();
+ EasyMock.expect(
+ leaderClient.go(
+ EasyMock.anyObject(),
+ EasyMock.anyObject(BytesFullResponseHandler.class)
+ )
+ ).andAnswer(responseHolder).once();
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
