Add cancellation support to IndicesRequestCache

[drempapis]

Related github issue #108703

The problem
When expensive queries fill up the search thread pool, threads can become blocked in IndicesRequestCache.getOrCompute waiting for other threads to compute cached results. If these queries are cancelled, the waiting threads don't react to the cancellation and continue blocking indefinitely. This can lead to search thread pool exhaustion, requiring node restarts to recover.

This PR adds cancellation support to the cache's blocking operations, allowing waiting threads to be notified when their task is cancelled. However, this PR does not prevent the search pool from filling with blocking tasks. A follow-up pr will follow this one, changing the cache to use SubscribableListener for a complete async solution. I haven't worked on it here for simplicity and to make the code updates discrete.

[elasticsearchmachine]

Pinging @elastic/es-search-foundations (Team:Search Foundations)

[eranweiss-elastic]

Java doc mentions TaskCancelledException being thrown, but the definition doesn't.

[drempapis]

That's true, removed

[eranweiss-elastic]

I think that segment.map.get(key) != future deserves some handling. It's not expected, but maybe at least a log.

[drempapis]

I added a debug log when the key maps to a different future: Skipped cleanup for key [] because the future was replaced.

Do you think that it is enough? Do you suggest doing something else here?

[eranweiss-elastic]

I'm a bit worried about this if. This assumes that a cancellationRegistrar != null is passed if a task can be cancelled, but doesn't enforce it. If a task is cancelled for a future when cancellationRegistrar is null, this with be a deadlock.
A safer way to do this will be to add the latch.countDown(); regardless of the value of cancellationRegistrar.

[drempapis]

The code is structured as follows

 future.whenComplete((value, throwable) -> {
            if (throwable != null) {
                error.set(throwable);
            } else {
                result.set(value);
            }
            latch.countDown();
        });

        if (cancellationRegistrar != null) {
            cancellationRegistrar.accept(() -> {
                cancelled.set(true);
                latch.countDown();
            });
        }

The whenComplete callback, which always calls latch.countDown(), is registered before the cancellationRegistrar check. The latch is always released when the future completes, regardless of whether a registrar is provided.

When cancellationRegistrar == null, the waiting thread cannot exit early if its task is cancelled. It will remain blocked until the computation finishes. This is by design, not a deadlock.

Add latch.countDown() regardless of cancellationRegistrar would immediately count down the latch causing latch.await() to return instantly with no result.

I've added a test to prove that when cancellationRegistrar is null, the thread is not deadlocked, it simply cannot exit early and must wait for the future to complete.

[eranweiss-elastic]

Because this is a static method, and has locking, I think it would be nice to test it in CacheTests.java.

[drempapis]

+1, added more tests