You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "oscerd (via GitHub)" <gi...@apache.org> on 2026/10/07 07:36:05 UTC
[PR] CAMEL-23458: camel-docling - add consumer support for async task completion events [camel]
oscerd opened a new pull request, #27477:
URL: https://github.com/apache/camel/pull/27477
# CAMEL-23458: camel-docling - add consumer support for async task completion events
`camel-docling` was producer-only: `DoclingEndpoint.createConsumer()` threw, so a route that wanted to act on an
async conversion had to poll `CHECK_CONVERSION_STATUS` by hand. This adds an event-driven consumer.
## What this adds
A **scheduled polling consumer** that drains the component-wide registry of tasks submitted via
`SUBMIT_ASYNC_CONVERSION` (the in-memory `DoclingComponent.pendingAsyncTasks` map) and emits one exchange as each task
completes:
```java
from("direct:submit")
.to("docling:convert?useDoclingServe=true&operation=SUBMIT_ASYNC_CONVERSION");
from("docling:onComplete?useDoclingServe=true&delay=5000")
.to("mock:done");
```
- a **successful** conversion carries the converted content as the body — the same shape the synchronous `CONVERT_*`
operations produce for the configured `outputFormat` — plus the task id in the `CamelDoclingTaskId` header;
- a **failed** conversion completes the exchange with the underlying exception, so the route's `onException` applies;
- the completed task is claimed atomically (`remove(key, value)`) so a concurrent `CHECK_CONVERSION_STATUS` or another
consumer cannot emit it twice.
## Design notes
- The two viable strategies in the issue were polling and webhook; docling's async tasks are SDK `CompletableFuture`s
held in-memory on the component, so the **polling** strategy fits (there is no server-side task list to webhook from).
- `DoclingEndpoint` now extends `ScheduledPollEndpoint` (was a producer-only `DefaultEndpoint`), which gives the
poll-interval configuration (`delay`, `initialDelay`, `useFixedDelay`, ...) for free.
- The content extraction shared by the synchronous operations and the consumer was moved into a small
`DoclingContentExtractor` util, so both yield the same body for a given `outputFormat` (no duplication).
- Security: the consumer adds no new input/output surface — it only drains conversions already submitted through the
existing producer path.
## Tests & docs
- `DoclingConsumerTest` — a completed conversion is emitted to the route with the converted body and drained from the
map.
- `DoclingConsumerFailureTest` — a conversion that completed exceptionally is routed to `onException`.
- `DoclingConsumerIT` — end-to-end against a real docling-serve (submit async → consume completion); disabled on CI like
the other docling ITs, so the deterministic failure path is covered by the unit test above.
- `docling-component.adoc` documents the consumer with a worked example.
`main` only.
---
_Claude Code on behalf of oscerd_
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23458: camel-docling - add consumer support for async task completion events [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd merged PR #27477:
URL: https://github.com/apache/camel/pull/27477
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23458: camel-docling - add consumer support for async task completion events [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27477:
URL: https://github.com/apache/camel/pull/27477#issuecomment-6033269839
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23458: camel-docling - add consumer support for async task completion events [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd commented on PR #27477:
URL: https://github.com/apache/camel/pull/27477#issuecomment-6034016855
Thanks for the thorough review, @davsclaus. Addressed in 4abae810a1d9:
- **Output-format mismatch (blocking):** fixed — the submit-time format is stored on `AsyncTaskEntry` and used by both the consumer and `checkLocalAsyncTask`; new unit test covers the html-submitted / markdown-consumed case. (Replied inline.)
- **Consumer scope & path segment:** the new NOTE says to run at most one consumer per component (a second competes for the same tasks), and clarifies that the path is still the required `operationId` but the consumer ignores the operation — it always drains the component's completed tasks.
- **In-memory only:** the NOTE now states pending tasks live in an in-memory map visible only to the same `CamelContext`, are lost on restart, and that stopping the context cancels in-flight conversions.
- **Minor — failure path runs the processor with the exception pre-set:** this is intentional. Setting the exception on the exchange and then routing it is what makes the *route's* `onException` handle the async failure (as `DoclingConsumerFailureTest` asserts). Calling the consumer's exception handler directly instead would bypass the route's error handling, which is the opposite of what we want here. Left as-is — happy to add an inline code comment if that reads clearer.
_Claude Code on behalf of oscerd_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23458: camel-docling - add consumer support for async task completion events [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27477:
URL: https://github.com/apache/camel/pull/27477#discussion_r4204456861
##########
components/camel-ai/camel-docling/src/main/docs/docling-component.adoc:
##########
@@ -1219,6 +1219,33 @@ YAML::
----
====
+== Consuming asynchronous conversions
+
+Instead of submitting a conversion with `SUBMIT_ASYNC_CONVERSION` and then polling `CHECK_CONVERSION_STATUS` by hand, a
+`docling:` consumer can emit each async conversion event-driven, as soon as it finishes:
+
+[source,java]
+----
+// submit conversions asynchronously; each returns a task id and runs in the background
+from("direct:submit")
+ .to("docling:convert?useDoclingServe=true&operation=SUBMIT_ASYNC_CONVERSION");
+
+// the consumer emits one exchange per completed task (the path, here "onComplete", is just a label)
+from("docling:onComplete?useDoclingServe=true&delay=5000")
+ .to("mock:done");
+----
+
+The consumer scans the tasks submitted via `SUBMIT_ASYNC_CONVERSION` on the same component and emits one exchange as
Review Comment:
**The consumer changes what `CHECK_CONVERSION_STATUS` returns.** The consumer removes every completed task from the component-wide map. After that, `CHECK_CONVERSION_STATUS` for a local id such as `task-3` no longer finds it locally and falls back to `pollTaskStatus` on docling-serve. The server doesn't know local ids, so it returns `FAILED` with an error message. In a context that has a consumer, polling by hand therefore reports successful conversions as failed.
Could the docs say that the consumer and manual `CHECK_CONVERSION_STATUS` polling shouldn't be mixed on the same component? Alternatively, the fallback could know which ids the consumer already took.
##########
components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingConsumer.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.util.Map;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+import ai.docling.serve.api.convert.response.ConvertDocumentResponse;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.support.ScheduledPollConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Event-driven consumer for docling async conversions.
+ * <p>
+ * It polls the component-wide registry of tasks submitted via {@code SUBMIT_ASYNC_CONVERSION} and, each time a task
+ * completes, emits one exchange whose body is the converted content (the same shape the synchronous {@code CONVERT_*}
+ * operations produce for the configured {@code outputFormat}), carrying the task id in the
+ * {@link DoclingHeaders#TASK_ID} header. A task that failed completes the exchange with the underlying exception, so
+ * the route's {@code onException} machinery applies. The poll interval is the standard scheduled-consumer {@code delay}
+ * option.
+ */
+public class DoclingConsumer extends ScheduledPollConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DoclingConsumer.class);
+
+ public DoclingConsumer(DoclingEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ }
+
+ @Override
+ public DoclingEndpoint getEndpoint() {
+ return (DoclingEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected int poll() throws Exception {
+ Map<String, AsyncTaskEntry> pendingAsyncTasks
+ = ((DoclingComponent) getEndpoint().getComponent()).getPendingAsyncTasks();
+
+ int polled = 0;
+ for (Map.Entry<String, AsyncTaskEntry> entry : pendingAsyncTasks.entrySet()) {
+ String taskId = entry.getKey();
+ AsyncTaskEntry task = entry.getValue();
+ if (!task.getFuture().isDone()) {
+ continue;
+ }
+ // claim the completed task atomically, so a concurrent CHECK_CONVERSION_STATUS or another consumer cannot
+ // emit the same completion twice
+ if (!pendingAsyncTasks.remove(taskId, task)) {
+ continue;
+ }
+
+ polled++;
+ Exchange exchange = createExchange(false);
+ try {
+ exchange.getIn().setHeader(DoclingHeaders.TASK_ID, taskId);
+ emitResult(exchange, taskId, task.getFuture());
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ } finally {
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException(
+ "Error processing docling async task " + taskId, exchange, exchange.getException());
+ }
+ releaseExchange(exchange, false);
+ }
+ }
+ return polled;
+ }
+
+ private void emitResult(Exchange exchange, String taskId, CompletableFuture<ConvertDocumentResponse> future) {
+ try {
+ ConvertDocumentResponse response = future.join();
+ exchange.getIn().setBody(
+ DoclingContentExtractor.extract(response, getEndpoint().getConfiguration().getOutputFormat()));
Review Comment:
**Output format mismatch, which can give an empty body.** The producer submits the async request with the format from the `CamelDoclingOutputFormat` header, falling back to the producer endpoint's `outputFormat` (`processSubmitAsyncConversion`). Here the consumer extracts with *its own* endpoint's `outputFormat` (default markdown).
Example: a task submitted with header `OutputFormat=html`, or from `docling:convert?outputFormat=html&operation=SUBMIT_ASYNC_CONVERSION`, and consumed by `docling:onComplete?useDoclingServe=true`. docling-serve returns only HTML, `getMarkdownContent()` is `null`, and the extractor returns `""`. The route gets an empty body with no error.
Suggestion: store the format on the task when it is submitted, and use it here:
```java
// AsyncTaskEntry: new field + getter
private final String outputFormat;
// DoclingProducer.processSubmitAsyncConversion
AsyncTaskEntry taskEntry = new AsyncTaskEntry(taskId, asyncResult, outputFormat);
// DoclingConsumer.emitResult
DoclingContentExtractor.extract(response, task.getOutputFormat())
```
`checkLocalAsyncTask` in the producer has the same pre-existing gap (it uses `configuration.getOutputFormat()`), so the same field fixes both. A unit test that submits with one format and consumes from an endpoint with the default format would cover it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23458: camel-docling - add consumer support for async task completion events [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27477:
URL: https://github.com/apache/camel/pull/27477#issuecomment-6033733616
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `catalog/camel-catalog`
- `components/camel-ai/camel-docling`
- `dsl/camel-componentdsl`
- `dsl/camel-endpointdsl`
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 10 of 699 tested, 24 compile-only — current: 10 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **10 affected modules** (current approach: 10).
**Skip-tests mode would test 10 modules** (4 direct + 9 downstream), **skip tests for 24** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (10)</summary>
- `camel-docling` ← components/camel-ai/camel-docling/src/generated/java/org/apache/camel/component/docling/DoclingComponentConfigurer.java, components/camel-ai/camel-docling/src/generated/java/org/apache/camel/component/docling/DoclingEndpointConfigurer.java, components/camel-ai/camel-docling/src/generated/java/org/apache/camel/component/docling/DoclingEndpointUriFactory.java, components/camel-ai/camel-docling/src/generated/resources/META-INF/org/apache/camel/component/docling/docling.json, components/camel-ai/camel-docling/src/main/docs/docling-component.adoc, components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingConsumer.java, components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingContentExtractor.java, components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingEndpoint.java, components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingProducer.ja
va, components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingConsumerFailureTest.java, components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingConsumerTest.java, components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/integration/DoclingConsumerIT.java
- `camel-jbang-mcp` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-mcp` ← downstream of org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-route-parser` ← downstream of org.apache.camel:camel-route-parser
- `camel-jbang-plugin-tui` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-validate` ← downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-langchain4j-ingest` ← downstream of org.apache.camel:camel-docling
- `camel-launcher-container` ← downstream of org.apache.camel:camel-launcher
- `camel-yaml-dsl-validator` ← downstream of org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` ← downstream of org.apache.camel:camel-yaml-dsl-validator
</details>
<details><summary>Modules with tests skipped (24)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `docs`
- `dummy-component`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
<details><summary>All tested modules (37 modules, 6m 48s total)</summary>
**Total reactor time:** 6m 48s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Launcher | 51.7s | SUCCESS |
| Camel :: JBang :: Plugin :: TUI | 44.4s | SUCCESS |
| Camel :: JBang :: MCP | 41.8s | SUCCESS |
| Camel :: AI :: LangChain4j :: Ingest | 33.2s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 25.2s | SUCCESS |
| Camel :: AI :: Docling | 23.6s | SUCCESS |
| Camel :: YAML DSL :: Validator | 22.4s | SUCCESS |
| Camel :: Component DSL | 21.1s | SUCCESS |
| Camel :: YAML DSL | 20.3s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 16.5s | SUCCESS |
| Camel :: Docs | 15.9s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 13.4s | SUCCESS |
| Camel :: Kamelet Main | 11.2s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 10.8s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 8.7s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 8.1s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 7.2s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 5.3s | SUCCESS |
| Camel :: All Components Sync point | 5.3s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.1s | SUCCESS |
| Camel :: Catalog :: Maven | 3.0s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.7s | SUCCESS |
| Camel :: Assembly | 2.0s | SUCCESS |
| Camel :: Coverage | 1.9s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.5s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.1s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.0s | SUCCESS |
| Camel :: JBang :: Plugin :: Route Parser | 0.9s | SUCCESS |
| Camel :: JBang :: Integration tests | 0.9s | SUCCESS |
| Camel :: Catalog :: Console | 0.9s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.8s | SUCCESS |
| Camel :: JBang :: Main | 0.8s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 0.7s | SUCCESS |
| Camel :: Launcher :: Container | 0.7s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
**Top 20 slowest modules:**
- `Camel :: Launcher` (51.7s)
- `Camel :: JBang :: Plugin :: TUI` (44.4s)
- `Camel :: JBang :: MCP` (41.8s)
- `Camel :: AI :: LangChain4j :: Ingest` (33.2s)
- `Camel :: Catalog :: Camel Catalog` (25.2s)
- `Camel :: AI :: Docling` (23.6s)
- `Camel :: YAML DSL :: Validator` (22.4s)
- `Camel :: Component DSL` (21.1s)
- `Camel :: YAML DSL` (20.3s)
- `Camel :: JBang :: Plugin :: Validate` (16.5s)
- `Camel :: Docs` (15.9s)
- `Camel :: JBang :: Plugin :: Testing` (13.4s)
- `Camel :: Kamelet Main` (11.2s)
- `Camel :: JBang :: Plugin :: Kubernetes` (10.8s)
- `Camel :: YAML DSL :: Deserializers` (8.7s)
- `Camel :: Catalog :: Camel Route Parser` (8.1s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (7.2s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (5.3s)
- `Camel :: All Components Sync point` (5.3s)
- `Camel :: YAML DSL :: Maven Plugins` (3.1s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/37588333311)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-23458: camel-docling - add consumer support for async task completion events [camel]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd commented on code in PR #27477:
URL: https://github.com/apache/camel/pull/27477#discussion_r4204674749
##########
components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingConsumer.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.util.Map;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+import ai.docling.serve.api.convert.response.ConvertDocumentResponse;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.support.ScheduledPollConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Event-driven consumer for docling async conversions.
+ * <p>
+ * It polls the component-wide registry of tasks submitted via {@code SUBMIT_ASYNC_CONVERSION} and, each time a task
+ * completes, emits one exchange whose body is the converted content (the same shape the synchronous {@code CONVERT_*}
+ * operations produce for the configured {@code outputFormat}), carrying the task id in the
+ * {@link DoclingHeaders#TASK_ID} header. A task that failed completes the exchange with the underlying exception, so
+ * the route's {@code onException} machinery applies. The poll interval is the standard scheduled-consumer {@code delay}
+ * option.
+ */
+public class DoclingConsumer extends ScheduledPollConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DoclingConsumer.class);
+
+ public DoclingConsumer(DoclingEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ }
+
+ @Override
+ public DoclingEndpoint getEndpoint() {
+ return (DoclingEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected int poll() throws Exception {
+ Map<String, AsyncTaskEntry> pendingAsyncTasks
+ = ((DoclingComponent) getEndpoint().getComponent()).getPendingAsyncTasks();
+
+ int polled = 0;
+ for (Map.Entry<String, AsyncTaskEntry> entry : pendingAsyncTasks.entrySet()) {
+ String taskId = entry.getKey();
+ AsyncTaskEntry task = entry.getValue();
+ if (!task.getFuture().isDone()) {
+ continue;
+ }
+ // claim the completed task atomically, so a concurrent CHECK_CONVERSION_STATUS or another consumer cannot
+ // emit the same completion twice
+ if (!pendingAsyncTasks.remove(taskId, task)) {
+ continue;
+ }
+
+ polled++;
+ Exchange exchange = createExchange(false);
+ try {
+ exchange.getIn().setHeader(DoclingHeaders.TASK_ID, taskId);
+ emitResult(exchange, taskId, task.getFuture());
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ } finally {
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException(
+ "Error processing docling async task " + taskId, exchange, exchange.getException());
+ }
+ releaseExchange(exchange, false);
+ }
+ }
+ return polled;
+ }
+
+ private void emitResult(Exchange exchange, String taskId, CompletableFuture<ConvertDocumentResponse> future) {
+ try {
+ ConvertDocumentResponse response = future.join();
+ exchange.getIn().setBody(
+ DoclingContentExtractor.extract(response, getEndpoint().getConfiguration().getOutputFormat()));
Review Comment:
Good catch — fixed in 4abae810a1d9. The output format is now resolved once at submit time (the `CamelDoclingOutputFormat` header, else the submitting endpoint's `outputFormat`), stored on `AsyncTaskEntry`, and used for extraction in both the consumer (`emitResult`) and the producer's `checkLocalAsyncTask` — so neither re-derives it from its own endpoint. That closes the pre-existing `checkLocalAsyncTask` gap too, as you noted.
Added `DoclingConsumerTest.extractsWithTheSubmittedFormatNotTheConsumerFormat`: a task submitted with `html` and consumed from a markdown-default `docling:onComplete` endpoint now emits the HTML body instead of an empty string.
_Claude Code on behalf of oscerd_
##########
components/camel-ai/camel-docling/src/main/docs/docling-component.adoc:
##########
@@ -1219,6 +1219,33 @@ YAML::
----
====
+== Consuming asynchronous conversions
+
+Instead of submitting a conversion with `SUBMIT_ASYNC_CONVERSION` and then polling `CHECK_CONVERSION_STATUS` by hand, a
+`docling:` consumer can emit each async conversion event-driven, as soon as it finishes:
+
+[source,java]
+----
+// submit conversions asynchronously; each returns a task id and runs in the background
+from("direct:submit")
+ .to("docling:convert?useDoclingServe=true&operation=SUBMIT_ASYNC_CONVERSION");
+
+// the consumer emits one exchange per completed task (the path, here "onComplete", is just a label)
+from("docling:onComplete?useDoclingServe=true&delay=5000")
+ .to("mock:done");
+----
+
+The consumer scans the tasks submitted via `SUBMIT_ASYNC_CONVERSION` on the same component and emits one exchange as
Review Comment:
Documented in 4abae810a1d9. The async-consumer section now carries a NOTE stating that the consumer drains the whole component's tasks, so it shouldn't be mixed with manual `CHECK_CONVERSION_STATUS` polling on the same component — once the consumer claims a task, a later status check for that id no longer finds it locally and falls back to the server, which doesn't know the local id.
I kept the draining behaviour (it's the point of the consumer) and documented the constraint, rather than tracking consumed ids on the fallback — that would add state for a usage pattern the docs now steer away from. Happy to revisit the code-side option if you'd prefer it.
_Claude Code on behalf of oscerd_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org