Remove DRPC per-function request queues once they are empty

[rzo1]

The map of function name to request queue in DRPC gained an entry for each function name a client asked about, and no code path ever removed one: cleanup() removed the request from a queue and cleanupAll() drained queues, but the queue object and its map entry stayed for the life of the process.

Entries are now dropped once nothing is waiting in them, using compute/computeIfPresent so the map stays race-free. A registered function is served exactly as before, its queue simply being recreated by the next execute(). Extends DRPCTest.

[reiabreu]

Disclaimer: this comment was generated with the help of an LLM.

The new test proves the map returns to empty (the leak fix), but it's essentially sequential — it wouldn't catch a lost-request race. Since correctness here depends on every queue mutation going through the per-key compute lock, a concurrent stress test would strengthen it. Rough sketch:

@Test
public void concurrentSubmitAndFetchLosesNoRequests() throws Exception {
    try (DRPC server = new DRPC(new StormMetricsRegistry(), null, 60_000)) {
        int n = 10_000;
        for (int i = 0; i < n; i++) {
            exec.submit(() -> server.executeBlocking("fn", "x"));
        }
        int served = 0;
        while (served < n) {
            DRPCRequest req = server.fetchRequest("fn");
            if (req != null && !req.get_request_id().isEmpty()) {
                server.returnResult(req.get_request_id(), "ok");
                served++;
            }
        }
        assertEquals(n, served);                          // none lost / duplicated
        assertEquals(0, server.getNumTrackedFunctions()); // no queue left behind
    }
}

Not a blocker — just closes the gap on the part of the change that's hardest to get right.

[rzo1]

Disclaimer: this comment was generated with the help of an LLM.

The new test proves the map returns to empty (the leak fix), but it's essentially sequential — it wouldn't catch a lost-request race. Since correctness here depends on every queue mutation going through the per-key compute lock, a concurrent stress test would strengthen it. Rough sketch:

@Test
public void concurrentSubmitAndFetchLosesNoRequests() throws Exception {
    try (DRPC server = new DRPC(new StormMetricsRegistry(), null, 60_000)) {
        int n = 10_000;
        for (int i = 0; i < n; i++) {
            exec.submit(() -> server.executeBlocking("fn", "x"));
        }
        int served = 0;
        while (served < n) {
            DRPCRequest req = server.fetchRequest("fn");
            if (req != null && !req.get_request_id().isEmpty()) {
                server.returnResult(req.get_request_id(), "ok");
                served++;
            }
        }
        assertEquals(n, served);                          // none lost / duplicated
        assertEquals(0, server.getNumTrackedFunctions()); // no queue left behind
    }
}

Not a blocker — just closes the gap on the part of the change that's hardest to get right.

Added something in that way. Thanks