You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2026/03/02 13:08:24 UTC
(camel) branch camel-4.18.x updated: Docling backports 4.18.x (#21676)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 12e1bcbae88a Docling backports 4.18.x (#21676)
12e1bcbae88a is described below
commit 12e1bcbae88aa4d7e13cd66408ee06b2c004c713
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Mon Mar 2 14:08:16 2026 +0100
Docling backports 4.18.x (#21676)
* CAMEL-23107 - camel-docling - Add CLI command validation in buildDoclingCommand (#21670)
Signed-off-by: Andrea Cosentino <an...@gmail.com>
* CAMEL-23106 - camel-docling - Remove temp file in DoclingProducer.getInputPath() (#21674)
Signed-off-by: Andrea Cosentino <an...@gmail.com>
* CAMEL-23105 - camel-docling - Fix broken async conversion workflow (discarded result and error masking) (#21675)
The CompletionStage returned by convertSourceAsync() was discarded and
a fabricated task ID with no server-side correlation was returned. The
subsequent CHECK_CONVERSION_STATUS would fail silently because
checkConversionStatusInternal() returned COMPLETED on any exception.
Store the CompletableFuture in a ConcurrentHashMap keyed by a generated
task ID. CHECK_CONVERSION_STATUS now checks the local map first and
returns the actual future state (IN_PROGRESS, COMPLETED with result,
or FAILED with error message). Also fix the error path to return FAILED
instead of COMPLETED when an exception occurs.
Signed-off-by: Andrea Cosentino <an...@gmail.com>
* Align for backport
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---------
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../jbang/camel-jbang-configuration-metadata.json | 2 -
.../camel/component/docling/DoclingProducer.java | 107 ++++++++++-
.../docling/DoclingAsyncConversionTest.java | 200 +++++++++++++++++++++
.../docling/DoclingCustomArgsValidationTest.java | 165 +++++++++++++++++
.../docling/DoclingTempFileCleanupTest.java | 105 +++++++++++
5 files changed, 572 insertions(+), 7 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/jbang/camel-jbang-configuration-metadata.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/jbang/camel-jbang-configuration-metadata.json
index 22ca438bb84d..01c0f4bbcdec 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/jbang/camel-jbang-configuration-metadata.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/jbang/camel-jbang-configuration-metadata.json
@@ -3,7 +3,6 @@
{ "name": "camel.jbang", "description": "Camel JBang configurations" }
],
"properties": [
- { "name": "camel.jbang.buildTool", "required": false, "description": "Build tool to use (Maven or Gradle)", "type": "string", "javaType": "String", "defaultValue": "Maven", "secret": false },
{ "name": "camel.jbang.camel-version", "required": false, "description": "The version of Apache Camel to use", "type": "string", "javaType": "String", "secret": false },
{ "name": "camel.jbang.camelSpringBootVersion", "required": false, "description": "To use a custom Camel version when running or export to Spring Boot", "label": "spring-boot", "type": "string", "javaType": "String", "secret": false },
{ "name": "camel.jbang.classpathFiles", "required": false, "description": "Additional files to add to classpath (Use commas to separate multiple files).", "type": "string", "javaType": "String", "secret": false },
@@ -17,7 +16,6 @@
{ "name": "camel.jbang.excludes", "required": false, "description": "Exclude files by name or pattern (Use commas to separate multiple files)", "type": "string", "javaType": "String", "secret": false },
{ "name": "camel.jbang.exportDir", "required": false, "description": "Directory where the project will be exported", "type": "string", "javaType": "String", "defaultValue": ".", "secret": false },
{ "name": "camel.jbang.gav", "required": false, "description": "Maven coordinate (groupId:artifactId:version)", "type": "string", "javaType": "String", "secret": false },
- { "name": "camel.jbang.gradleWrapper", "required": false, "description": "Include Gradle Wrapper files in the exported project", "type": "boolean", "javaType": "boolean", "defaultValue": true, "secret": false },
{ "name": "camel.jbang.groovyFiles", "required": false, "description": "Additional groovy source files to export to src\/main\/resources\/camel-groovy directory (Use commas to separate multiple files)", "type": "string", "javaType": "String", "secret": false },
{ "name": "camel.jbang.health", "required": false, "description": "Health check at \/observe\/health on local HTTP server (port 8080 by default)", "type": "boolean", "javaType": "boolean", "defaultValue": false, "secret": false, "deprecated": true },
{ "name": "camel.jbang.ignoreLoadingError", "required": false, "description": "Whether to ignore route loading and compilation errors (use this with care!)", "label": "advanced", "type": "boolean", "javaType": "boolean", "defaultValue": false, "secret": false },
diff --git a/components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingProducer.java b/components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingProducer.java
index 38b1b7b720ea..5b06d92154ed 100644
--- a/components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingProducer.java
+++ b/components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingProducer.java
@@ -32,12 +32,14 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -66,6 +68,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.WrappedFile;
import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.SynchronizationAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,6 +82,8 @@ public class DoclingProducer extends DefaultProducer {
private DoclingConfiguration configuration;
private DoclingServeApi doclingServeApi;
private ObjectMapper objectMapper;
+ private final Map<String, CompletableFuture<ConvertDocumentResponse>> pendingAsyncTasks = new ConcurrentHashMap<>();
+ private final AtomicLong taskIdCounter = new AtomicLong();
public DoclingProducer(DoclingEndpoint endpoint) {
super(endpoint);
@@ -121,6 +126,9 @@ public class DoclingProducer extends DefaultProducer {
@Override
protected void doStop() throws Exception {
super.doStop();
+ // Cancel any pending async tasks
+ pendingAsyncTasks.forEach((id, future) -> future.cancel(true));
+ pendingAsyncTasks.clear();
if (doclingServeApi != null) {
doclingServeApi = null;
LOG.info("DoclingServeApi reference cleared");
@@ -261,10 +269,12 @@ public class DoclingProducer extends DefaultProducer {
// Start async conversion
ConvertDocumentRequest request = buildConvertRequest(inputPath, outputFormat);
- CompletionStage<ConvertDocumentResponse> asyncResult = doclingServeApi.convertSourceAsync(request);
+ CompletableFuture<ConvertDocumentResponse> asyncResult
+ = doclingServeApi.convertSourceAsync(request).toCompletableFuture();
- // Generate a task ID for tracking
- String taskId = "task-" + System.currentTimeMillis() + "-" + inputPath.hashCode();
+ // Generate a unique task ID and store the future for later status checks
+ String taskId = "task-" + taskIdCounter.incrementAndGet();
+ pendingAsyncTasks.put(taskId, asyncResult);
LOG.debug("Started async conversion with task ID: {}", taskId);
// Set task ID in body and header
@@ -319,6 +329,13 @@ public class DoclingProducer extends DefaultProducer {
private ConversionStatus checkConversionStatusInternal(String taskId) {
LOG.debug("Checking status for task: {}", taskId);
+ // Check the local pending tasks map first (tasks submitted via SUBMIT_ASYNC_CONVERSION)
+ CompletableFuture<ConvertDocumentResponse> future = pendingAsyncTasks.get(taskId);
+ if (future != null) {
+ return checkLocalAsyncTask(taskId, future);
+ }
+
+ // Fall back to server-side task polling
try {
TaskStatusPollRequest pollRequest = TaskStatusPollRequest.builder()
.taskId(taskId)
@@ -348,8 +365,37 @@ public class DoclingProducer extends DefaultProducer {
return new ConversionStatus(taskId, status);
} catch (Exception e) {
LOG.warn("Failed to check task status for {}: {}", taskId, e.getMessage());
- // If the task ID doesn't exist on the server, return a completed status as a fallback
- return new ConversionStatus(taskId, ConversionStatus.Status.COMPLETED);
+ String errorMsg = e.getMessage() != null ? e.getMessage() : e.getClass().getName();
+ return new ConversionStatus(taskId, ConversionStatus.Status.FAILED, null, errorMsg, null);
+ }
+ }
+
+ private ConversionStatus checkLocalAsyncTask(String taskId, CompletableFuture<ConvertDocumentResponse> future) {
+ if (!future.isDone()) {
+ return new ConversionStatus(taskId, ConversionStatus.Status.IN_PROGRESS);
+ }
+
+ if (future.isCompletedExceptionally() || future.isCancelled()) {
+ // Remove completed task from map
+ pendingAsyncTasks.remove(taskId);
+ String errorMessage;
+ try {
+ future.join();
+ errorMessage = "Unknown error";
+ } catch (Exception e) {
+ errorMessage = e.getCause() != null ? e.getCause().getMessage() : e.getMessage();
+ }
+ return new ConversionStatus(taskId, ConversionStatus.Status.FAILED, null, errorMessage, null);
+ }
+
+ // Completed successfully — extract the result and remove from map
+ pendingAsyncTasks.remove(taskId);
+ try {
+ ConvertDocumentResponse response = future.join();
+ String result = extractConvertedContent(response, configuration.getOutputFormat());
+ return new ConversionStatus(taskId, ConversionStatus.Status.COMPLETED, result, null, null);
+ } catch (Exception e) {
+ return new ConversionStatus(taskId, ConversionStatus.Status.FAILED, null, e.getMessage(), null);
}
}
@@ -1357,6 +1403,7 @@ public class DoclingProducer extends DefaultProducer {
// Treat as content to be written to a temp file
Path tempFile = Files.createTempFile("docling-", ".tmp");
Files.write(tempFile, content.getBytes());
+ registerTempFileCleanup(exchange, tempFile);
validateFileSize(tempFile.toString());
return tempFile.toString();
}
@@ -1366,6 +1413,7 @@ public class DoclingProducer extends DefaultProducer {
}
Path tempFile = Files.createTempFile("docling-", ".tmp");
Files.write(tempFile, content);
+ registerTempFileCleanup(exchange, tempFile);
return tempFile.toString();
} else if (body instanceof File file) {
validateFileSize(file.getAbsolutePath());
@@ -1375,6 +1423,20 @@ public class DoclingProducer extends DefaultProducer {
throw new InvalidPayloadException(exchange, String.class);
}
+ private void registerTempFileCleanup(Exchange exchange, Path tempFile) {
+ exchange.getExchangeExtension().addOnCompletion(new SynchronizationAdapter() {
+ @Override
+ public void onDone(Exchange exchange) {
+ try {
+ Files.deleteIfExists(tempFile);
+ LOG.debug("Cleaned up temp file: {}", tempFile);
+ } catch (IOException e) {
+ LOG.warn("Failed to clean up temp file: {}", tempFile, e);
+ }
+ }
+ });
+ }
+
private void validateFileSize(String filePath) throws IOException {
Path path = Paths.get(filePath);
if (Files.exists(path)) {
@@ -1603,11 +1665,46 @@ public class DoclingProducer extends DefaultProducer {
@SuppressWarnings("unchecked")
List<String> customArgs = exchange.getIn().getHeader(DoclingHeaders.CUSTOM_ARGUMENTS, List.class);
if (customArgs != null && !customArgs.isEmpty()) {
+ validateCustomArguments(customArgs);
LOG.debug("Adding custom Docling arguments: {}", customArgs);
command.addAll(customArgs);
}
}
+ /**
+ * Validates custom CLI arguments to ensure they do not conflict with producer-managed options such as the output
+ * directory.
+ */
+ private void validateCustomArguments(List<String> customArgs) {
+ // The output directory is managed by the producer via endpoint configuration
+ // or the OUTPUT_FILE_PATH header, so it must not be overridden through custom arguments.
+ List<String> blockedFlags = List.of("--output", "-o");
+
+ for (int i = 0; i < customArgs.size(); i++) {
+ String arg = customArgs.get(i);
+
+ if (arg == null) {
+ throw new IllegalArgumentException("Custom argument at index " + i + " is null");
+ }
+
+ String argLower = arg.toLowerCase();
+ for (String blocked : blockedFlags) {
+ if (argLower.equals(blocked) || argLower.startsWith(blocked + "=")) {
+ throw new IllegalArgumentException(
+ "Custom argument '" + blocked
+ + "' is not allowed because the output directory is managed by the producer. "
+ + "Use the " + DoclingHeaders.OUTPUT_FILE_PATH
+ + " header or endpoint configuration instead.");
+ }
+ }
+
+ if (arg.contains("../") || arg.contains("..\\")) {
+ throw new IllegalArgumentException(
+ "Custom argument at index " + i + " contains a relative path traversal sequence");
+ }
+ }
+ }
+
private void addOutputFormatArguments(List<String> command, String outputFormat) {
if (outputFormat != null && !outputFormat.isEmpty()) {
command.add("--to");
diff --git a/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingAsyncConversionTest.java b/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingAsyncConversionTest.java
new file mode 100644
index 000000000000..3fd13d8c2cd1
--- /dev/null
+++ b/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingAsyncConversionTest.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docling;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+import ai.docling.serve.api.convert.response.ConvertDocumentResponse;
+import ai.docling.serve.api.convert.response.DocumentResponse;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Tests the SUBMIT_ASYNC_CONVERSION and CHECK_CONVERSION_STATUS two-step async workflow.
+ *
+ * <p>
+ * Before the fix, the {@code CompletionStage} returned by {@code convertSourceAsync()} was discarded and a fabricated
+ * task ID with no server-side correlation was returned. CHECK_CONVERSION_STATUS would then fail because the server had
+ * no record of the fake ID, and the error was silently masked by returning COMPLETED.
+ *
+ * <p>
+ * After the fix, the {@code CompletableFuture} is stored in a local map keyed by the generated task ID. When
+ * CHECK_CONVERSION_STATUS is called, it checks the local map first and returns the actual status of the async task.
+ */
+class DoclingAsyncConversionTest extends CamelTestSupport {
+
+ @Test
+ void submitReturnsTaskIdLinkedToFuture() throws Exception {
+ DoclingEndpoint endpoint = context.getEndpoint(
+ "docling:convert?operation=SUBMIT_ASYNC_CONVERSION&useDoclingServe=true", DoclingEndpoint.class);
+ DoclingProducer producer = (DoclingProducer) endpoint.createProducer();
+
+ // Access the pendingAsyncTasks map via reflection to verify the future is stored
+ Map<String, CompletableFuture<ConvertDocumentResponse>> pendingTasks = getPendingAsyncTasks(producer);
+ assertNotNull(pendingTasks, "pendingAsyncTasks map should exist");
+ assertTrue(pendingTasks.isEmpty(), "pendingAsyncTasks should start empty");
+ }
+
+ @Test
+ void checkStatusReturnsFailedForUnknownTaskId() throws Exception {
+ // When CHECK_CONVERSION_STATUS is called with an unknown task ID and the server
+ // is not available, it should return FAILED — not COMPLETED (the old bug).
+ try {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(DoclingHeaders.TASK_ID, "nonexistent-task-id");
+ exchange.getIn().setHeader(DoclingHeaders.OPERATION, DoclingOperations.CHECK_CONVERSION_STATUS);
+
+ template.send("direct:check-status", exchange);
+
+ Object body = exchange.getIn().getBody();
+ assertInstanceOf(ConversionStatus.class, body);
+ ConversionStatus status = (ConversionStatus) body;
+
+ // The key assertion: unknown task IDs should NOT return COMPLETED
+ assertNotEquals(ConversionStatus.Status.COMPLETED, status.getStatus(),
+ "Unknown task ID should not return COMPLETED status");
+ assertEquals(ConversionStatus.Status.FAILED, status.getStatus(),
+ "Unknown task ID with unavailable server should return FAILED");
+ assertNotNull(status.getErrorMessage(), "Error message should be populated");
+ } catch (CamelExecutionException e) {
+ // If the exchange throws instead of setting FAILED status, that's also acceptable —
+ // the important thing is it doesn't silently return COMPLETED
+ }
+ }
+
+ @Test
+ void checkStatusReturnsCompletedForFinishedLocalTask() throws Exception {
+ DoclingEndpoint endpoint = context.getEndpoint(
+ "docling:convert?operation=CHECK_CONVERSION_STATUS&useDoclingServe=true", DoclingEndpoint.class);
+ DoclingProducer producer = (DoclingProducer) endpoint.createProducer();
+
+ // Manually insert a completed future into the pending tasks map
+ Map<String, CompletableFuture<ConvertDocumentResponse>> pendingTasks = getPendingAsyncTasks(producer);
+
+ // Create a completed future with a mock response
+ ConvertDocumentResponse mockResponse = ConvertDocumentResponse.builder()
+ .document(DocumentResponse.builder()
+ .markdownContent("# Converted Document")
+ .build())
+ .build();
+ CompletableFuture<ConvertDocumentResponse> completedFuture = CompletableFuture.completedFuture(mockResponse);
+ pendingTasks.put("test-task-1", completedFuture);
+
+ // Check the status — should find it in local map and return COMPLETED with result
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(DoclingHeaders.TASK_ID, "test-task-1");
+
+ producer.process(exchange);
+
+ Object body = exchange.getIn().getBody();
+ assertInstanceOf(ConversionStatus.class, body);
+ ConversionStatus status = (ConversionStatus) body;
+ assertEquals(ConversionStatus.Status.COMPLETED, status.getStatus());
+ assertNotNull(status.getResult(), "Result should contain the converted content");
+
+ // Future should be removed from map after completion
+ assertFalse(pendingTasks.containsKey("test-task-1"),
+ "Completed task should be removed from pending map");
+ }
+
+ @Test
+ void checkStatusReturnsInProgressForPendingLocalTask() throws Exception {
+ DoclingEndpoint endpoint = context.getEndpoint(
+ "docling:convert?operation=CHECK_CONVERSION_STATUS&useDoclingServe=true", DoclingEndpoint.class);
+ DoclingProducer producer = (DoclingProducer) endpoint.createProducer();
+
+ Map<String, CompletableFuture<ConvertDocumentResponse>> pendingTasks = getPendingAsyncTasks(producer);
+
+ // Insert an incomplete future
+ CompletableFuture<ConvertDocumentResponse> incompleteFuture = new CompletableFuture<>();
+ pendingTasks.put("test-task-2", incompleteFuture);
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(DoclingHeaders.TASK_ID, "test-task-2");
+
+ producer.process(exchange);
+
+ Object body = exchange.getIn().getBody();
+ assertInstanceOf(ConversionStatus.class, body);
+ ConversionStatus status = (ConversionStatus) body;
+ assertEquals(ConversionStatus.Status.IN_PROGRESS, status.getStatus());
+
+ // Future should remain in map since it's not done yet
+ assertTrue(pendingTasks.containsKey("test-task-2"),
+ "In-progress task should remain in pending map");
+
+ // Clean up
+ incompleteFuture.cancel(true);
+ }
+
+ @Test
+ void checkStatusReturnsFailedForExceptionalLocalTask() throws Exception {
+ DoclingEndpoint endpoint = context.getEndpoint(
+ "docling:convert?operation=CHECK_CONVERSION_STATUS&useDoclingServe=true", DoclingEndpoint.class);
+ DoclingProducer producer = (DoclingProducer) endpoint.createProducer();
+
+ Map<String, CompletableFuture<ConvertDocumentResponse>> pendingTasks = getPendingAsyncTasks(producer);
+
+ // Insert a failed future
+ CompletableFuture<ConvertDocumentResponse> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new RuntimeException("Server connection refused"));
+ pendingTasks.put("test-task-3", failedFuture);
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(DoclingHeaders.TASK_ID, "test-task-3");
+
+ producer.process(exchange);
+
+ Object body = exchange.getIn().getBody();
+ assertInstanceOf(ConversionStatus.class, body);
+ ConversionStatus status = (ConversionStatus) body;
+ assertEquals(ConversionStatus.Status.FAILED, status.getStatus());
+ assertNotNull(status.getErrorMessage());
+ assertTrue(status.getErrorMessage().contains("Server connection refused"));
+
+ // Failed task should be removed from map
+ assertFalse(pendingTasks.containsKey("test-task-3"),
+ "Failed task should be removed from pending map");
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<String, CompletableFuture<ConvertDocumentResponse>> getPendingAsyncTasks(DoclingProducer producer)
+ throws Exception {
+ Field field = DoclingProducer.class.getDeclaredField("pendingAsyncTasks");
+ field.setAccessible(true);
+ return (Map<String, CompletableFuture<ConvertDocumentResponse>>) field.get(producer);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:check-status")
+ .to("docling:convert?operation=CHECK_CONVERSION_STATUS&useDoclingServe=true");
+ }
+ };
+ }
+}
diff --git a/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingCustomArgsValidationTest.java b/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingCustomArgsValidationTest.java
new file mode 100644
index 000000000000..926f0a9403b3
--- /dev/null
+++ b/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingCustomArgsValidationTest.java
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docling;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Tests that custom CLI arguments passed via the {@link DoclingHeaders#CUSTOM_ARGUMENTS} header are validated to ensure
+ * they do not conflict with producer-managed options such as the output directory.
+ */
+class DoclingCustomArgsValidationTest extends CamelTestSupport {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void customArgsWithOutputFlagAreRejected() throws Exception {
+ Path inputFile = createInputFile();
+
+ // The --output flag conflicts with the producer-managed output directory
+ // and should be rejected before the process is started.
+ CamelExecutionException ex = assertThrows(CamelExecutionException.class, () -> {
+ template.requestBodyAndHeaders("direct:cli-convert",
+ inputFile.toString(),
+ java.util.Map.of(DoclingHeaders.CUSTOM_ARGUMENTS, List.of("--output", "/tmp/other-dir")));
+ });
+
+ assertInstanceOf(IllegalArgumentException.class, ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("--output"));
+ assertTrue(ex.getCause().getMessage().contains("not allowed"));
+ }
+
+ @Test
+ void customArgsWithOutputEqualsFormAreRejected() throws Exception {
+ Path inputFile = createInputFile();
+
+ CamelExecutionException ex = assertThrows(CamelExecutionException.class, () -> {
+ template.requestBodyAndHeaders("direct:cli-convert",
+ inputFile.toString(),
+ java.util.Map.of(DoclingHeaders.CUSTOM_ARGUMENTS, List.of("--output=/tmp/other-dir")));
+ });
+
+ assertInstanceOf(IllegalArgumentException.class, ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("--output"));
+ }
+
+ @Test
+ void customArgsWithShortOutputFlagAreRejected() throws Exception {
+ Path inputFile = createInputFile();
+
+ CamelExecutionException ex = assertThrows(CamelExecutionException.class, () -> {
+ template.requestBodyAndHeaders("direct:cli-convert",
+ inputFile.toString(),
+ java.util.Map.of(DoclingHeaders.CUSTOM_ARGUMENTS, List.of("-o", "/tmp/other-dir")));
+ });
+
+ assertInstanceOf(IllegalArgumentException.class, ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("-o"));
+ }
+
+ @Test
+ void customArgsWithPathTraversalAreRejected() throws Exception {
+ Path inputFile = createInputFile();
+
+ CamelExecutionException ex = assertThrows(CamelExecutionException.class, () -> {
+ template.requestBodyAndHeaders("direct:cli-convert",
+ inputFile.toString(),
+ java.util.Map.of(DoclingHeaders.CUSTOM_ARGUMENTS, List.of("--some-flag", "../../etc/passwd")));
+ });
+
+ assertInstanceOf(IllegalArgumentException.class, ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("relative path traversal"));
+ }
+
+ @Test
+ void customArgsWithNullEntryAreRejected() throws Exception {
+ Path inputFile = createInputFile();
+ List<String> argsWithNull = new java.util.ArrayList<>();
+ argsWithNull.add(null);
+
+ CamelExecutionException ex = assertThrows(CamelExecutionException.class, () -> {
+ template.requestBodyAndHeaders("direct:cli-convert",
+ inputFile.toString(),
+ java.util.Map.of(DoclingHeaders.CUSTOM_ARGUMENTS, argsWithNull));
+ });
+
+ assertInstanceOf(IllegalArgumentException.class, ex.getCause());
+ assertTrue(ex.getCause().getMessage().contains("null"));
+ }
+
+ @Test
+ void safeCustomArgsPassValidation() throws Exception {
+ Path inputFile = createInputFile();
+
+ // Safe arguments should pass validation and proceed to process execution.
+ // The process itself will fail (docling binary not installed in test env),
+ // but the error should NOT be IllegalArgumentException — that proves
+ // the validation passed and execution moved on to ProcessBuilder.
+ CamelExecutionException ex = assertThrows(CamelExecutionException.class, () -> {
+ template.requestBodyAndHeaders("direct:cli-convert",
+ inputFile.toString(),
+ java.util.Map.of(DoclingHeaders.CUSTOM_ARGUMENTS, List.of("--verbose", "--table-mode", "fast")));
+ });
+
+ // The failure should be from process execution, not from argument validation
+ assertFalse(ex.getCause() instanceof IllegalArgumentException,
+ "Safe custom arguments should pass validation; failure should come from process execution, not argument validation");
+ }
+
+ @Test
+ void noCustomArgsIsAllowed() throws Exception {
+ Path inputFile = createInputFile();
+
+ // With no custom arguments at all, validation is skipped entirely.
+ // The process will still fail (no docling binary), but not from validation.
+ CamelExecutionException ex = assertThrows(CamelExecutionException.class, () -> {
+ template.requestBody("direct:cli-convert", inputFile.toString());
+ });
+
+ assertFalse(ex.getCause() instanceof IllegalArgumentException,
+ "No custom arguments should not trigger argument validation");
+ }
+
+ private Path createInputFile() throws Exception {
+ Path file = tempDir.resolve("test-input.txt");
+ Files.writeString(file, "test content");
+ return file;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // CLI mode (useDoclingServe=false is the default)
+ from("direct:cli-convert")
+ .to("docling:convert?operation=CONVERT_TO_MARKDOWN");
+ }
+ };
+ }
+}
diff --git a/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingTempFileCleanupTest.java b/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingTempFileCleanupTest.java
new file mode 100644
index 000000000000..4a70d652d711
--- /dev/null
+++ b/components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingTempFileCleanupTest.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docling;
+
+import java.io.IOException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Verifies that temporary files created in {@code DoclingProducer.getInputPath()} are cleaned up after exchange
+ * processing completes.
+ *
+ * <p>
+ * Before the fix, temp files created for String content and byte[] bodies accumulated on disk indefinitely. After the
+ * fix, an {@code addOnCompletion} callback deletes them when the exchange finishes.
+ */
+class DoclingTempFileCleanupTest extends CamelTestSupport {
+
+ @Test
+ void tempFileFromStringContentIsCleanedUp() throws Exception {
+ // Snapshot temp files before
+ List<Path> before = listDoclingTempFiles();
+
+ // Send string content (not a path, not a URL) — this triggers temp file creation.
+ // The docling CLI will fail (not installed), but the temp file cleanup
+ // runs on exchange completion regardless of success or failure.
+ try {
+ template.requestBody("direct:convert", "This is raw text content to convert");
+ } catch (CamelExecutionException e) {
+ // Expected — docling binary not available in test env
+ }
+
+ // After exchange completes, temp files should have been cleaned up
+ List<Path> after = listDoclingTempFiles();
+ List<Path> leaked = new ArrayList<>(after);
+ leaked.removeAll(before);
+
+ assertTrue(leaked.isEmpty(),
+ "Temp files leaked after exchange completion: " + leaked);
+ }
+
+ @Test
+ void tempFileFromByteArrayIsCleanedUp() throws Exception {
+ List<Path> before = listDoclingTempFiles();
+
+ try {
+ template.requestBody("direct:convert", "Binary content for conversion".getBytes());
+ } catch (CamelExecutionException e) {
+ // Expected — docling binary not available in test env
+ }
+
+ List<Path> after = listDoclingTempFiles();
+ List<Path> leaked = new ArrayList<>(after);
+ leaked.removeAll(before);
+
+ assertTrue(leaked.isEmpty(),
+ "Temp files leaked after exchange completion: " + leaked);
+ }
+
+ private List<Path> listDoclingTempFiles() throws IOException {
+ List<Path> result = new ArrayList<>();
+ Path tmpDir = Path.of(System.getProperty("java.io.tmpdir"));
+ try (DirectoryStream<Path> stream = Files.newDirectoryStream(tmpDir, "docling-*.tmp")) {
+ for (Path entry : stream) {
+ result.add(entry);
+ }
+ }
+ return result;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:convert")
+ .to("docling:convert?operation=CONVERT_TO_MARKDOWN");
+ }
+ };
+ }
+}