feat(client): support custom Executor for async operations

docling-serve/docling-serve-api/src/main/java/ai/docling/serve/api/DoclingServeApi.java

@@ -4,9 +4,10 @@
 
 import java.net.URI;
 import java.time.Duration;
+import java.util.concurrent.Executor;
 import java.util.stream.Collectors;
 
 import org.jspecify.annotations.Nullable;
 
 import ai.docling.serve.api.convert.request.ConvertDocumentRequest;
 import ai.docling.serve.api.spi.DoclingServeApiBuilderFactory;
@@ -15,8 +16,7 @@
 /**
  * Docling Serve API interface.
  */
-public interface DoclingServeApi
-    extends DoclingServeHealthApi, DoclingServeConvertApi, DoclingServeChunkApi, DoclingServeClearApi, DoclingServeTaskApi {
+public interface DoclingServeApi extends DoclingServeHealthApi, DoclingServeConvertApi, DoclingServeChunkApi, DoclingServeClearApi, DoclingServeTaskApi {
 
   /**
    * Creates and returns a builder instance capable of constructing implementations of {@link DoclingServeApi}.
@@ -35,12 +35,14 @@
 
     if (factories.isEmpty()) {
       // No factory found
-      throw new IllegalStateException("No instance of %s found to build a %s instance. You are probably missing a library on your classpath.".formatted(DoclingServeApiBuilderFactory.class.getName(), DoclingApiBuilder.class.getName()));
+      throw new IllegalStateException("No instance of %s found to build a %s instance. You are probably missing a library on your classpath."
+          .formatted(DoclingServeApiBuilderFactory.class.getName(), DoclingApiBuilder.class.getName()));
     }
 
     if (factories.size() > 1) {
       // Multiple factories found
-      throw new IllegalStateException("Multiple instances of %s found to build a %s instance: [%s]".formatted(DoclingServeApiBuilderFactory.class.getName(), DoclingApiBuilder.class.getName(), factories.stream().map(f -> f.getClass().getName()).collect(Collectors.joining(", "))));
+      throw new IllegalStateException("Multiple instances of %s found to build a %s instance: [%s]".formatted(DoclingServeApiBuilderFactory.class.getName(), DoclingApiBuilder.class
+          .getName(), factories.stream().map(f -> f.getClass().getName()).collect(Collectors.joining(", "))));
     }
 
     // Only 1 factory (what we want)
@@ -202,6 +204,31 @@
      */
     B asyncTimeout(Duration asyncTimeout);
 
+    /**
+     * Sets the {@link Executor} used to run async operations.
+     *
+     * <p>This configures where the work of the async methods (such as
+     * {@link DoclingServeApi#convertSourceAsync(ConvertDocumentRequest)}) is executed: submitting
+     * the task, polling for its status and retrieving its result. If not set, async operations run
+     * on the default async executor of {@link java.util.concurrent.CompletableFuture}.
+     *
+     * <p>The lifecycle of the executor is owned by the caller: the client never shuts it down.
+     * Avoid direct executors such as {@code Runnable::run}: the blocking HTTP requests would then run
+     * on the calling thread, making the async methods partially blocking, and on the shared scheduler
+     * thread of {@link java.util.concurrent.CompletableFuture#delayedExecutor(long, java.util.concurrent.TimeUnit, Executor)}.
+     *
+     * <p>The default implementation throws {@link UnsupportedOperationException}, so that existing
+     * builder implementations keep compiling; builders supporting a custom executor override it.
+     *
+     * @param asyncExecutor the executor to use for async operations (must not be null)
+     * @return this builder instance for method chaining
+     * @throws IllegalArgumentException      if asyncExecutor is null
+     * @throws UnsupportedOperationException if this builder does not support a custom executor
+     */
+    default B asyncExecutor(Executor asyncExecutor) {
+      throw new UnsupportedOperationException("A custom async executor is not supported by " + getClass().getName());
+    }
+
     /**
      * Builds and returns an instance of the specified type, representing the completed configuration
      * of the builder. The returned instance is typically an implementation of the Docling API.

docling-serve/docling-serve-api/src/test/java/ai/docling/serve/api/DoclingServeApiTests.java

@@ -2,6 +2,10 @@
 
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
+import java.net.URI;
+import java.time.Duration;
+
+import org.jspecify.annotations.Nullable;
 import org.junit.jupiter.api.Test;
 
 import ai.docling.serve.api.DoclingServeApi.DoclingApiBuilder;
@@ -12,6 +16,67 @@ class DoclingServeApiTests {
   void noFactoryFound() {
     assertThatExceptionOfType(IllegalStateException.class)
         .isThrownBy(() -> DoclingServeApi.builder())
-        .withMessage("No instance of %s found to build a %s instance. You are probably missing a library on your classpath.", DoclingServeApiBuilderFactory.class.getName(), DoclingApiBuilder.class.getName());
+        .withMessage("No instance of %s found to build a %s instance. You are probably missing a library on your classpath.", DoclingServeApiBuilderFactory.class
+            .getName(), DoclingApiBuilder.class.getName());
+  }
+
+  @Test
+  void asyncExecutorIsUnsupportedByDefault() {
+    assertThatExceptionOfType(UnsupportedOperationException.class)
+        .isThrownBy(() -> new MinimalBuilder().asyncExecutor(Runnable::run));
+  }
+
+  // A builder implementing only the abstract methods of the interface, e.g. one provided through the SPI.
+  // This class compiling is what guarantees that new DoclingApiBuilder methods don't break such builders.
+  private static final class MinimalBuilder implements DoclingApiBuilder<DoclingServeApi, MinimalBuilder> {
+    @Override
+    public MinimalBuilder baseUrl(URI baseUrl) {
+      return this;
+    }
+
+    @Override
+    public MinimalBuilder apiKey(@Nullable String apiKey) {
+      return this;
+    }
+
+    @Override
+    public MinimalBuilder logRequests(boolean logRequests) {
+      return this;
+    }
+
+    @Override
+    public MinimalBuilder logResponses(boolean logResponses) {
+      return this;
+    }
+
+    @Override
+    public MinimalBuilder prettyPrint(boolean prettyPrint) {
+      return this;
+    }
+
+    @Override
+    public MinimalBuilder connectTimeout(Duration connectTimeout) {
+      return this;
+    }
+
+    @Override
+    public MinimalBuilder readTimeout(Duration readTimeout) {
+      return this;
+    }
+
+    @Override
+    public MinimalBuilder asyncPollInterval(Duration asyncPollInterval) {
+      return this;
+    }
+
+    @Override
+    public MinimalBuilder asyncTimeout(Duration asyncTimeout) {
+      return this;
+    }
+
+    @Override
+    public DoclingServeApi build() {
+      throw new UnsupportedOperationException();
+    }
   }
 }

docling-serve/docling-serve-client/src/main/java/ai/docling/serve/client/DoclingServeClient.java

@@ -20,6 +20,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Flow.Subscriber;
 import java.util.stream.Collectors;
 
@@ -90,6 +91,7 @@ public abstract class DoclingServeClient extends HttpOperations implements Docli
   private final Duration readTimeout;
   private final Duration asyncPollInterval;
   private final Duration asyncTimeout;
+  private final @Nullable Executor asyncExecutor;
 
   private final HealthOperations healthOps;
   private final ConvertOperations convertOps;
@@ -129,12 +131,13 @@ protected DoclingServeClient(DoclingServeClientBuilder builder) {
     this.apiKey = builder.apiKey;
     this.asyncPollInterval = builder.asyncPollInterval;
     this.asyncTimeout = builder.asyncTimeout;
+    this.asyncExecutor = builder.asyncExecutor;
 
     // Initialize operations handlers
     this.healthOps = new HealthOperations(this);
     this.taskOps = new TaskOperations(this);
-    this.convertOps = new ConvertOperations(this, this.taskOps, this.asyncPollInterval, this.asyncTimeout);
-    this.chunkOps = new ChunkOperations(this, this.taskOps, this.asyncPollInterval, this.asyncTimeout);
+    this.convertOps = new ConvertOperations(this, this.taskOps, this.asyncPollInterval, this.asyncTimeout, this.asyncExecutor);
+    this.chunkOps = new ChunkOperations(this, this.taskOps, this.asyncPollInterval, this.asyncTimeout, this.asyncExecutor);
     this.clearOps = new ClearOperations(this);
   }
 
@@ -176,7 +179,7 @@ protected void logRequest(HttpRequest request) {
           .stream()
           .map(this::maskSensitiveHeaderValues)
           .forEach(entry -> stringBuilder.append("  %s: %s\n".formatted(entry.getKey(), String.join(", ", entry.getValue())))
-      );
+          );
 
       LOG.info(stringBuilder.toString());
     }
@@ -188,8 +191,7 @@ private boolean isSensitiveHeader(String headerName) {
 
   private Map.Entry<String, List<String>> maskSensitiveHeaderValues(Map.Entry<String, List<String>> entry) {
     return Map.entry(
-        entry.getKey(),
-        entry.getValue().stream()
+        entry.getKey(), entry.getValue().stream()
             .map(value -> isSensitiveHeader(entry.getKey()) ? "*".repeat(value.length()) : value)
             .toList()
     );
@@ -201,8 +203,7 @@ protected void logResponse(HttpResponse<String> response, Optional<String> respo
       stringBuilder.append("\n← RESPONSE: %s\n".formatted(response.statusCode()));
       stringBuilder.append("  HEADERS:\n");
 
-      response.headers().map().forEach((key, values) ->
-          stringBuilder.append("  %s: %s\n".formatted(key, String.join(", ", values)))
+      response.headers().map().forEach((key, values) -> stringBuilder.append("  %s: %s\n".formatted(key, String.join(", ", values)))
       );
 
       responseBody
@@ -221,9 +222,10 @@ protected <T> T execute(HttpRequest request, Class<T> expectedValueType) {
 
     try {
       HttpResponse<?> response = null;
-      if(StreamResponse.class.equals(expectedValueType)) {
+      if (StreamResponse.class.equals(expectedValueType)) {
         response = this.httpClient.send(request, BodyHandlers.ofInputStream());
-      } else {
+      }
+      else {
         response = this.httpClient.send(request, BodyHandlers.ofString());
       }
       return getResponse(request, response, expectedValueType);
@@ -281,9 +283,9 @@ protected <I, O> HttpRequest.Builder createRequestBuilder(RequestContext<I, O> r
         .header("Accept", "application/json")
         .timeout(this.readTimeout);
 
-      if (Utils.isNotNullOrBlank(this.apiKey)) {
-        requestBuilder.header(API_KEY_HEADER_NAME, this.apiKey);
-      }
+    if (Utils.isNotNullOrBlank(this.apiKey)) {
+      requestBuilder.header(API_KEY_HEADER_NAME, this.apiKey);
+    }
 
     return requestBuilder;
   }
@@ -309,26 +311,27 @@ protected <T> T getResponse(HttpRequest request, HttpResponse<?> response, Class
       if (StreamResponse.class.equals(expectedReturnType)) {
         // typical 4XX  & 5XX responses are usually accompanied by JSON response bodies
         // hence, reading the stream here.
-        try (InputStream is = (InputStream) body){
+        try (InputStream is = (InputStream) body) {
           body = new String(is.readAllBytes(), StandardCharsets.UTF_8);
-        } catch (IOException e) {
+        }
+        catch (IOException e) {
           throw new DoclingServeClientException(e);
         }
       }
 
       if (statusCode == 422) {
         var validationError = readValue(body.toString(), ValidationError.class);
         var errorText = validationError.getErrorDetails()
-          .stream()
-          .map(ValidationErrorDetail::getMessage)
-          .filter(Objects::nonNull)
-          .collect(Collectors.joining("\n"));
+            .stream()
+            .map(ValidationErrorDetail::getMessage)
+            .filter(Objects::nonNull)
+            .collect(Collectors.joining("\n"));
 
         throw new ValidationException(
-          validationError,
-          "An error occurred while making %s request to %s:\n%s".formatted(request.method(), request.uri(), errorText)
+            validationError, "An error occurred while making %s request to %s:\n%s".formatted(request.method(), request.uri(), errorText)
         );
-      } else {
+      }
+      else {
         throw new DoclingServeClientException("An error occurred: %s".formatted(body.toString()), statusCode, body.toString());
       }
     }
@@ -337,9 +340,10 @@ protected <T> T getResponse(HttpRequest request, HttpResponse<?> response, Class
       return (T) StreamResponse
           .builder()
           .headers(headerName -> response.headers().firstValue(headerName))
-          .body((InputStream)body)
+          .body((InputStream) body)
           .build();
-    } else {
+    }
+    else {
       return readValue(body.toString(), expectedReturnType);
     }
   }
@@ -466,6 +470,7 @@ public abstract static class DoclingServeClientBuilder<C extends DoclingServeCli
     private Duration readTimeout = Duration.ofSeconds(30);
     private Duration asyncPollInterval = Duration.ofSeconds(2);
     private Duration asyncTimeout = Duration.ofMinutes(5);
+    private @Nullable Executor asyncExecutor;
 
     /**
      * Protected constructor for use by subclasses of {@link DoclingServeClientBuilder}.
@@ -493,6 +498,7 @@ protected DoclingServeClientBuilder(DoclingServeClient doclingClient) {
       this.prettyPrintJson = doclingClient.prettyPrintJson;
       this.asyncPollInterval = doclingClient.asyncPollInterval;
       this.asyncTimeout = doclingClient.asyncTimeout;
+      this.asyncExecutor = doclingClient.asyncExecutor;
     }
 
     /**
@@ -590,5 +596,21 @@ public B asyncTimeout(Duration asyncTimeout) {
       this.asyncTimeout = asyncTimeout;
       return (B) this;
     }
+
+    /**
+     * Sets the {@link Executor} used to run async operations.
+     *
+     * <p>If not set, async operations run on the default async executor of
+     * {@link java.util.concurrent.CompletableFuture}. The executor is never shut down by the client.
+     *
+     * @param asyncExecutor the executor to use for async operations (must not be null)
+     * @return this builder instance for method chaining
+     * @throws IllegalArgumentException if asyncExecutor is null
+     */
+    @Override
+    public B asyncExecutor(Executor asyncExecutor) {
+      this.asyncExecutor = ensureNotNull(asyncExecutor, "asyncExecutor");
+      return (B) this;
+    }
   }
 }