fix: make DgraphAsyncClient non-blocking to avoid ForkJoinPool.commonPool starvation

CHANGELOG.md

@@ -6,6 +6,41 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.1.0/),
 
 ## [Unreleased]
 
+**Changed**
+
+- `AsyncTransaction.close()` logs a failed abort instead of throwing it, so closing no longer masks
+  the result of the work the transaction wrapped. Code that catches a close-time failure from an
+  async transaction no longer sees one. `Transaction.close()` still throws; both javadocs record the
+  difference. ([#294])
+- `DgraphAsyncClient.withRetry` completes exceptionally with a bare `DgraphException` on every path,
+  so `whenComplete`, `handle`, and `exceptionally` callbacks receive the `DgraphException` itself.
+  Retry exhaustion previously handed those callbacks a `CompletionException` wrapping it, while a
+  first-attempt failure handed them the exception directly. `join()` and `get()` are unaffected:
+  both wrapped before and still wrap. `DgraphClient.withRetry` runs a separate synchronous retry
+  loop and is unchanged. ([#294])
+
+**Fixed**
+
+- fix: `DgraphAsyncClient` no longer blocks a `ForkJoinPool.commonPool()` thread for the full
+  duration of each gRPC call, which could starve the JVM-wide common pool under load.
+  `CompletableFutures.runWithRetries` now composes on the gRPC future instead of calling a blocking
+  `.get()`, and `jwt` writes are guarded by the write lock. ([#294])
+- fix: futures returned by `DgraphAsyncClient` complete on the executor given to the constructor,
+  including the JWT-refresh retry and `withRetry`'s backoff, which previously completed on a gRPC
+  channel thread and on the common pool respectively. The one exception is a rejection by that
+  executor, which completes the future on whichever thread observed the rejection. ([#294])
+- fix: a `RejectedExecutionException` from the callback executor no longer leaves the returned
+  future permanently incomplete. `withRetry` now relays the rejection instead of hanging, and every
+  rejection surfaces as a `DgraphException` like any other failure. ([#294])
+
+**Deprecated**
+
+- `DgraphAsyncClient(DgraphGrpc.DgraphStub...)` is deprecated in favor of
+  `DgraphAsyncClient(Executor, DgraphGrpc.DgraphStub...)`. It still defaults to
+  `ForkJoinPool.commonPool()`, which is unsuitable for I/O continuations: it is a JVM-wide singleton
+  sized `availableProcessors() - 1`, cannot be tuned per library, and runs unnamed daemon threads
+  that hide contention in a thread dump. Compiling against it warns; nothing breaks. ([#294])
+
 ## [25.0.0] - 2026-04-01
 
 **Added**
@@ -104,6 +139,7 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.1.0/),
 
 - chore: added a test for best effort queries ([#182])
 
+[#294]: https://github.com/dgraph-io/dgraph4j/pull/294
 [#287]: https://github.com/dgraph-io/dgraph4j/pull/287
 [#220]: https://github.com/hypermodeinc/dgraph4j/pull/220
 [#215]: https://github.com/hypermodeinc/dgraph4j/pull/215

src/main/java/io/dgraph/AsyncTransaction.java

@@ -12,8 +12,12 @@
 import io.dgraph.DgraphProto.TxnContext;
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * This is the implementation of asynchronous Dgraph transaction. The asynchrony is backed-up by
@@ -24,6 +28,7 @@
  * @author Michail Klimenkov
  */
 public class AsyncTransaction implements AutoCloseable {
+  private static final Logger LOG = LoggerFactory.getLogger(AsyncTransaction.class);
 
   // these can potentially be set from different threads executing the Stub callback
   private volatile TxnContext context;
@@ -390,8 +395,21 @@ private void mergeContext(final TxnContext src) {
     this.context = builder.build();
   }
 
+  /**
+   * Discards the transaction, blocking until the abort completes. Blocks only when the transaction
+   * has uncommitted mutations, since {@link #discard()} short-circuits otherwise. A failed abort is
+   * logged rather than thrown: the server cleans up abandoned transactions on its own, and throwing
+   * here would mask the outcome of the work this transaction wrapped.
+   *
+   * <p>This diverges from {@link Transaction#close()}, which throws. That one is user-invoked
+   * through try-with-resources, whereas this one also runs from internal cleanup paths.
+   */
   @Override
   public void close() {
-    discard().join();
+    try {
+      discard().join();
+    } catch (CompletionException | CancellationException e) {
+      LOG.warn("discarding the transaction during close failed", e);
+    }
   }
 }

src/main/java/io/dgraph/CompletableFutures.java

@@ -7,6 +7,7 @@
 
 import io.grpc.Context;
 import java.util.concurrent.*;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,40 +41,84 @@ static <T> CompletableFuture<T> runWithRetries(
       Executor executor) {
     final Callable<CompletableFuture<T>> ctxCallable = Context.current().wrap(callable);
 
-    return CompletableFuture.supplyAsync(
-        () -> {
-          try {
-            return ctxCallable.call().get();
-          } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            LOG.error("The " + operation + " got interrupted:", e);
-            throw new DgraphException("The " + operation + " got interrupted", e);
-          } catch (ExecutionException e) {
-            if (Exceptions.isJwtExpired(e.getCause())) {
-              try {
-                retryLogin.get().get();
-                return ctxCallable.call().get();
-              } catch (InterruptedException ie) {
-                Thread.currentThread().interrupt();
-                LOG.error("The retried " + operation + " got interrupted:", ie);
-                throw new DgraphException(
-                    "The retried " + operation + " got interrupted", ie);
-              } catch (ExecutionException ie) {
-                LOG.error(
-                    "The retried " + operation + " encounters an execution exception:", ie);
-                throw new CompletionException(Exceptions.translate(ie.getCause()));
-              } catch (Exception ie) {
-                LOG.error(
-                    "The retried " + operation + " encounters a completion exception:", ie);
-                throw new CompletionException(Exceptions.translate(ie));
-              }
-            }
-            throw new CompletionException(Exceptions.translate(e.getCause()));
-          } catch (Exception e) {
-            throw new CompletionException(Exceptions.translate(e));
-          }
-        },
-        executor);
+    // Composing on the gRPC future rather than blocking on it keeps every stage off the
+    // critical path: no thread is parked for the round trip.
+    return invoke(operation, ctxCallable)
+        .handleAsync(
+            (value, error) ->
+                classify(operation, value, error, ctxCallable, retryLogin, executor),
+            executor)
+        .thenCompose(Function.identity())
+        .handle(CompletableFutures::translateTerminal);
+  }
+
+  /**
+   * Upholds the contract that every failure is a {@code CompletionException} wrapping a {@link
+   * DgraphException}. Runs synchronously so it still fires when {@code executor} rejects a stage.
+   */
+  private static <T> T translateTerminal(T value, Throwable error) {
+    if (error == null) {
+      return value;
+    }
+    throw new CompletionException(Exceptions.translate(unwrap(error)));
+  }
+
+  /** Runs the callable, turning a synchronous throw or a null result into a failed future. */
+  private static <T> CompletableFuture<T> invoke(
+      String operation, Callable<CompletableFuture<T>> callable) {
+    try {
+      CompletableFuture<T> future = callable.call();
+      if (future != null) {
+        return future;
+      }
+      return CompletableFuture.failedFuture(
+          new DgraphException("The " + operation + " returned a null future"));
+    } catch (Exception e) {
+      return CompletableFuture.failedFuture(e);
+    }
+  }
+
+  /** Strips one CompletionException wrapper so error classification sees the real cause. */
+  private static Throwable unwrap(Throwable t) {
+    if (t instanceof CompletionException && t.getCause() != null) {
+      return t.getCause();
+    }
+    return t;
+  }
+
+  /**
+   * Passes a success through, retries once after a JWT refresh on an expired-token failure, and
+   * translates every other failure to {@code CompletionException(DgraphException)}. All paths
+   * complete on {@code executor}.
+   */
+  private static <T> CompletableFuture<T> classify(
+      String operation,
+      T value,
+      Throwable error,
+      Callable<CompletableFuture<T>> ctxCallable,
+      Supplier<CompletableFuture<Void>> retryLogin,
+      Executor executor) {
+    if (error == null) {
+      return CompletableFuture.completedFuture(value);
+    }
+
+    Throwable cause = unwrap(error);
+    if (Exceptions.isJwtExpired(cause)) {
+      return retryLogin
+          .get()
+          .thenComposeAsync(ignored -> invoke(operation, ctxCallable), executor)
+          .handleAsync(
+              (retryValue, retryError) -> {
+                if (retryError != null) {
+                  LOG.error("The retried {} failed", operation, unwrap(retryError));
+                  throw new CompletionException(Exceptions.translate(unwrap(retryError)));
+                }
+                return retryValue;
+              },
+              executor);
+    }
+
+    return CompletableFuture.failedFuture(new CompletionException(Exceptions.translate(cause)));
   }
 
   /**
@@ -85,13 +130,16 @@ static <T> CompletableFuture<T> runWithRetries(
    * @param op the operation to execute within a fresh transaction on each attempt
    * @param attempt the current attempt number (0-based)
    * @param txnFactory creates a new read-write or read-only transaction per the policy
+   * @param executor the callback executor; the backoff delay and the returned future's completion
+   *     run on it, and no thread is held for the duration of a delay
    * @return a CompletableFuture that completes with the result or fails after exhausting retries
    */
   static <T> CompletableFuture<T> attemptAsync(
       RetryPolicy policy,
       AsyncTransactionOp<T> op,
       int attempt,
-      Supplier<AsyncTransaction> txnFactory) {
+      Supplier<AsyncTransaction> txnFactory,
+      Executor executor) {
 
     AsyncTransaction txn = txnFactory.get();
     if (policy.isBestEffort()) {
@@ -100,8 +148,10 @@ static <T> CompletableFuture<T> attemptAsync(
 
     CompletableFuture<T> result = new CompletableFuture<>();
 
+    // handleAsync rather than whenCompleteAsync: the callback consumes the attempt's outcome, so
+    // the stage below carries only a rejection or a bug in the callback, never a retryable failure.
     op.execute(txn)
-        .whenComplete(
+        .handleAsync(
             (value, throwable) -> {
               try {
                 txn.discard();
@@ -111,29 +161,41 @@ static <T> CompletableFuture<T> attemptAsync(
 
               if (throwable == null) {
                 result.complete(value);
-                return;
+                return null;
               }
 
               DgraphException ex = Exceptions.translate(throwable);
               if (!ex.isRetryable() || attempt >= policy.getMaxRetries()) {
                 result.completeExceptionally(ex);
-                return;
+                return null;
               }
 
-              // Schedule retry after backoff delay
               long delayMs = policy.calculateDelay(attempt);
-              Executor delayed =
-                  CompletableFuture.delayedExecutor(delayMs, TimeUnit.MILLISECONDS);
-              CompletableFuture.supplyAsync(() -> null, delayed)
-                  .thenCompose(ignored -> attemptAsync(policy, op, attempt + 1, txnFactory))
+              // The timer takes no executor on purpose: delayedExecutor submits from the internal
+              // Delayer thread, which swallows a rejection and leaves this future incomplete.
+              Executor delayed = CompletableFuture.delayedExecutor(delayMs, TimeUnit.MILLISECONDS);
+              CompletableFuture.runAsync(() -> {}, delayed)
+                  .thenComposeAsync(
+                      ignored -> attemptAsync(policy, op, attempt + 1, txnFactory, executor),
+                      executor)
                   .whenComplete(
                       (retryValue, retryThrowable) -> {
                         if (retryThrowable != null) {
-                          result.completeExceptionally(retryThrowable);
+                          result.completeExceptionally(Exceptions.translate(retryThrowable));
                         } else {
                           result.complete(retryValue);
                         }
                       });
+              return null;
+            },
+            executor)
+        // A rejection by executor completes this stage exceptionally but leaves result untouched,
+        // so relay it or the caller waits forever.
+        .whenComplete(
+            (ignored, t) -> {
+              if (t != null) {
+                result.completeExceptionally(Exceptions.translate(t));
+              }
             });
 
     return result;