refactor(inkless:consume): only remote fetches to run on data executor (#466)

* refactor(inkless:consume): only remote fetches to run on data executor

Refactor fetch planning by changing how fetch operations are scheduled
on the data executor. It only assigns the data executor for remote fetch
operations and let the calling thread to run the cache.get(), avoiding
the scenario where calls to cache are blocked by all data executor
threads being used.

By doing this, the CacheFetchJob component becomes irrelevant as most of
the logic is already on the planning, even the scheduling decision.

Changes:
- Delete CacheFetchJob
- Extend ObjectCache to allow passing the executor to run the load
function
- Refactor the FetchPlanner to use the new approach
- Update documentation on FetchPlanner and Reader to clarify how stages
are ran
- Move CacheFetchJobTest to FetchPlannerTest to keep and improve
coverage

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

Remove the temporal merged as this PR does not use bytes or timestamp.
Let that for the following PRs to define

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

Apply suggestion on simplified test

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

Improve error handling on cache async get

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

Improve comment as suggested

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

Further refactoring: Use same constructor as existing methods

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

* fixup! refactor(inkless:consume): only remote fetches to run on data executor

Author: jeqo

storage/inkless/src/main/java/io/aiven/inkless/cache/CaffeineCache.java

@@ -1,3 +1,20 @@
+/*
+ * Inkless
+ * Copyright (C) 2024 - 2025 Aiven OY
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
 package io.aiven.inkless.cache;
 
 import com.github.benmanes.caffeine.cache.AsyncCache;
@@ -7,6 +24,7 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import java.util.function.Function;
 
 import io.aiven.inkless.generated.CacheKey;
@@ -26,7 +44,8 @@ public final class CaffeineCache implements ObjectCache {
     public CaffeineCache(
         final long maxCacheSize,
         final long lifespanSeconds,
-        final int maxIdleSeconds) {
+        final int maxIdleSeconds
+    ) {
         cache = Caffeine.newBuilder()
                 .maximumSize(maxCacheSize)
                 .expireAfterWrite(Duration.ofSeconds(lifespanSeconds))
@@ -42,20 +61,30 @@ public void close() throws IOException {
     }
 
     @Override
-    public FileExtent computeIfAbsent(final CacheKey key, final Function<CacheKey, FileExtent> mappingFunction) {
-        final CompletableFuture<FileExtent> future = new CompletableFuture<>();
-        final CompletableFuture<FileExtent> existingFuture = cache.asMap().computeIfAbsent(key, (cacheKey) -> {
-            return future;
+    public CompletableFuture<FileExtent> computeIfAbsent(
+        final CacheKey key,
+        final Function<CacheKey, FileExtent> load,
+        final Executor loadExecutor
+    ) {
+        // Caffeine's AsyncCache.get() provides atomic cache population per key.
+        // When multiple threads concurrently request the same uncached key, the mapping function
+        // is invoked only once, and all waiting threads receive the same CompletableFuture.
+        // This guarantees that the load function is called at most once per key for successful operations,
+        // preventing duplicate fetch operations from object storage. Failed loads are invalidated and may be retried.
+        return cache.get(key, (k, defaultExecutor) -> {
+            // Use the provided executor instead of Caffeine's default executor.
+            // This allows us to control which thread pool handles the fetch and blocks there,
+            // while Caffeine's internal threads remain unblocked, so cache operations can continue to be served.
+            return CompletableFuture.supplyAsync(() -> load.apply(k), loadExecutor)
+                .whenComplete((result, throwable) -> {
+                    // Evict the entry if the future completed exceptionally.
+                    // While Caffeine has built-in failed future cleanup, it happens asynchronously.
+                    // Explicit invalidation ensures immediate removal for faster retry on subsequent requests.
+                    if (throwable != null) {
+                        cache.synchronous().invalidate(k);
+                    }
+                });
         });
-        // If existing future is not the same object as created in this function
-        // there was a pending cache load and this call is required to join the existing future
-        // and discard the created one.
-        if (future != existingFuture) {
-            return existingFuture.join();
-        }
-        final FileExtent fileExtent = mappingFunction.apply(key);
-        future.complete(fileExtent);
-        return fileExtent;
     }
 
     @Override

storage/inkless/src/main/java/io/aiven/inkless/cache/CaffeineCacheMetrics.java

@@ -1,3 +1,20 @@
+/*
+ * Inkless
+ * Copyright (C) 2024 - 2025 Aiven OY
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
 package io.aiven.inkless.cache;
 
 import org.apache.kafka.common.MetricNameTemplate;

storage/inkless/src/main/java/io/aiven/inkless/cache/MemoryCache.java

@@ -20,11 +20,29 @@
 import org.apache.kafka.common.cache.Cache;
 
 import java.io.Closeable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import java.util.function.Function;
 
 import io.aiven.inkless.generated.CacheKey;
 import io.aiven.inkless.generated.FileExtent;
 
 public interface ObjectCache extends Cache<CacheKey, FileExtent>, Closeable {
-    FileExtent computeIfAbsent(CacheKey key, Function<CacheKey, FileExtent> mappingFunction);
+    /**
+     * Asynchronously computes the value if absent, using the provided executor for the computation.
+     * This method always returns a {@link CompletableFuture} promptly, regardless of whether the key
+     * is already present in the cache. The cache lookup and bookkeeping may still perform synchronous
+     * work on the calling thread, but on a cache miss the actual load computation is executed on the
+     * supplied {@code loadExecutor}.
+     *
+     * @param key the cache key
+     * @param load the function to compute the value if absent
+     * @param loadExecutor the executor to use for async computation on cache miss
+     * @return a CompletableFuture that will complete with the cached or computed value
+     */
+    CompletableFuture<FileExtent> computeIfAbsent(
+        CacheKey key,
+        Function<CacheKey, FileExtent> load,
+        Executor loadExecutor
+    );
 }