This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch fix/drpc-function-queue-eviction in repository https://gitbox.apache.org/repos/asf/storm.git
commit 12f324a362bd77a83b8b73991a49400c3130e626 Author: Richard Zowalla <[email protected]> AuthorDate: Fri Aug 21 14:54:49 2026 +0200 Remove DRPC per-function request queues once they are empty --- .../java/org/apache/storm/daemon/drpc/DRPC.java | 48 ++++++++++++++++------ .../org/apache/storm/daemon/drpc/DRPCTest.java | 30 ++++++++++++++ 2 files changed, 65 insertions(+), 13 deletions(-) diff --git a/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java b/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java index 23183f08e..df9ed364f 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java @@ -28,6 +28,7 @@ import java.util.TimerTask; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import org.apache.storm.DaemonConfig; import org.apache.storm.daemon.StormCommon; import org.apache.storm.generated.AuthorizationException; @@ -145,8 +146,15 @@ public class DRPC implements AutoCloseable { private void cleanup(String id) { OutstandingRequest req = requests.remove(id); - if (req != null && !req.wasFetched()) { - queues.get(req.getFunction()).remove(req); + if (req != null) { + queues.computeIfPresent(req.getFunction(), (function, queue) -> { + if (!req.wasFetched()) { + queue.remove(req); + } + //Drop the queue itself once nothing is waiting in it, otherwise the map keeps an + // entry for every function name a client has ever asked about. + return queue.isEmpty() ? null : queue; + }); } } @@ -165,16 +173,15 @@ public class DRPC implements AutoCloseable { return String.valueOf(ctr.incrementAndGet()); } - private ConcurrentLinkedQueue<OutstandingRequest> getQueue(String function) { + private static void checkFunctionName(String function) { if (function == null) { throw new IllegalArgumentException("The function for a request cannot be null"); } - ConcurrentLinkedQueue<OutstandingRequest> queue = queues.get(function); - if (queue == null) { - queues.putIfAbsent(function, new ConcurrentLinkedQueue<>()); - queue = queues.get(function); - } - return queue; + } + + @VisibleForTesting + int getNumTrackedFunctions() { + return queues.size(); } public void returnResult(String id, String result) throws AuthorizationException { @@ -190,8 +197,17 @@ public class DRPC implements AutoCloseable { public DRPCRequest fetchRequest(String functionName) throws AuthorizationException { meterFetchRequestCalls.mark(); checkAuthorizationNoLog("fetchRequest", functionName); - ConcurrentLinkedQueue<OutstandingRequest> q = getQueue(functionName); - OutstandingRequest req = q.poll(); + checkFunctionName(functionName); + //Never create a queue here. A function name comes from the client, so a queue that no one + // ever puts a request into would stay in the map forever. Poll and drop an emptied queue + // under the same lock execute() adds under, so a request can never be left in a queue that + // was just removed from the map. + AtomicReference<OutstandingRequest> polled = new AtomicReference<>(); + queues.computeIfPresent(functionName, (function, queue) -> { + polled.set(queue.poll()); + return queue.isEmpty() ? null : queue; + }); + OutstandingRequest req = polled.get(); if (req != null) { //Only log accesses that fetched something logAccess("fetchRequest", functionName); @@ -219,12 +235,18 @@ public class DRPC implements AutoCloseable { AuthorizationException { meterExecuteCalls.mark(); checkAuthorization("execute", functionName); + checkFunctionName(functionName); String id = nextId(); LOG.debug("Execute {} {}", functionName, funcArgs); T req = factory.mkRequest(functionName, new DRPCRequest(funcArgs, id)); requests.put(id, req); - ConcurrentLinkedQueue<OutstandingRequest> q = getQueue(functionName); - q.add(req); + queues.compute(functionName, (function, queue) -> { + if (queue == null) { + queue = new ConcurrentLinkedQueue<>(); + } + queue.add(req); + return queue; + }); return req; } diff --git a/storm-server/src/test/java/org/apache/storm/daemon/drpc/DRPCTest.java b/storm-server/src/test/java/org/apache/storm/daemon/drpc/DRPCTest.java index 210aa7246..f37a1b8c3 100644 --- a/storm-server/src/test/java/org/apache/storm/daemon/drpc/DRPCTest.java +++ b/storm-server/src/test/java/org/apache/storm/daemon/drpc/DRPCTest.java @@ -140,6 +140,36 @@ public class DRPCTest { } } + @Test + public void testQueuesAreRemovedWhenEmpty() throws Exception { + try (DRPC server = new DRPC(new StormMetricsRegistry(), null, 1000)) { + //Fetching for a function nothing was ever submitted for must not leave state behind + DRPCRequest nothing = server.fetchRequest("never-registered"); + assertNotNull(nothing); + assertEquals("", nothing.get_request_id()); + assertEquals(0, server.getNumTrackedFunctions()); + + //A registered function is still served repeatedly, and is not left behind once idle + for (int i = 0; i < 3; i++) { + Future<String> found = exec.submit(() -> server.executeBlocking("testing", "test")); + DRPCRequest request = getNextAvailableRequest(server, "testing"); + assertNotNull(request); + server.returnResult(request.get_request_id(), "tested"); + assertEquals("tested", found.get(10, TimeUnit.MILLISECONDS)); + } + assertEquals(0, server.getNumTrackedFunctions()); + + //Nor is a function whose only request timed out + try { + server.executeBlocking("timing-out", "test"); + fail("Should have timed out...."); + } catch (DRPCExecutionException e) { + assertEquals(DRPCExceptionType.SERVER_TIMEOUT, e.get_type()); + } + assertEquals(0, server.getNumTrackedFunctions()); + } + } + @Test public void testDeny() { try (DRPC server = new DRPC(new StormMetricsRegistry(), new DenyAuthorizer(), 100)) {
