You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ji...@apache.org on 2026/09/15 11:17:42 UTC
(camel-quarkus) branch camel-main updated (7053412ca5 -> c11838382c)
This is an automated email from the ASF dual-hosted git repository.
JiriOndrusek pushed a change to branch camel-main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
from 7053412ca5 Fixed javascript, jira, jdbc-grouped test modules
new 0d00739a01 Fixes #9142. Delegate langchain4j-ingest to the camel-langchain4j-ingest component
new c11838382c Gen
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/modules/ROOT/pages/migration-guide/4.0.0.adoc | 20 +
docs/modules/ROOT/pages/migration-guide/index.adoc | 1 +
.../reference/extensions/langchain4j-ingest.adoc | 131 ++++++-
.../deployment/Langchain4jIngestProcessor.java | 104 ++++-
...figTest.java => IngestBatchSizeConfigTest.java} | 9 +-
...> IngestBuilderParserMissingComponentTest.java} | 14 +-
...t.java => IngestMaxDocumentSizeConfigTest.java} | 9 +-
....java => IngestParserComponentMissingTest.java} | 12 +-
...ava => IngestPipelineParserValidationTest.java} | 31 +-
...onfigTest.java => IngestUnknownParserTest.java} | 9 +-
.../ingest/deployment/IngestUnnamedBeansTest.java | 2 +-
extensions/langchain4j-ingest/runtime/pom.xml | 4 +
.../runtime/src/main/doc/usage.adoc | 92 ++++-
.../langchain4j/ingest/IngestBuildTimeConfig.java | 42 ++
.../ingest/IngestComponentPresence.java | 11 +
.../langchain4j/ingest/IngestHeaders.java | 7 +
.../langchain4j/ingest/IngestPipeline.java | 79 ++++
.../ingest/IngestPipelineDefinition.java | 351 +++++++++++++++++
.../ingest/IngestPipelineRouteBuilder.java | 423 +++++++++++++++++++++
.../component/langchain4j/ingest/IngestRoutes.java | 305 +++++----------
.../langchain4j/ingest/core/IngestResult.java | 39 --
.../langchain4j/ingest/core/IngestService.java | 95 -----
integration-tests/langchain4j-ingest/pom.xml | 45 +++
.../langchain4j/ingest/it/IngestItProducers.java | 28 ++
.../langchain4j/ingest/it/IngestResource.java | 66 +++-
.../src/main/resources/application.properties | 42 ++
.../ingest/it/IngestDoclingTestResource.java | 87 +++++
...afkaIT.java => Langchain4jIngestDoclingIT.java} | 2 +-
...Test.java => Langchain4jIngestDoclingTest.java} | 41 +-
...jIngestIT.java => Langchain4jIngestTikaIT.java} | 2 +-
.../ingest/it/Langchain4jIngestTikaTest.java | 184 +++++++++
.../src/test/resources/notice.html | 6 +-
.../langchain4j-ingest/src/test/resources/pump.pdf | Bin 0 -> 850 bytes
.../src/test/resources/valve.pdf | Bin 0 -> 848 bytes
pom.xml | 1 +
poms/bom/pom.xml | 20 +
poms/bom/src/main/generated/flattened-full-pom.xml | 16 +
.../src/main/generated/flattened-reduced-pom.xml | 16 +
.../generated/flattened-reduced-verbose-pom.xml | 16 +
39 files changed, 1904 insertions(+), 458 deletions(-)
create mode 100644 docs/modules/ROOT/pages/migration-guide/4.0.0.adoc
copy extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/{IngestSplitterBoundsConfigTest.java => IngestBatchSizeConfigTest.java} (81%)
copy extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/{IngestBuilderMissingComponentTest.java => IngestBuilderParserMissingComponentTest.java} (79%)
copy extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/{IngestSplitterBoundsConfigTest.java => IngestMaxDocumentSizeConfigTest.java} (81%)
copy extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/{IngestUnknownComponentTest.java => IngestParserComponentMissingTest.java} (78%)
copy extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/{IngestBuilderBlankNameTest.java => IngestPipelineParserValidationTest.java} (57%)
copy extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/{IngestSplitterBoundsConfigTest.java => IngestUnknownParserTest.java} (81%)
create mode 100644 extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineDefinition.java
create mode 100644 extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java
delete mode 100644 extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java
delete mode 100644 extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestService.java
create mode 100644 integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestDoclingTestResource.java
copy integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/{Langchain4jIngestKafkaIT.java => Langchain4jIngestDoclingIT.java} (92%)
copy integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/{Langchain4jIngestS3Test.java => Langchain4jIngestDoclingTest.java} (51%)
copy integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/{Langchain4jIngestIT.java => Langchain4jIngestTikaIT.java} (93%)
create mode 100644 integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaTest.java
copy integration-test-groups/xml/native/xslt-classpath/src/test/resources/test.html => integration-tests/langchain4j-ingest/src/test/resources/notice.html (81%)
create mode 100644 integration-tests/langchain4j-ingest/src/test/resources/pump.pdf
create mode 100644 integration-tests/langchain4j-ingest/src/test/resources/valve.pdf
(camel-quarkus) 02/02: Gen
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JiriOndrusek pushed a commit to branch camel-main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
commit c11838382cdff5f9c7fac19de3fb7bab77b6ea83
Author: Jiri Ondrusek <on...@gmail.com>
AuthorDate: Tue Sep 15 13:16:31 2026 +0200
Gen
---
.../ROOT/pages/reference/extensions/langchain4j-ingest.adoc | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
index a8ee2588cf..80d1f529e6 100644
--- a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
+++ b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
@@ -19,6 +19,13 @@ endif::[]
Ingest documents into a LangChain4j EmbeddingStore: split, embed and store the message body.
+[id="extensions-langchain4j-ingest-whats-inside"]
+== What's inside
+
+* xref:{cq-camel-components}::langchain4j-ingest-component.adoc[LangChain4j Ingest component], URI syntax: `langchain4j-ingest:pipelineName`
+
+Please refer to the above link for usage and configuration details.
+
[id="extensions-langchain4j-ingest-maven-coordinates"]
== Maven coordinates
(camel-quarkus) 01/02: Fixes #9142. Delegate langchain4j-ingest to the camel-langchain4j-ingest component
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JiriOndrusek pushed a commit to branch camel-main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
commit 0d00739a01eb03f4c73012a5f7085730517c0cb8
Author: Jiří Ondrušek <on...@gmail.com>
AuthorDate: Thu Sep 10 10:28:14 2026 +0000
Fixes #9142. Delegate langchain4j-ingest to the camel-langchain4j-ingest component
The extension's own engine is deleted; the routes now end in the
producer of the camel-langchain4j-ingest component introduced in Camel
4.23.0 (CAMEL-24621), while the Quarkus declaration surface (config
model, @Ingest API, validation, CDI resolution, native support) stays.
New options exposed from the component: parser (tika/docling, fixes
#9076), embedding-batch-size, document-splitter and max-document-size.
Behaviour changes (route ids, producer-worded missing-id failure,
IngestResult import) are documented in the migration guide; the
CamelIngestDocumentId header default stays pinned.
Verified: 30 deployment test suites, 18 JVM and 18 native integration
tests green.
Co-Authored-By: Claude Fable 5 <no...@anthropic.com>
---
docs/modules/ROOT/pages/migration-guide/4.0.0.adoc | 20 +
docs/modules/ROOT/pages/migration-guide/index.adoc | 1 +
.../reference/extensions/langchain4j-ingest.adoc | 138 +++++--
.../deployment/Langchain4jIngestProcessor.java | 104 ++++-
.../deployment/IngestBatchSizeConfigTest.java | 39 ++
.../IngestBuilderParserMissingComponentTest.java | 57 +++
.../IngestMaxDocumentSizeConfigTest.java | 39 ++
.../IngestParserComponentMissingTest.java | 43 +++
.../IngestPipelineParserValidationTest.java | 42 ++
.../ingest/deployment/IngestUnknownParserTest.java | 39 ++
.../ingest/deployment/IngestUnnamedBeansTest.java | 2 +-
extensions/langchain4j-ingest/runtime/pom.xml | 4 +
.../runtime/src/main/doc/usage.adoc | 92 ++++-
.../langchain4j/ingest/IngestBuildTimeConfig.java | 42 ++
.../ingest/IngestComponentPresence.java | 11 +
.../langchain4j/ingest/IngestHeaders.java | 7 +
.../langchain4j/ingest/IngestPipeline.java | 79 ++++
.../ingest/IngestPipelineDefinition.java | 351 +++++++++++++++++
.../ingest/IngestPipelineRouteBuilder.java | 423 +++++++++++++++++++++
.../component/langchain4j/ingest/IngestRoutes.java | 305 +++++----------
.../langchain4j/ingest/core/IngestService.java | 95 -----
integration-tests/langchain4j-ingest/pom.xml | 45 +++
.../langchain4j/ingest/it/IngestItProducers.java | 28 ++
.../langchain4j/ingest/it/IngestResource.java | 66 +++-
.../src/main/resources/application.properties | 42 ++
.../ingest/it/IngestDoclingTestResource.java | 87 +++++
.../ingest/it/Langchain4jIngestDoclingIT.java | 24 +-
.../ingest/it/Langchain4jIngestDoclingTest.java | 59 +++
.../ingest/it/Langchain4jIngestTikaIT.java | 18 +-
.../ingest/it/Langchain4jIngestTikaTest.java | 184 +++++++++
.../src/test/resources/notice.html | 22 ++
.../langchain4j-ingest/src/test/resources/pump.pdf | Bin 0 -> 850 bytes
.../src/test/resources/valve.pdf | Bin 0 -> 848 bytes
pom.xml | 1 +
poms/bom/pom.xml | 20 +
poms/bom/src/main/generated/flattened-full-pom.xml | 16 +
.../src/main/generated/flattened-reduced-pom.xml | 16 +
.../generated/flattened-reduced-verbose-pom.xml | 16 +
38 files changed, 2193 insertions(+), 384 deletions(-)
diff --git a/docs/modules/ROOT/pages/migration-guide/4.0.0.adoc b/docs/modules/ROOT/pages/migration-guide/4.0.0.adoc
new file mode 100644
index 0000000000..37f2ed4596
--- /dev/null
+++ b/docs/modules/ROOT/pages/migration-guide/4.0.0.adoc
@@ -0,0 +1,20 @@
+= Camel Quarkus 4.0.0 Migration Guide
+
+The following guide outlines how to adapt your code to changes that were made in Camel Quarkus 4.0.0.
+
+== LangChain4j Ingest extension changes
+
+=== The ingestion engine moved to the camel-langchain4j-ingest component
+
+The extension now drives the `camel-langchain4j-ingest` component instead of carrying its own implementation.
+The declaration surface is unchanged — the same `quarkus.camel.langchain4j.ingest.*` properties and the same `@Ingest` builder API — and grew new options (`parser`, `embedding-batch-size`, `document-splitter`, `max-document-size`).
+Java code reading the request-reply outcome as `IngestResult` now imports `org.apache.camel.component.langchain4j.ingest.IngestResult` — the extension's `org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestResult` is gone with the engine, so this surfaces at compile time.
+
+A pipeline's route id is now `langchain4j-ingest-<pipeline-name>` rather than `camel-quarkus-langchain4j-ingest-<pipeline-name>`.
+Tooling that addresses the routes by id — JMX, the management endpoints, `camel.main.*` route policies keyed on ids — must use the new names.
+The prefix is the component's scheme name, so it follows the component.
+
+A consumer-fed delivery whose document id resolves to nothing still fails the exchange, but the message now comes from the component and names its `documentIdHeader` endpoint option instead of the `quarkus.camel.langchain4j.ingest.<name>.source.document-id` property.
+Code matching on the message text must be adjusted; matching on the exception type is unaffected.
+
+The default document-id header on consumer-fed pipelines remains `CamelIngestDocumentId` as released — the extension pins it over the component's own `CamelLangChain4jIngestDocumentId` default — so routes that set the header keep working unchanged.
diff --git a/docs/modules/ROOT/pages/migration-guide/index.adoc b/docs/modules/ROOT/pages/migration-guide/index.adoc
index 150b2340bc..e5104b1953 100644
--- a/docs/modules/ROOT/pages/migration-guide/index.adoc
+++ b/docs/modules/ROOT/pages/migration-guide/index.adoc
@@ -4,6 +4,7 @@ We do frequent releases, a release almost every month, and even though we strive
Listed here are guides on how to migrate between major versions and anything of significance to watch for when upgrading from minor versions.
+* xref:migration-guide/4.0.0.adoc[Camel Quarkus 3.40.x to Camel Quarkus 4.0.0 migration guide]
* xref:migration-guide/3.40.0.adoc[Camel Quarkus 3.39.x to Camel Quarkus 3.40.0 migration guide]
* xref:migration-guide/3.39.0.adoc[Camel Quarkus 3.38.x to Camel Quarkus 3.39.0 migration guide]
* xref:migration-guide/3.38.0.adoc[Camel Quarkus 3.36.x to Camel Quarkus 3.38.0 migration guide]
diff --git a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
index 4b66b6a908..a8ee2588cf 100644
--- a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
+++ b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
@@ -19,13 +19,6 @@ endif::[]
Ingest documents into a LangChain4j EmbeddingStore: split, embed and store the message body.
-[id="extensions-langchain4j-ingest-whats-inside"]
-== What's inside
-
-* xref:{cq-camel-components}::langchain4j-ingest-component.adoc[LangChain4j Ingest component], URI syntax: `langchain4j-ingest:pipelineName`
-
-Please refer to the above link for usage and configuration details.
-
[id="extensions-langchain4j-ingest-maven-coordinates"]
== Maven coordinates
@@ -49,7 +42,8 @@ endif::[]
[id="extensions-langchain4j-ingest-usage-ingesting-a-directory"]
=== Ingesting a directory
-Point a pipeline at a folder and the files in it become a knowledge base. No route, no ingestion code:
+Point a pipeline at a folder and the files in it become a knowledge base.
+No route, no ingestion code:
[source,properties]
----
@@ -58,17 +52,26 @@ quarkus.camel.langchain4j.ingest.products.embedding-store=products
quarkus.camel.langchain4j.ingest.products.embedding-model=my-model
----
-Each file is read as UTF-8 text — there is no format parsing, so convert a PDF or DOCX before it reaches the pipeline — split into overlapping segments (`max-segment-size`, `max-overlap-size`), embedded in batches and written to the store; a document is held in memory whole while it is split. `embedding-store` and `embedding-model` name CDI beans and may be omitted when the application has exactly one of each. Every segment carries `camel_ingest_pipeline` and `camel_ingest_document_id` m [...]
+Each file is read as UTF-8 text — unless a `parser` turns binary formats into text first, see below — split into overlapping segments (`max-segment-size`, `max-overlap-size`; or a `DocumentSplitter` CDI bean named by `document-splitter`, which then replaces the sizes), embedded in batches (`embedding-batch-size`, 32 by default) and written to the store; a document is held in memory whole while it is split, so `max-document-size` can cap it in characters — 0, the default, means no limit, [...]
+`embedding-store` and `embedding-model` name CDI beans and may be omitted when the application has exactly one of each.
+Every segment carries `camel_ingest_pipeline` and `camel_ingest_document_id` metadata, so retrieval can cite which document an answer came from (up to 3.39.x the keys were `camel_quarkus_pipeline` and `camel_quarkus_document_id` — see the migration guide).
+Apart from `enabled` and the `source.*` settings shown here, properties are fixed at build time.
+A pipeline declared through runtime properties alone — nothing but a `source.directory`, say — is invisible to build-time validation; its checks, including the clash with an equally named `@Ingest` pipeline, report at startup instead.
[NOTE]
====
-This experimental extension keeps no record of what it wrote: an edited document re-ingests on top of its old segments, and removing a document from the source removes nothing from the store. Replace and delete arrive with the synchronising engine in a later release. What is remembered is which documents were already ingested — the idempotent repository below.
+This experimental extension keeps no record of what it wrote: an edited document re-ingests on top of its old segments, and removing a document from the source removes nothing from the store.
+Replace and delete arrive with the synchronising engine in a later release.
+What is remembered is which documents were already ingested — the idempotent repository below.
====
[id="extensions-langchain4j-ingest-usage-the-idempotent-repository"]
=== The idempotent repository
-A directory pipeline registers what it ingested, keyed on the file's path, modification time and size: unchanged files are skipped, edited files re-ingest (their previous segments remain, see the note above). The default register is in-memory (100,000 keys) and is there for the poll loop, not for restarts: the consumer re-scans the directory on every poll, so without a register a running application would re-embed the whole directory every few seconds. A restart loses it and re-ingests t [...]
+A directory pipeline registers what it ingested, keyed on the file's path, modification time and size: unchanged files are skipped, edited files re-ingest (their previous segments remain, see the note above).
+The default register is in-memory (100,000 keys) and is there for the poll loop, not for restarts: the consumer re-scans the directory on every poll, so without a register a running application would re-embed the whole directory every few seconds.
+A restart loses it and re-ingests the directory once; a persistent repository makes restarts free.
+`source.idempotent-repository` names an `IdempotentRepository` bean to use instead:
[source,properties]
----
@@ -77,7 +80,8 @@ quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=productsR
The bean can be provided three ways.
-*Auto-created*: an in-memory register (100,000 keys) is created and bound under the configured name — no bean definition needed, still lost on restart. If a bean with the name already exists, the existing bean wins:
+*Auto-created*: an in-memory register (100,000 keys) is created and bound under the configured name — no bean definition needed, still lost on restart.
+If a bean with the name already exists, the existing bean wins:
[source,properties]
----
@@ -85,7 +89,8 @@ quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=productsR
quarkus.camel.langchain4j.ingest.products.source.idempotent-repository-auto-create=true
----
-*Defined in properties* through Camel's `camel.beans.` syntax — here a file-backed register that survives restarts (camel-core, no extra dependency). Use a name without dashes; Camel normalises dashed `camel.beans.` keys to camelCase:
+*Defined in properties* through Camel's `camel.beans.` syntax — here a file-backed register that survives restarts (camel-core, no extra dependency).
+Use a name without dashes; Camel normalises dashed `camel.beans.` keys to camelCase:
[source,properties]
----
@@ -96,9 +101,12 @@ camel.beans.productsRegister.cacheSize=100000
camel.beans.productsRegister.maxFileStoreSize=33554432
----
-`#class:` beans are created reflectively. For native mode the extension registers the camel-core repositories shown here and every `IdempotentRepository` implementation found in the Jandex index — application classes always, component-provided ones through their jar's index. A class from a jar without an index needs `@RegisterForReflection(targets = ...)` or a `quarkus.index-dependency.*` entry.
+`#class:` beans are created reflectively.
+For native mode the extension registers the camel-core repositories shown here and every `IdempotentRepository` implementation found in the Jandex index — application classes always, component-provided ones through their jar's index.
+A class from a jar without an index needs `@RegisterForReflection(targets = ...)` or a `quarkus.index-dependency.*` entry.
-*Defined as a CDI producer* — needed when construction takes other beans. A JDBC register also deduplicates across instances: reliably for consumer-fed pipelines keyed on the document id, for directory pipelines only when all instances see identical paths and modification times:
+*Defined as a CDI producer* — needed when construction takes other beans.
+A JDBC register also deduplicates across instances: reliably for consumer-fed pipelines keyed on the document id, for directory pipelines only when all instances see identical paths and modification times:
[source,java]
----
@@ -110,16 +118,52 @@ IdempotentRepository productsRegister(DataSource dataSource) {
More repository types — Caffeine, Infinispan, MongoDB, Cassandra, Hazelcast, Kafka and others, each provided by its component's extension — along with the pattern's details are covered in the xref:{cq-camel-components}:eips:idempotentConsumer-eip.adoc[Idempotent Consumer EIP guide].
-Sizing. An in-memory register smaller than the directory evicts keys and re-ingests those files during normal operation — hence the 100,000 default. `FileIdempotentRepository` falls back to its file store on a cache miss, so `cacheSize` is a performance setting; the correctness limit is `maxFileStoreSize`: the default of about 1 MB holds roughly 10–17 thousand of these keys, beyond it the oldest 1000 are dropped with a warning. A JDBC register has no size limit. Every edit adds a new key [...]
+Sizing.
+An in-memory register smaller than the directory evicts keys and re-ingests those files during normal operation — hence the 100,000 default.
+`FileIdempotentRepository` falls back to its file store on a cache miss, so `cacheSize` is a performance setting; the correctness limit is `maxFileStoreSize`: the default of about 1 MB holds roughly 10–17 thousand of these keys, beyond it the oldest 1000 are dropped with a warning.
+A JDBC register has no size limit.
+Every edit adds a new key and old keys are never removed, so a persistent register grows under churn until the synchronising engine arrives.
+There is no size option; size the bean.
+Use one register per pipeline — keys are source-derived, and pipelines sharing a register over the same directory skip each other's work.
+
+The register cannot be switched off: the consumer leaves files in place (`noop`), so without a register every poll would re-ingest the directory.
+To force a full re-ingest, restart (default register) or clear the persistent one.
+A pipeline that should consume its source — move or delete files after ingestion — uses `source.uri` (for example `file:/inbox?delete=true&idempotent=false` with `source.document-id=CamelFileName`), where all consumer options are available.
-The register cannot be switched off: the consumer leaves files in place (`noop`), so without a register every poll would re-ingest the directory. To force a full re-ingest, restart (default register) or clear the persistent one. A pipeline that should consume its source — move or delete files after ingestion — uses `source.uri` (for example `file:/inbox?delete=true&idempotent=false` with `source.document-id=CamelFileName`), where all consumer options are available.
+On a consumer-fed pipeline (`source.uri` or `@Ingest`), a configured register deduplicates deliveries by document id: a redelivered record or re-listed object ingests once, and a request-reply caller receives `skipped` instead of `ingested`.
+First write wins per id — an update with the same id is skipped, not replaced; streams that carry updates need a version-aware `source.document-id`.
+Only a delivery that wrote segments claims its id: a blank document answers `empty` and releases the claim, so an object created empty and populated later under the same id still ingests the content.
+With eager idempotency, a duplicate racing an in-flight first delivery is answered `skipped` even if that delivery then fails and releases the key.
-On a consumer-fed pipeline (`source.uri` or `@Ingest`), a configured register deduplicates deliveries by document id: a redelivered record or re-listed object ingests once, and a request-reply caller receives `skipped` instead of `ingested`. First write wins per id — an update with the same id is skipped, not replaced; streams that carry updates need a version-aware `source.document-id`. Only a delivery that wrote segments claims its id: a blank document answers `empty` and releases the [...]
+[id="extensions-langchain4j-ingest-usage-parsing-pdfs-and-office-documents"]
+=== Parsing PDFs and office documents
+
+Setting `parser` puts a document parser between the consumer and the splitter:
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.reports.source.directory=/var/data/reports
+quarkus.camel.langchain4j.ingest.reports.parser=tika
+----
+
+`tika` extracts plain text in-process with Apache Tika.
+It needs the `camel-quarkus-tika` extension, plus the Tika parser module for each format the pipeline should read — `org.apache.tika:tika-parser-pdf-module` for PDF, say.
+A format without its module parses to nothing and the document is skipped as empty (logged as a warning); so is a scanned, image-only PDF, since Tika does no OCR here.
+On a directory pipeline an empty parse still commits the file's key in the duplicate register, so the file is not retried until it changes — after adding a missing parser module, force a re-ingest (restart with the default register, or clear the persistent one).
+A consumer-fed pipeline releases the claim of an empty delivery instead.
+
+`docling` converts documents to structure-preserving markdown through a https://github.com/docling-project/docling-serve[Docling Serve] instance, OCR included.
+It needs the `camel-quarkus-docling` extension and the instance's address in `camel.component.docling.docling-serve-url`.
+
+A missing parser extension fails the build naming the artifact, the same way a missing consumer component does.
+The Java twin is `IngestPipeline.parser("tika")`, and a consumer-fed pipeline parses the same way — the payload the consumer delivers goes to the parser first.
+With a parser, `max-document-size` also rejects a raw payload larger than the cap (in bytes) before it reaches the parser, on top of capping the extracted text (in characters).
[id="extensions-langchain4j-ingest-usage-other-sources-declared-in-java"]
=== Other sources, declared in Java
-Any Camel consumer can feed a pipeline — the roughly 300 components, each with its own options and its own documentation. Such a pipeline is declared in Java with `@Ingest` and the Camel Endpoint DSL:
+Any Camel consumer can feed a pipeline — the roughly 300 components, each with its own options and its own documentation.
+Such a pipeline is declared in Java with `@Ingest` and the Camel Endpoint DSL:
[source,java]
----
@@ -131,13 +175,20 @@ IngestPipeline events() {
}
----
-Typing `dsl.` lists a factory for every component, each completing its own typed options. The method runs once at startup; it must return `IngestPipeline`, take no parameters and use a name no configured pipeline uses. `enabled=false` in configuration switches a Java-declared pipeline off. A component missing from the classpath fails at startup with an error naming the extension artifact that provides it — the DSL compiles regardless, since its factories all ship in one artifact.
+Typing `dsl.` lists a factory for every component, each completing its own typed options.
+The method runs once at startup; it must return `IngestPipeline`, take no parameters and use a name no configured pipeline uses.
+`enabled=false` in configuration switches a Java-declared pipeline off.
+A component missing from the classpath fails at startup with an error naming the extension artifact that provides it — the DSL compiles regardless, since its factories all ship in one artifact.
-Ingestion needs a stable id per document, and where it lives is the consumer's business: `documentId` names the header — the record key `CamelKafkaKey` above, `CamelAwsS3Key` for S3 — or gives a simple-language expression. Without it, the pipeline expects the `CamelIngestDocumentId` header and fails the exchange when it is absent. Mind each component's own defaults, too: the `aws2-s3` consumer *deletes objects after reading them* unless `deleteAfterRead(false)` is set — a knowledge base [...]
+Ingestion needs a stable id per document, and where it lives is the consumer's business: `documentId` names the header — the record key `CamelKafkaKey` above, `CamelAwsS3Key` for S3 — or gives a simple-language expression.
+Without it, the pipeline expects the `CamelIngestDocumentId` header and fails the exchange when it is absent.
+Mind each component's own defaults, too: the `aws2-s3` consumer *deletes objects after reading them* unless `deleteAfterRead(false)` is set — a knowledge base reads its source, it does not consume it.
After such a pipeline ingests a document, the exchange body is replaced with the `IngestResult`, so a request-reply caller receives the outcome of its call.
-The same pipeline can be declared purely in properties: `source.uri` takes the consumer URI as written, `source.document-id` the header name — or a simple-language expression written `+$simple{...}+`, the one form MicroProfile Config leaves untouched. The URI is fixed at build time by design — a runtime-overridable consumer URI would be arbitrary component invocation — while property placeholders inside it still resolve at startup, keeping credentials and endpoints runtime configuration. [...]
+The same pipeline can be declared purely in properties: `source.uri` takes the consumer URI as written, `source.document-id` the header name — or a simple-language expression written `+$simple{...}+`, the one form MicroProfile Config leaves untouched.
+The URI is fixed at build time by design — a runtime-overridable consumer URI would be arbitrary component invocation — while property placeholders inside it still resolve at startup, keeping credentials and endpoints runtime configuration.
+Treat runtime configuration as the trust boundary it is: the directory and the id expression decide what the process reads into an often external store.
[source,properties]
----
@@ -149,7 +200,10 @@ quarkus.camel.langchain4j.ingest.s3docs.embedding-store=products
[id="extensions-langchain4j-ingest-usage-when-ingestion-fails"]
=== When ingestion fails
-A failure while splitting, embedding or storing — a rate-limited model, an unreachable store — propagates to the consumer; there is no dead-letter channel in this increment. For a directory pipeline the file stays where it is and is retried on the next poll, because the duplicate-protection key is only committed on success — which also means a permanently failing file is retried forever, loudly. For a consumer-fed pipeline the component's own error handling applies: a request-reply calle [...]
+A failure while parsing, splitting, embedding or storing — an encrypted or corrupt document, an unreachable docling-serve, a rate-limited model, an unreachable store — propagates to the consumer; there is no dead-letter channel in this increment.
+For a directory pipeline the file stays where it is and is retried on the next poll, because the duplicate-protection key is only committed on success — which also means a permanently failing file is retried forever, loudly, on every poll: with a parser that includes a document the parser cannot read, so remove or fix such a file, or declare the pipeline with `source.uri` and the file consumer's `moveFailed` option to quarantine failures instead.
+For a consumer-fed pipeline the component's own error handling applies: a request-reply caller receives the exception, while a Kafka consumer with default settings logs the failure and commits the offset, so the record is *dropped* — and since this engine keeps no record either, nothing remembers it.
+A record whose configured `document-id` resolves to nothing (a Kafka record without a key, say) fails the same way, one exchange at a time.
[id="extensions-langchain4j-ingest-quarkus-langchain4j-bom"]
@@ -201,6 +255,17 @@ resolve at startup, so credentials and endpoints remain runtime configuration.
| `string`
|
+a|icon:lock[title=Fixed at build time] [[quarkus-camel-langchain4j-ingest-pipeline-name-parser]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-parser[quarkus.camel.langchain4j.ingest."pipeline-name".parser]`
+
+How a consumed payload becomes text before it is split: `tika` extracts plain text
+from PDF, office and similar formats in-process, `docling` converts to
+structure-preserving markdown through a Docling Serve instance (configure it with
+`camel.component.docling.*`). When not set, the payload is read as text as-is. Each
+value needs its extension on the classpath: `camel-quarkus-tika` or
+`camel-quarkus-docling`.
+| `string`
+|
+
a|icon:lock[title=Fixed at build time] [[quarkus-camel-langchain4j-ingest-pipeline-name-embedding-store]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-embedding-store[quarkus.camel.langchain4j.ingest."pipeline-name".embedding-store]`
Name of the `EmbeddingStore` bean to write to. When not set, the only one present is
@@ -228,6 +293,33 @@ sentence split across a boundary retrievable from either side.
| `int`
| `50`
+a|icon:lock[title=Fixed at build time] [[quarkus-camel-langchain4j-ingest-pipeline-name-embedding-batch-size]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-embedding-batch-size[quarkus.camel.langchain4j.ingest."pipeline-name".embedding-batch-size]`
+
+How many segments are embedded per request to the embedding model. Providers with
+generous per-request limits ingest large documents faster with a bigger batch; a batch
+carries at most `embedding-batch-size` × `max-segment-size` characters, so tune the two
+together against the provider's token limits.
+| `int`
+| `32`
+
+a|icon:lock[title=Fixed at build time] [[quarkus-camel-langchain4j-ingest-pipeline-name-max-document-size]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-max-document-size[quarkus.camel.langchain4j.ingest."pipeline-name".max-document-size]`
+
+Maximum size of one document in characters, applied to the text about to be split;
+0, the default, means no limit. The pipeline holds a document in memory whole, so the
+cap is the protection against oversized — on a consumer-fed pipeline, attacker-sized —
+payloads. An oversized document fails the exchange cleanly.
+| `int`
+| `0`
+
+a|icon:lock[title=Fixed at build time] [[quarkus-camel-langchain4j-ingest-pipeline-name-document-splitter]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-document-splitter[quarkus.camel.langchain4j.ingest."pipeline-name".document-splitter]`
+
+Name of the `DocumentSplitter` bean replacing the default recursive splitting;
+`max-segment-size` and `max-overlap-size` are then ignored. Looked up by name only —
+an application may hold unrelated splitters. Segments returned without the identity
+metadata are re-stamped, so a custom splitter cannot break citation.
+| `string`
+|
+
a| [[quarkus-camel-langchain4j-ingest-pipeline-name-enabled]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-enabled[quarkus.camel.langchain4j.ingest."pipeline-name".enabled]`
Whether this pipeline starts. Useful to switch ingestion off in dev mode.
diff --git a/extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java b/extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java
index 3f1c1148ba..51bded4359 100644
--- a/extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java
+++ b/extensions/langchain4j-ingest/deployment/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/Langchain4jIngestProcessor.java
@@ -28,6 +28,7 @@ import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
import io.quarkus.arc.deployment.SyntheticBeansRuntimeInitBuildItem;
import io.quarkus.arc.deployment.ValidationPhaseBuildItem.ValidationErrorBuildItem;
+import io.quarkus.bootstrap.classloading.QuarkusClassLoader;
import io.quarkus.deployment.annotations.BuildProducer;
import io.quarkus.deployment.annotations.BuildStep;
import io.quarkus.deployment.annotations.Consume;
@@ -36,7 +37,11 @@ import io.quarkus.deployment.annotations.Record;
import io.quarkus.deployment.builditem.ApplicationArchivesBuildItem;
import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.ExcludeConfigBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceDirectoryBuildItem;
import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.RuntimeInitializedClassBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.RuntimeInitializedPackageBuildItem;
import io.quarkus.runtime.configuration.ConfigurationException;
import jakarta.inject.Singleton;
import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
@@ -73,6 +78,72 @@ class Langchain4jIngestProcessor {
.build();
}
+ /**
+ * The documented PDF recipe adds {@code tika-parser-pdf-module}, which drags
+ * {@code jaxb-runtime} → {@code angus-activation} into the image. Angus' own GraalVM
+ * feature then reflects over every registered data-content handler and crashes on
+ * BouncyCastle's S/MIME handlers ({@code bcjmail}'s mailcap references
+ * {@code jakarta.mail.Part}, and jakarta.mail is not on the classpath). The feature only
+ * registers mailcap handlers, which nothing on the ingest path uses, so its registration
+ * is excluded — but only in the constellation that crashes: PDFBox present (the recipe) and
+ * jakarta.mail absent. An application that uses mail and angus for real keeps its feature.
+ *
+ * TODO: move to camel-quarkus-tika, see #9087.
+ */
+ @BuildStep
+ void excludeAngusActivationFeature(BuildProducer<ExcludeConfigBuildItem> excludeConfig) {
+ if (!QuarkusClassLoader.isClassPresentAtRuntime("org.apache.pdfbox.pdmodel.PDDocument")
+ || QuarkusClassLoader.isClassPresentAtRuntime("jakarta.mail.Part")) {
+ return;
+ }
+ excludeConfig.produce(new ExcludeConfigBuildItem("org\\.eclipse\\.angus\\.angus-activation-.*\\.jar",
+ "/META-INF/native-image/org.eclipse.angus/angus-activation/native-image.properties"));
+ }
+
+ /**
+ * The same recipe brings PDFBox itself. Native image needs its AWT-touching classes
+ * initialized at run time and its resource tree — font metrics, glyph lists, ICC profile —
+ * embedded. The camel-quarkus-pdf extension embeds the same files as a hand-listed constant;
+ * this registers the {@code org/apache/pdfbox/resources} directory wholesale instead, so the
+ * set cannot drift when a PDFBox upgrade adds or renames a resource. The runtime-init side is
+ * widened past camel-pdf to the rendering and graphics packages Tika's text extraction
+ * reaches. Produced only when PDFBox is on the classpath.
+ *
+ * TODO: move to camel-quarkus-tika (#9087) and align camel-quarkus-pdf with it (#9091).
+ */
+ @BuildStep
+ void pdfboxInNative(
+ BuildProducer<RuntimeInitializedClassBuildItem> runtimeInitialized,
+ BuildProducer<RuntimeInitializedPackageBuildItem> runtimeInitializedPackages,
+ BuildProducer<NativeImageResourceDirectoryBuildItem> nativeResourceDirectories,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+ if (!QuarkusClassLoader.isClassPresentAtRuntime("org.apache.pdfbox.pdmodel.PDDocument")) {
+ return;
+ }
+ for (String className : new String[] {
+ "org.apache.pdfbox.pdmodel.font.PDType1Font",
+ "org.apache.pdfbox.pdmodel.PDDocument",
+ "org.apache.pdfbox.pdmodel.encryption.StandardSecurityHandler" }) {
+ runtimeInitialized.produce(new RuntimeInitializedClassBuildItem(className));
+ }
+ // text extraction reaches AWT-holding statics across the rendering and graphics trees
+ // (SoftMask's DirectColorModel, the CIE color spaces' ICC_ColorSpace), and Tika's own
+ // PDF classes embed them in enum constants (PDFParserConfig$TikaImageType wraps
+ // rendering.ImageType), so the packages are deferred wholesale rather than chasing one
+ // class at a time
+ for (String packageName : new String[] {
+ "org.apache.pdfbox.rendering",
+ "org.apache.pdfbox.pdmodel.graphics",
+ "org.apache.tika.parser.pdf" }) {
+ runtimeInitializedPackages.produce(new RuntimeInitializedPackageBuildItem(packageName));
+ }
+ reflectiveClasses.produce(ReflectiveClassBuildItem
+ .builder("org.apache.pdfbox.pdmodel.encryption.StandardSecurityHandler",
+ "org.apache.pdfbox.pdmodel.documentinterchange.logicalstructure.PDParentTreeValue")
+ .constructors().methods().build());
+ nativeResourceDirectories.produce(new NativeImageResourceDirectoryBuildItem("org/apache/pdfbox/resources"));
+ }
+
/**
* {@code #class:} beans are instantiated reflectively, which native mode allows only for
* registered classes: the two camel-core repositories the documentation recommends, plus
@@ -205,8 +276,9 @@ class Langchain4jIngestProcessor {
/**
* A pipeline whose consumer URI names a component that is not on the classpath stops the
- * build, naming the artifact that fixes it rather than failing at startup. Only configured URIs
- * can be checked: a builder-declared pipeline composes its URI at startup, where the pre-start
+ * build, naming the artifact that fixes it rather than failing at startup. The same goes for
+ * the {@code parser} value and its component. Only configured pipelines can be checked: a
+ * builder-declared pipeline composes its URI and parser at startup, where the pre-start
* task above applies the same hint. Component services are REGISTRY-destination, so they are
* read from the application archives directly — they never appear among the DISCOVERY
* {@code CamelServiceBuildItem}s.
@@ -222,6 +294,22 @@ class Langchain4jIngestProcessor {
.collect(Collectors.toSet());
for (Map.Entry<String, IngestBuildTimeConfig.PipelineBuildTimeConfig> entry : config.pipelines().entrySet()) {
+ String parser = entry.getValue().parser().orElse(null);
+ if (parser != null) {
+ if (!IngestPipeline.SUPPORTED_PARSERS.contains(parser)) {
+ validationErrors.produce(new ValidationErrorBuildItem(new ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "' sets parser '" + parser
+ + "'. Supported parsers: "
+ + IngestPipeline.SUPPORTED_PARSERS.stream().sorted()
+ .collect(Collectors.joining(", ")))));
+ } else if (!components.contains(parser)) {
+ validationErrors.produce(new ValidationErrorBuildItem(new ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "' parses with '" + parser
+ + "', but the Camel component '" + parser + "' is not on the classpath. "
+ + "Add the extension that provides it, e.g. org.apache.camel.quarkus:camel-quarkus-"
+ + parser)));
+ }
+ }
String uri = entry.getValue().source().uri().orElse(null);
if (uri == null) {
continue;
@@ -261,6 +349,18 @@ class Langchain4jIngestProcessor {
+ "max-overlap-size must be smaller than it (got " + pipeline.maxSegmentSize()
+ " / " + pipeline.maxOverlapSize() + ")")));
}
+
+ if (pipeline.embeddingBatchSize() < 1) {
+ validationErrors.produce(new ValidationErrorBuildItem(new ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "': embedding-batch-size must be positive (got "
+ + pipeline.embeddingBatchSize() + ")")));
+ }
+
+ if (pipeline.maxDocumentSize() < 0) {
+ validationErrors.produce(new ValidationErrorBuildItem(new ConfigurationException(
+ "Ingestion pipeline '" + entry.getKey() + "': max-document-size must not be negative, 0 "
+ + "meaning no limit (got " + pipeline.maxDocumentSize() + ")")));
+ }
}
}
}
diff --git a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBatchSizeConfigTest.java b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBatchSizeConfigTest.java
new file mode 100644
index 0000000000..7d04ddf4e4
--- /dev/null
+++ b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBatchSizeConfigTest.java
@@ -0,0 +1,39 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** The embedding batch size is validated at build time on the configuration path. */
+class IngestBatchSizeConfigTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> {
+ })
+ .overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-batch-size", "0")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "embedding-batch-size must be positive"));
+
+ @Test
+ void buildMustFail() {
+ Assertions.fail("The build was expected to fail");
+ }
+}
diff --git a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderParserMissingComponentTest.java b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderParserMissingComponentTest.java
new file mode 100644
index 0000000000..b6c0d04082
--- /dev/null
+++ b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderParserMissingComponentTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Ingest;
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Source;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * A builder-declared parser is composed at startup, out of reach of the build-time check — so a
+ * missing parser component must fail in the pre-start task with the same artifact hint.
+ */
+class IngestBuilderParserMissingComponentTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> jar.addClasses(Pipelines.class, TestEmbeddingBeans.class))
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "parses with 'tika'", "component 'tika' is not on the classpath",
+ "org.apache.camel.quarkus:camel-quarkus-tika"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+
+ @ApplicationScoped
+ public static class Pipelines {
+
+ @Ingest("docs")
+ IngestPipeline docs() {
+ return IngestPipeline.from(Source.file("target/parser-docs"))
+ .parser("tika")
+ .embeddingStore("store")
+ .embeddingModel("model");
+ }
+ }
+}
diff --git a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMaxDocumentSizeConfigTest.java b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMaxDocumentSizeConfigTest.java
new file mode 100644
index 0000000000..571a0e4094
--- /dev/null
+++ b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMaxDocumentSizeConfigTest.java
@@ -0,0 +1,39 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** The document size cap is validated at build time on the configuration path. */
+class IngestMaxDocumentSizeConfigTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> {
+ })
+ .overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.max-document-size", "-1")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "max-document-size must not be negative"));
+
+ @Test
+ void buildMustFail() {
+ Assertions.fail("The build was expected to fail");
+ }
+}
diff --git a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestParserComponentMissingTest.java b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestParserComponentMissingTest.java
new file mode 100644
index 0000000000..1d047245ec
--- /dev/null
+++ b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestParserComponentMissingTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * A configured parser whose component is not on the classpath stops the build, naming the
+ * artifact that fixes it — the same treatment a missing consumer component gets.
+ */
+class IngestParserComponentMissingTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> {
+ })
+ .overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.parser", "tika")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "parses with 'tika'", "component 'tika' is not on the classpath",
+ "org.apache.camel.quarkus:camel-quarkus-tika"));
+
+ @Test
+ void buildMustFail() {
+ Assertions.fail("The build was expected to fail");
+ }
+}
diff --git a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestPipelineParserValidationTest.java b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestPipelineParserValidationTest.java
new file mode 100644
index 0000000000..f53a1a5efe
--- /dev/null
+++ b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestPipelineParserValidationTest.java
@@ -0,0 +1,42 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.deployment;
+
+import org.apache.camel.quarkus.component.langchain4j.ingest.IngestPipeline;
+import org.apache.camel.quarkus.component.langchain4j.ingest.Source;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** The builder rejects an unsupported parser eagerly, like the configuration path at build time. */
+class IngestPipelineParserValidationTest {
+
+ @Test
+ void unknownParserRejected() {
+ IngestPipeline pipeline = IngestPipeline.from(Source.file("target/docs"));
+ IllegalArgumentException e = Assertions.assertThrows(IllegalArgumentException.class,
+ () -> pipeline.parser("pdfmagic"));
+ Assertions.assertTrue(e.getMessage().contains("parser must be one of"), e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains("pdfmagic"), e.getMessage());
+ }
+
+ @Test
+ void supportedParsersAccepted() {
+ for (String parser : IngestPipeline.SUPPORTED_PARSERS) {
+ IngestPipeline.from(Source.file("target/docs")).parser(parser);
+ }
+ }
+}
diff --git a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnknownParserTest.java b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnknownParserTest.java
new file mode 100644
index 0000000000..f4e67d7af8
--- /dev/null
+++ b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnknownParserTest.java
@@ -0,0 +1,39 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.deployment;
+
+import io.quarkus.test.QuarkusExtensionTest;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** A parser value outside the supported set stops the build naming the alternatives. */
+class IngestUnknownParserTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> {
+ })
+ .overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.parser", "pdfmagic")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "sets parser 'pdfmagic'", "Supported parsers: docling, tika"));
+
+ @Test
+ void buildMustFail() {
+ Assertions.fail("The build was expected to fail");
+ }
+}
diff --git a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnnamedBeansTest.java b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnnamedBeansTest.java
index 219829bca4..0bc6f94b10 100644
--- a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnnamedBeansTest.java
+++ b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestUnnamedBeansTest.java
@@ -55,7 +55,7 @@ class IngestUnnamedBeansTest {
@Test
void unnamedStoreAndModelResolve() {
- Assertions.assertNotNull(context.getRoute("camel-quarkus-langchain4j-ingest-docs"),
+ Assertions.assertNotNull(context.getRoute("langchain4j-ingest-docs"),
"the pipeline must start with the only store and model beans, unnamed");
}
diff --git a/extensions/langchain4j-ingest/runtime/pom.xml b/extensions/langchain4j-ingest/runtime/pom.xml
index e444421f2b..bd56f95246 100644
--- a/extensions/langchain4j-ingest/runtime/pom.xml
+++ b/extensions/langchain4j-ingest/runtime/pom.xml
@@ -41,6 +41,10 @@
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-core</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-langchain4j-ingest</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-file</artifactId>
diff --git a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
index 62663db5e1..b9710c6bdc 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
+++ b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
@@ -1,6 +1,7 @@
=== Ingesting a directory
-Point a pipeline at a folder and the files in it become a knowledge base. No route, no ingestion code:
+Point a pipeline at a folder and the files in it become a knowledge base.
+No route, no ingestion code:
[source,properties]
----
@@ -9,16 +10,25 @@ quarkus.camel.langchain4j.ingest.products.embedding-store=products
quarkus.camel.langchain4j.ingest.products.embedding-model=my-model
----
-Each file is read as UTF-8 text — there is no format parsing, so convert a PDF or DOCX before it reaches the pipeline — split into overlapping segments (`max-segment-size`, `max-overlap-size`), embedded in batches and written to the store; a document is held in memory whole while it is split. `embedding-store` and `embedding-model` name CDI beans and may be omitted when the application has exactly one of each. Every segment carries `camel_ingest_pipeline` and `camel_ingest_document_id` m [...]
+Each file is read as UTF-8 text — unless a `parser` turns binary formats into text first, see below — split into overlapping segments (`max-segment-size`, `max-overlap-size`; or a `DocumentSplitter` CDI bean named by `document-splitter`, which then replaces the sizes), embedded in batches (`embedding-batch-size`, 32 by default) and written to the store; a document is held in memory whole while it is split, so `max-document-size` can cap it in characters — 0, the default, means no limit, [...]
+`embedding-store` and `embedding-model` name CDI beans and may be omitted when the application has exactly one of each.
+Every segment carries `camel_ingest_pipeline` and `camel_ingest_document_id` metadata, so retrieval can cite which document an answer came from (up to 3.39.x the keys were `camel_quarkus_pipeline` and `camel_quarkus_document_id` — see the migration guide).
+Apart from `enabled` and the `source.*` settings shown here, properties are fixed at build time.
+A pipeline declared through runtime properties alone — nothing but a `source.directory`, say — is invisible to build-time validation; its checks, including the clash with an equally named `@Ingest` pipeline, report at startup instead.
[NOTE]
====
-This experimental extension keeps no record of what it wrote: an edited document re-ingests on top of its old segments, and removing a document from the source removes nothing from the store. Replace and delete arrive with the synchronising engine in a later release. What is remembered is which documents were already ingested — the idempotent repository below.
+This experimental extension keeps no record of what it wrote: an edited document re-ingests on top of its old segments, and removing a document from the source removes nothing from the store.
+Replace and delete arrive with the synchronising engine in a later release.
+What is remembered is which documents were already ingested — the idempotent repository below.
====
=== The idempotent repository
-A directory pipeline registers what it ingested, keyed on the file's path, modification time and size: unchanged files are skipped, edited files re-ingest (their previous segments remain, see the note above). The default register is in-memory (100,000 keys) and is there for the poll loop, not for restarts: the consumer re-scans the directory on every poll, so without a register a running application would re-embed the whole directory every few seconds. A restart loses it and re-ingests t [...]
+A directory pipeline registers what it ingested, keyed on the file's path, modification time and size: unchanged files are skipped, edited files re-ingest (their previous segments remain, see the note above).
+The default register is in-memory (100,000 keys) and is there for the poll loop, not for restarts: the consumer re-scans the directory on every poll, so without a register a running application would re-embed the whole directory every few seconds.
+A restart loses it and re-ingests the directory once; a persistent repository makes restarts free.
+`source.idempotent-repository` names an `IdempotentRepository` bean to use instead:
[source,properties]
----
@@ -27,7 +37,8 @@ quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=productsR
The bean can be provided three ways.
-*Auto-created*: an in-memory register (100,000 keys) is created and bound under the configured name — no bean definition needed, still lost on restart. If a bean with the name already exists, the existing bean wins:
+*Auto-created*: an in-memory register (100,000 keys) is created and bound under the configured name — no bean definition needed, still lost on restart.
+If a bean with the name already exists, the existing bean wins:
[source,properties]
----
@@ -35,7 +46,8 @@ quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=productsR
quarkus.camel.langchain4j.ingest.products.source.idempotent-repository-auto-create=true
----
-*Defined in properties* through Camel's `camel.beans.` syntax — here a file-backed register that survives restarts (camel-core, no extra dependency). Use a name without dashes; Camel normalises dashed `camel.beans.` keys to camelCase:
+*Defined in properties* through Camel's `camel.beans.` syntax — here a file-backed register that survives restarts (camel-core, no extra dependency).
+Use a name without dashes; Camel normalises dashed `camel.beans.` keys to camelCase:
[source,properties]
----
@@ -46,9 +58,12 @@ camel.beans.productsRegister.cacheSize=100000
camel.beans.productsRegister.maxFileStoreSize=33554432
----
-`#class:` beans are created reflectively. For native mode the extension registers the camel-core repositories shown here and every `IdempotentRepository` implementation found in the Jandex index — application classes always, component-provided ones through their jar's index. A class from a jar without an index needs `@RegisterForReflection(targets = ...)` or a `quarkus.index-dependency.*` entry.
+`#class:` beans are created reflectively.
+For native mode the extension registers the camel-core repositories shown here and every `IdempotentRepository` implementation found in the Jandex index — application classes always, component-provided ones through their jar's index.
+A class from a jar without an index needs `@RegisterForReflection(targets = ...)` or a `quarkus.index-dependency.*` entry.
-*Defined as a CDI producer* — needed when construction takes other beans. A JDBC register also deduplicates across instances: reliably for consumer-fed pipelines keyed on the document id, for directory pipelines only when all instances see identical paths and modification times:
+*Defined as a CDI producer* — needed when construction takes other beans.
+A JDBC register also deduplicates across instances: reliably for consumer-fed pipelines keyed on the document id, for directory pipelines only when all instances see identical paths and modification times:
[source,java]
----
@@ -60,15 +75,50 @@ IdempotentRepository productsRegister(DataSource dataSource) {
More repository types — Caffeine, Infinispan, MongoDB, Cassandra, Hazelcast, Kafka and others, each provided by its component's extension — along with the pattern's details are covered in the xref:{cq-camel-components}:eips:idempotentConsumer-eip.adoc[Idempotent Consumer EIP guide].
-Sizing. An in-memory register smaller than the directory evicts keys and re-ingests those files during normal operation — hence the 100,000 default. `FileIdempotentRepository` falls back to its file store on a cache miss, so `cacheSize` is a performance setting; the correctness limit is `maxFileStoreSize`: the default of about 1 MB holds roughly 10–17 thousand of these keys, beyond it the oldest 1000 are dropped with a warning. A JDBC register has no size limit. Every edit adds a new key [...]
+Sizing.
+An in-memory register smaller than the directory evicts keys and re-ingests those files during normal operation — hence the 100,000 default.
+`FileIdempotentRepository` falls back to its file store on a cache miss, so `cacheSize` is a performance setting; the correctness limit is `maxFileStoreSize`: the default of about 1 MB holds roughly 10–17 thousand of these keys, beyond it the oldest 1000 are dropped with a warning.
+A JDBC register has no size limit.
+Every edit adds a new key and old keys are never removed, so a persistent register grows under churn until the synchronising engine arrives.
+There is no size option; size the bean.
+Use one register per pipeline — keys are source-derived, and pipelines sharing a register over the same directory skip each other's work.
-The register cannot be switched off: the consumer leaves files in place (`noop`), so without a register every poll would re-ingest the directory. To force a full re-ingest, restart (default register) or clear the persistent one. A pipeline that should consume its source — move or delete files after ingestion — uses `source.uri` (for example `file:/inbox?delete=true&idempotent=false` with `source.document-id=CamelFileName`), where all consumer options are available.
+The register cannot be switched off: the consumer leaves files in place (`noop`), so without a register every poll would re-ingest the directory.
+To force a full re-ingest, restart (default register) or clear the persistent one.
+A pipeline that should consume its source — move or delete files after ingestion — uses `source.uri` (for example `file:/inbox?delete=true&idempotent=false` with `source.document-id=CamelFileName`), where all consumer options are available.
-On a consumer-fed pipeline (`source.uri` or `@Ingest`), a configured register deduplicates deliveries by document id: a redelivered record or re-listed object ingests once, and a request-reply caller receives `skipped` instead of `ingested`. First write wins per id — an update with the same id is skipped, not replaced; streams that carry updates need a version-aware `source.document-id`. Only a delivery that wrote segments claims its id: a blank document answers `empty` and releases the [...]
+On a consumer-fed pipeline (`source.uri` or `@Ingest`), a configured register deduplicates deliveries by document id: a redelivered record or re-listed object ingests once, and a request-reply caller receives `skipped` instead of `ingested`.
+First write wins per id — an update with the same id is skipped, not replaced; streams that carry updates need a version-aware `source.document-id`.
+Only a delivery that wrote segments claims its id: a blank document answers `empty` and releases the claim, so an object created empty and populated later under the same id still ingests the content.
+With eager idempotency, a duplicate racing an in-flight first delivery is answered `skipped` even if that delivery then fails and releases the key.
+
+=== Parsing PDFs and office documents
+
+Setting `parser` puts a document parser between the consumer and the splitter:
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.reports.source.directory=/var/data/reports
+quarkus.camel.langchain4j.ingest.reports.parser=tika
+----
+
+`tika` extracts plain text in-process with Apache Tika.
+It needs the `camel-quarkus-tika` extension, plus the Tika parser module for each format the pipeline should read — `org.apache.tika:tika-parser-pdf-module` for PDF, say.
+A format without its module parses to nothing and the document is skipped as empty (logged as a warning); so is a scanned, image-only PDF, since Tika does no OCR here.
+On a directory pipeline an empty parse still commits the file's key in the duplicate register, so the file is not retried until it changes — after adding a missing parser module, force a re-ingest (restart with the default register, or clear the persistent one).
+A consumer-fed pipeline releases the claim of an empty delivery instead.
+
+`docling` converts documents to structure-preserving markdown through a https://github.com/docling-project/docling-serve[Docling Serve] instance, OCR included.
+It needs the `camel-quarkus-docling` extension and the instance's address in `camel.component.docling.docling-serve-url`.
+
+A missing parser extension fails the build naming the artifact, the same way a missing consumer component does.
+The Java twin is `IngestPipeline.parser("tika")`, and a consumer-fed pipeline parses the same way — the payload the consumer delivers goes to the parser first.
+With a parser, `max-document-size` also rejects a raw payload larger than the cap (in bytes) before it reaches the parser, on top of capping the extracted text (in characters).
=== Other sources, declared in Java
-Any Camel consumer can feed a pipeline — the roughly 300 components, each with its own options and its own documentation. Such a pipeline is declared in Java with `@Ingest` and the Camel Endpoint DSL:
+Any Camel consumer can feed a pipeline — the roughly 300 components, each with its own options and its own documentation.
+Such a pipeline is declared in Java with `@Ingest` and the Camel Endpoint DSL:
[source,java]
----
@@ -80,13 +130,20 @@ IngestPipeline events() {
}
----
-Typing `dsl.` lists a factory for every component, each completing its own typed options. The method runs once at startup; it must return `IngestPipeline`, take no parameters and use a name no configured pipeline uses. `enabled=false` in configuration switches a Java-declared pipeline off. A component missing from the classpath fails at startup with an error naming the extension artifact that provides it — the DSL compiles regardless, since its factories all ship in one artifact.
+Typing `dsl.` lists a factory for every component, each completing its own typed options.
+The method runs once at startup; it must return `IngestPipeline`, take no parameters and use a name no configured pipeline uses.
+`enabled=false` in configuration switches a Java-declared pipeline off.
+A component missing from the classpath fails at startup with an error naming the extension artifact that provides it — the DSL compiles regardless, since its factories all ship in one artifact.
-Ingestion needs a stable id per document, and where it lives is the consumer's business: `documentId` names the header — the record key `CamelKafkaKey` above, `CamelAwsS3Key` for S3 — or gives a simple-language expression. Without it, the pipeline expects the `CamelIngestDocumentId` header and fails the exchange when it is absent. Mind each component's own defaults, too: the `aws2-s3` consumer *deletes objects after reading them* unless `deleteAfterRead(false)` is set — a knowledge base [...]
+Ingestion needs a stable id per document, and where it lives is the consumer's business: `documentId` names the header — the record key `CamelKafkaKey` above, `CamelAwsS3Key` for S3 — or gives a simple-language expression.
+Without it, the pipeline expects the `CamelIngestDocumentId` header and fails the exchange when it is absent.
+Mind each component's own defaults, too: the `aws2-s3` consumer *deletes objects after reading them* unless `deleteAfterRead(false)` is set — a knowledge base reads its source, it does not consume it.
After such a pipeline ingests a document, the exchange body is replaced with the `IngestResult`, so a request-reply caller receives the outcome of its call.
-The same pipeline can be declared purely in properties: `source.uri` takes the consumer URI as written, `source.document-id` the header name — or a simple-language expression written `+$simple{...}+`, the one form MicroProfile Config leaves untouched. The URI is fixed at build time by design — a runtime-overridable consumer URI would be arbitrary component invocation — while property placeholders inside it still resolve at startup, keeping credentials and endpoints runtime configuration. [...]
+The same pipeline can be declared purely in properties: `source.uri` takes the consumer URI as written, `source.document-id` the header name — or a simple-language expression written `+$simple{...}+`, the one form MicroProfile Config leaves untouched.
+The URI is fixed at build time by design — a runtime-overridable consumer URI would be arbitrary component invocation — while property placeholders inside it still resolve at startup, keeping credentials and endpoints runtime configuration.
+Treat runtime configuration as the trust boundary it is: the directory and the id expression decide what the process reads into an often external store.
[source,properties]
----
@@ -97,4 +154,7 @@ quarkus.camel.langchain4j.ingest.s3docs.embedding-store=products
=== When ingestion fails
-A failure while splitting, embedding or storing — a rate-limited model, an unreachable store — propagates to the consumer; there is no dead-letter channel in this increment. For a directory pipeline the file stays where it is and is retried on the next poll, because the duplicate-protection key is only committed on success — which also means a permanently failing file is retried forever, loudly. For a consumer-fed pipeline the component's own error handling applies: a request-reply calle [...]
+A failure while parsing, splitting, embedding or storing — an encrypted or corrupt document, an unreachable docling-serve, a rate-limited model, an unreachable store — propagates to the consumer; there is no dead-letter channel in this increment.
+For a directory pipeline the file stays where it is and is retried on the next poll, because the duplicate-protection key is only committed on success — which also means a permanently failing file is retried forever, loudly, on every poll: with a parser that includes a document the parser cannot read, so remove or fix such a file, or declare the pipeline with `source.uri` and the file consumer's `moveFailed` option to quarantine failures instead.
+For a consumer-fed pipeline the component's own error handling applies: a request-reply caller receives the exception, while a Kafka consumer with default settings logs the failure and commits the offset, so the record is *dropped* — and since this engine keeps no record either, nothing remembers it.
+A record whose configured `document-id` resolves to nothing (a Kafka record without a key, say) fails the same way, one exchange at a time.
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java
index a11df04a85..98742f8d64 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java
+++ b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestBuildTimeConfig.java
@@ -40,6 +40,12 @@ public interface IngestBuildTimeConfig {
/** Mirrors the {@code @WithDefault} below, which can only carry a literal. */
int DEFAULT_MAX_OVERLAP_SIZE = 50;
+ /** Mirrors the {@code @WithDefault} below, which can only carry a literal. */
+ int DEFAULT_EMBEDDING_BATCH_SIZE = 32;
+
+ /** Mirrors the {@code @WithDefault} below, which can only carry a literal; 0 means no limit. */
+ int DEFAULT_MAX_DOCUMENT_SIZE = 0;
+
/**
* Ingestion pipelines by name.
*/
@@ -54,6 +60,16 @@ public interface IngestBuildTimeConfig {
*/
SourceBuildTimeConfig source();
+ /**
+ * How a consumed payload becomes text before it is split: `tika` extracts plain text
+ * from PDF, office and similar formats in-process, `docling` converts to
+ * structure-preserving markdown through a Docling Serve instance (configure it with
+ * `camel.component.docling.*`). When not set, the payload is read as text as-is. Each
+ * value needs its extension on the classpath: `camel-quarkus-tika` or
+ * `camel-quarkus-docling`.
+ */
+ Optional<String> parser();
+
/**
* Name of the `EmbeddingStore` bean to write to. When not set, the only one present is
* used.
@@ -79,6 +95,32 @@ public interface IngestBuildTimeConfig {
@WithDefault("50")
int maxOverlapSize();
+ /**
+ * How many segments are embedded per request to the embedding model. Providers with
+ * generous per-request limits ingest large documents faster with a bigger batch; a batch
+ * carries at most `embedding-batch-size` × `max-segment-size` characters, so tune the two
+ * together against the provider's token limits.
+ */
+ @WithDefault("32")
+ int embeddingBatchSize();
+
+ /**
+ * Maximum size of one document in characters, applied to the text about to be split;
+ * 0, the default, means no limit. The pipeline holds a document in memory whole, so the
+ * cap is the protection against oversized — on a consumer-fed pipeline, attacker-sized —
+ * payloads. An oversized document fails the exchange cleanly.
+ */
+ @WithDefault("0")
+ int maxDocumentSize();
+
+ /**
+ * Name of the `DocumentSplitter` bean replacing the default recursive splitting;
+ * `max-segment-size` and `max-overlap-size` are then ignored. Looked up by name only —
+ * an application may hold unrelated splitters. Segments returned without the identity
+ * metadata are re-stamped, so a custom splitter cannot break citation.
+ */
+ Optional<String> documentSplitter();
+
/**
* A pipeline reads either a directory or a Camel consumer, and the two halves of that
* choice deliberately sit in different config roots: the consumer URI is fixed at build
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestComponentPresence.java b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestComponentPresence.java
index 7fb10e8ae9..ce6e1e5539 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestComponentPresence.java
+++ b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestComponentPresence.java
@@ -72,6 +72,17 @@ final class IngestComponentPresence {
if ("endpoint".equals(definition.sourceType())) {
require(context, entry.name(), definition.sourceUri());
}
+ // a builder-declared parser is out of reach of the build-time check, like the URI;
+ // resolved without starting the component - the route builder starts it later. No
+ // catch on purpose: a component that resolves but fails to initialise surfaces its
+ // real error, only an unresolvable one earns the artifact hint below
+ String parser = definition.parser().orElse(null);
+ if (parser != null && context.getComponent(parser, true, false) == null) {
+ throw new IllegalStateException("Ingestion pipeline '" + entry.name() + "' parses with '" + parser
+ + "', but the Camel component '" + parser + "' is not on the classpath. "
+ + "Add the extension that provides it, e.g. org.apache.camel.quarkus:camel-quarkus-"
+ + parser);
+ }
}
}
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestHeaders.java b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestHeaders.java
index d2834bce41..439e34d9e9 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestHeaders.java
+++ b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestHeaders.java
@@ -25,6 +25,13 @@ public final class IngestHeaders {
* The stable document id, when {@code source.document-id} does not name another header.
* Identity is what future update and delete semantics will be built on, so a generated
* fallback would silently break replacement — an absent id is an error.
+ *
+ * <p>
+ * Deliberately not the {@code camel-langchain4j-ingest} component's own default
+ * ({@code LangChain4jIngestHeaders.DOCUMENT_ID}, {@code CamelLangChain4jIngestDocumentId}):
+ * this name is the contract this extension released in 3.39.0, and it stays pinned on
+ * consumer pipelines so existing routes keep working — the compatibility burden lives
+ * downstream, where it was created.
*/
public static final String DOCUMENT_ID = "CamelIngestDocumentId";
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java
index 7f677daf48..d5c9ad8c9e 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java
+++ b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipeline.java
@@ -17,6 +17,7 @@
package org.apache.camel.quarkus.component.langchain4j.ingest;
import java.util.Optional;
+import java.util.Set;
/**
* A pipeline declared in Java rather than in configuration, returned from an {@link Ingest}
@@ -24,11 +25,18 @@ import java.util.Optional;
*/
public final class IngestPipeline {
+ /** The values {@link #parser(String)} and the {@code parser} configuration property accept. */
+ public static final Set<String> SUPPORTED_PARSERS = IngestPipelineDefinition.SUPPORTED_PARSERS;
+
private final Source source;
private String embeddingStoreName;
private String embeddingModelName;
+ private String parser;
private int maxSegmentSize = IngestBuildTimeConfig.DEFAULT_MAX_SEGMENT_SIZE;
private int maxOverlapSize = IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE;
+ private int embeddingBatchSize = IngestBuildTimeConfig.DEFAULT_EMBEDDING_BATCH_SIZE;
+ private int maxDocumentSize = IngestBuildTimeConfig.DEFAULT_MAX_DOCUMENT_SIZE;
+ private String documentSplitterName;
private IngestPipeline(Source source) {
this.source = source;
@@ -48,6 +56,23 @@ public final class IngestPipeline {
return this;
}
+ /**
+ * Parses the consumed payload into text before splitting: {@code tika} extracts plain text
+ * in-process, {@code docling} converts to markdown through a Docling Serve instance. The
+ * corresponding extension must be on the classpath; the twin of the {@code parser}
+ * configuration property.
+ */
+ public IngestPipeline parser(String parser) {
+ // the same rule the configuration path is held to at build time; the null check comes
+ // first because the unmodifiable set's contains(null) throws a bare NPE
+ if (parser == null || !SUPPORTED_PARSERS.contains(parser)) {
+ throw new IllegalArgumentException("parser must be one of " + SUPPORTED_PARSERS
+ + " (got '" + parser + "')");
+ }
+ this.parser = parser;
+ return this;
+ }
+
public IngestPipeline splitter(int maxSegmentSize, int maxOverlapSize) {
// the same rule the configuration path is held to at build time, so both paths reject
// the same values rather than failing later inside the splitter
@@ -60,6 +85,44 @@ public final class IngestPipeline {
return this;
}
+ /**
+ * How many segments are embedded per request to the embedding model; the twin of the
+ * {@code embedding-batch-size} configuration property.
+ */
+ public IngestPipeline embeddingBatchSize(int embeddingBatchSize) {
+ // the same rule the configuration path is held to at build time
+ if (embeddingBatchSize < 1) {
+ throw new IllegalArgumentException(
+ "embedding-batch-size must be positive (got " + embeddingBatchSize + ")");
+ }
+ this.embeddingBatchSize = embeddingBatchSize;
+ return this;
+ }
+
+ /**
+ * Maximum size of one document in characters; 0, the default, means no limit. The twin of
+ * the {@code max-document-size} configuration property.
+ */
+ public IngestPipeline maxDocumentSize(int maxDocumentSize) {
+ // the same rule the configuration path is held to at build time
+ if (maxDocumentSize < 0) {
+ throw new IllegalArgumentException(
+ "max-document-size must not be negative, 0 meaning no limit (got " + maxDocumentSize + ")");
+ }
+ this.maxDocumentSize = maxDocumentSize;
+ return this;
+ }
+
+ /**
+ * Name of the {@code DocumentSplitter} bean replacing the default recursive splitting; the
+ * twin of the {@code document-splitter} configuration property. The splitter sizes are then
+ * ignored.
+ */
+ public IngestPipeline documentSplitter(String beanName) {
+ this.documentSplitterName = beanName;
+ return this;
+ }
+
String sourceType() {
return source.type();
}
@@ -76,6 +139,10 @@ public final class IngestPipeline {
return Optional.ofNullable(embeddingModelName);
}
+ Optional<String> parser() {
+ return Optional.ofNullable(parser);
+ }
+
int maxSegmentSize() {
return maxSegmentSize;
}
@@ -84,6 +151,18 @@ public final class IngestPipeline {
return maxOverlapSize;
}
+ int embeddingBatchSize() {
+ return embeddingBatchSize;
+ }
+
+ int maxDocumentSize() {
+ return maxDocumentSize;
+ }
+
+ Optional<String> documentSplitterName() {
+ return Optional.ofNullable(documentSplitterName);
+ }
+
/** The configuration view, so a builder pipeline reuses every configuration path verbatim. */
IngestRunTimeConfig.PipelineRunTimeConfig asRunTimeConfig() {
IngestRunTimeConfig.PipelineRunTimeConfig.SourceRunTimeConfig sourceConfig = source.asRunTimeConfig();
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineDefinition.java b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineDefinition.java
new file mode 100644
index 0000000000..e94a65684e
--- /dev/null
+++ b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineDefinition.java
@@ -0,0 +1,351 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest;
+
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.apache.camel.builder.EndpointConsumerBuilder;
+
+/**
+ * A declarative description of one ingestion pipeline, turned into a route by {@link IngestPipelineRouteBuilder}: what
+ * to consume, how to identify each document, whether to parse it, and where to embed and store it.
+ *
+ * <p>
+ * A pipeline reads either a directory ({@link #directory(String, String)}) or any Camel consumer
+ * ({@link #consumer(String, String)}); every other property has a matching endpoint option on the
+ * {@code langchain4j-ingest} producer the generated route ends in.
+ *
+ * <p>
+ * An internal copy of the pipeline assembly proposed alongside the upstream component — see
+ * {@link IngestPipelineRouteBuilder}.
+ */
+final class IngestPipelineDefinition {
+
+ /** The supported document parsers; the configuration value is the lower-case name. */
+ public enum Parser {
+ TIKA,
+ DOCLING;
+
+ static Parser of(String value) {
+ try {
+ return valueOf(value.toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "parser must be one of " + SUPPORTED_PARSERS + " (got '" + value + "')");
+ }
+ }
+ }
+
+ // derived from the enum, so a parser added there cannot diverge from the validated set
+ public static final Set<String> SUPPORTED_PARSERS = Arrays.stream(Parser.values())
+ .map(parser -> parser.name().toLowerCase(Locale.ROOT))
+ .collect(Collectors.toUnmodifiableSet());
+
+ private final String name;
+ private final String directory;
+ private final String uri;
+
+ private boolean recursive = true;
+ private String documentId;
+ private Parser parser;
+ private int maxSegmentSize = IngestBuildTimeConfig.DEFAULT_MAX_SEGMENT_SIZE;
+ private int maxOverlapSize = IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE;
+ private int embeddingBatchSize = IngestBuildTimeConfig.DEFAULT_EMBEDDING_BATCH_SIZE;
+ private int maxDocumentSize = IngestBuildTimeConfig.DEFAULT_MAX_DOCUMENT_SIZE;
+
+ private String embeddingStoreRef;
+ private EmbeddingStore<TextSegment> embeddingStore;
+ private String embeddingModelRef;
+ private EmbeddingModel embeddingModel;
+ private String documentSplitterRef;
+
+ private String idempotentRepositoryRef;
+ private boolean idempotentRepositoryAutoCreate;
+
+ private IngestPipelineDefinition(String name, String directory, String uri) {
+ this.name = name;
+ this.directory = directory;
+ this.uri = uri;
+ }
+
+ /**
+ * A pipeline reading the files of a directory: they are left in place (a knowledge base reads its source, it does
+ * not consume it), unchanged files are remembered and skipped, and a file still being copied in is waited for
+ * rather than half-ingested.
+ */
+ public static IngestPipelineDefinition directory(String name, String directory) {
+ requireText(directory, "directory");
+ if (directory.contains("?") || directory.contains("#")) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + name + "': the directory must not contain '?' or '#' (got '"
+ + directory + "')");
+ }
+ return new IngestPipelineDefinition(requireName(name), directory, null);
+ }
+
+ /**
+ * A pipeline consuming from any Camel component, with the component's own options — the escape hatch beside
+ * {@link #directory(String, String)}. Which part of the exchange identifies the document is the consumer's
+ * business, so say it with {@link #documentId}: {@code CamelAwsS3Key} for an S3 consumer, {@code CamelKafkaKey} for
+ * a Kafka one.
+ */
+ public static IngestPipelineDefinition consumer(String name, String uri) {
+ requireText(uri, "uri");
+ if (!uri.contains(":")) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + name + "': '" + uri + "' is not a consumer URI");
+ }
+ return new IngestPipelineDefinition(requireName(name), null, uri);
+ }
+
+ /**
+ * The same, built with the Camel Endpoint DSL (add the {@code camel-endpointdsl} dependency and use its static
+ * builders or an {@code EndpointBuilderFactory}).
+ */
+ public static IngestPipelineDefinition consumer(String name, EndpointConsumerBuilder uri) {
+ return consumer(name, uri.getRawUri());
+ }
+
+ /** Whether subdirectories are ingested too; directory pipelines only. */
+ public IngestPipelineDefinition recursive(boolean recursive) {
+ this.recursive = recursive;
+ return this;
+ }
+
+ /**
+ * Where the document id lives in the exchange the consumer delivers: the name of a header, or a simple-language
+ * expression (a value containing {@code ${} or {@code $simple{} is parsed as one). When not set, a directory
+ * pipeline uses the file name and a consumer pipeline the {@code CamelLangChain4jIngestDocumentId} header. The id
+ * is captured before any parse, so a document cannot forge its own identity.
+ */
+ public IngestPipelineDefinition documentId(String documentId) {
+ this.documentId = requireText(documentId, "documentId");
+ return this;
+ }
+
+ /**
+ * Parses the consumed payload into text before splitting: {@code tika} extracts plain text in-process,
+ * {@code docling} converts to markdown through a Docling Serve instance. The corresponding component must be on the
+ * classpath: {@code camel-tika} or {@code camel-docling}.
+ */
+ public IngestPipelineDefinition parser(String parser) {
+ this.parser = Parser.of(requireText(parser, "parser"));
+ return this;
+ }
+
+ /**
+ * The typed twin of {@link #parser(String)}, for Java callers; the String form exists for configuration
+ * translators, whose input is text by nature.
+ */
+ public IngestPipelineDefinition parser(Parser parser) {
+ if (parser == null) {
+ throw new IllegalArgumentException("Ingestion pipeline parser must not be null");
+ }
+ this.parser = parser;
+ return this;
+ }
+
+ /** The splitter bounds; the same rule the endpoint enforces, failing early here. */
+ public IngestPipelineDefinition splitter(int maxSegmentSize, int maxOverlapSize) {
+ if (maxSegmentSize <= 0 || maxOverlapSize < 0 || maxOverlapSize >= maxSegmentSize) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + name + "': maxSegmentSize must be positive and maxOverlapSize must be"
+ + " non-negative and smaller than it (got " + maxSegmentSize + " / "
+ + maxOverlapSize + ")");
+ }
+ this.maxSegmentSize = maxSegmentSize;
+ this.maxOverlapSize = maxOverlapSize;
+ return this;
+ }
+
+ /**
+ * How many segments are embedded per request to the embedding model; the same rule the endpoint enforces, failing
+ * early here. A batch carries at most {@code embeddingBatchSize x maxSegmentSize} characters, so tune the two
+ * together against the provider's token limits.
+ */
+ public IngestPipelineDefinition embeddingBatchSize(int embeddingBatchSize) {
+ if (embeddingBatchSize < 1) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + name + "': embeddingBatchSize must be positive (got "
+ + embeddingBatchSize + ")");
+ }
+ this.embeddingBatchSize = embeddingBatchSize;
+ return this;
+ }
+
+ /**
+ * Maximum size of one document; unset means no limit. The pipeline holds a document in memory whole, so the cap is
+ * the protection against oversized — on a consumer-fed pipeline, attacker-sized — payloads. It is enforced twice: a
+ * pipeline with a {@link #parser(Parser)} rejects a larger raw payload (in bytes) before the parse ever runs, and
+ * the endpoint rejects longer extracted text (in characters) before splitting — a small file can still parse into a
+ * lot of text. An oversized document fails the exchange cleanly: a directory pipeline quarantines it like any other
+ * failure, a consumer pipeline propagates the error to its source, and a dedup claim is released either way.
+ */
+ public IngestPipelineDefinition maxDocumentSize(int maxDocumentSize) {
+ if (maxDocumentSize < 0) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + name + "': maxDocumentSize must not be negative, 0 meaning no limit (got "
+ + maxDocumentSize + ")");
+ }
+ this.maxDocumentSize = maxDocumentSize;
+ return this;
+ }
+
+ /**
+ * Name of the {@code DocumentSplitter} bean replacing the default recursive splitting; {@code maxSegmentSize} and
+ * {@code maxOverlapSize} are then ignored. Referenced by name on purpose — an application may hold unrelated
+ * splitters, so none is picked up by type. Segments returned without the identity metadata are re-stamped, so a
+ * custom splitter cannot break citation.
+ */
+ public IngestPipelineDefinition documentSplitter(String beanName) {
+ this.documentSplitterRef = requireText(beanName, "documentSplitter");
+ return this;
+ }
+
+ /** Name of the {@code EmbeddingStore} bean to write to; omitted, the single bean of the type is used. */
+ public IngestPipelineDefinition embeddingStore(String beanName) {
+ this.embeddingStoreRef = requireText(beanName, "embeddingStore");
+ return this;
+ }
+
+ /** The {@code EmbeddingStore} to write to, as an instance; it is bound to the registry under a generated name. */
+ public IngestPipelineDefinition embeddingStore(EmbeddingStore<TextSegment> store) {
+ this.embeddingStore = store;
+ return this;
+ }
+
+ /** Name of the {@code EmbeddingModel} bean to embed with; omitted, the single bean of the type is used. */
+ public IngestPipelineDefinition embeddingModel(String beanName) {
+ this.embeddingModelRef = requireText(beanName, "embeddingModel");
+ return this;
+ }
+
+ /** The {@code EmbeddingModel} to embed with, as an instance; it is bound to the registry under a generated name. */
+ public IngestPipelineDefinition embeddingModel(EmbeddingModel model) {
+ this.embeddingModel = model;
+ return this;
+ }
+
+ /**
+ * Name of the {@code IdempotentRepository} bean remembering ingested documents. A directory pipeline uses it in its
+ * file consumer, keyed on path, modification time and size, so a persistent repository makes restarts free. A
+ * consumer pipeline deduplicates deliveries by document id: first write wins.
+ */
+ public IngestPipelineDefinition idempotentRepository(String beanName) {
+ this.idempotentRepositoryRef = requireText(beanName, "idempotentRepository");
+ return this;
+ }
+
+ /**
+ * When {@code true}, an in-memory register (100 000 keys, lost on restart) is created and bound under the
+ * {@link #idempotentRepository(String)} name, unless a bean with that name exists — the existing bean wins.
+ */
+ public IngestPipelineDefinition idempotentRepositoryAutoCreate(boolean autoCreate) {
+ this.idempotentRepositoryAutoCreate = autoCreate;
+ return this;
+ }
+
+ private static String requireName(String name) {
+ requireText(name, "name");
+ if (!name.matches("[A-Za-z0-9._-]+")) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline name '" + name + "' may only contain letters, digits, '.', '_' and '-'");
+ }
+ return name;
+ }
+
+ private static String requireText(String value, String what) {
+ if (value == null || value.isBlank()) {
+ throw new IllegalArgumentException("Ingestion pipeline " + what + " must not be null or blank");
+ }
+ return value;
+ }
+
+ String name() {
+ return name;
+ }
+
+ String directory() {
+ return directory;
+ }
+
+ String uri() {
+ return uri;
+ }
+
+ boolean recursive() {
+ return recursive;
+ }
+
+ String documentId() {
+ return documentId;
+ }
+
+ Parser parser() {
+ return parser;
+ }
+
+ int maxSegmentSize() {
+ return maxSegmentSize;
+ }
+
+ int maxOverlapSize() {
+ return maxOverlapSize;
+ }
+
+ int embeddingBatchSize() {
+ return embeddingBatchSize;
+ }
+
+ int maxDocumentSize() {
+ return maxDocumentSize;
+ }
+
+ String documentSplitterRef() {
+ return documentSplitterRef;
+ }
+
+ String embeddingStoreRef() {
+ return embeddingStoreRef;
+ }
+
+ EmbeddingStore<TextSegment> embeddingStore() {
+ return embeddingStore;
+ }
+
+ String embeddingModelRef() {
+ return embeddingModelRef;
+ }
+
+ EmbeddingModel embeddingModel() {
+ return embeddingModel;
+ }
+
+ String idempotentRepositoryRef() {
+ return idempotentRepositoryRef;
+ }
+
+ boolean idempotentRepositoryAutoCreate() {
+ return idempotentRepositoryAutoCreate;
+ }
+}
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java
new file mode 100644
index 0000000000..2f5e97eb2a
--- /dev/null
+++ b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java
@@ -0,0 +1,423 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest;
+
+import java.io.InputStream;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.Component;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.langchain4j.ingest.IngestResult;
+import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest;
+import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders;
+import org.apache.camel.component.langchain4j.ingest.TikaTextDecode;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.builder.ExpressionBuilder;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.util.StringHelper;
+import org.apache.camel.util.URISupport;
+
+/**
+ * Generates one Camel route per {@link IngestPipelineDefinition}: consume, resolve the document id, optionally parse,
+ * then split, embed and store through the {@code langchain4j-ingest} producer.
+ *
+ * <p>
+ * A directory pipeline watches its folder with the safe file-consumer defaults — documents are left in place, unchanged
+ * files are remembered in a duplicate register keyed on path, modification time and size, and a file still being copied
+ * in is waited for. A consumer pipeline reads any component and deduplicates by document id when a repository is
+ * configured. The id is always captured into an exchange property <em>before</em> the parse stage: a parser copies
+ * document metadata over the headers, so a crafted document could otherwise forge its own identity.
+ *
+ * <p>
+ * An internal copy of the pipeline assembly proposed alongside the upstream component: the delegation to Apache Camel
+ * is engine-only, so the topology lives here, package-private — replaceable by an upstream artifact or kamelets if the
+ * community adopts one of them.
+ *
+ * <p>
+ * The class is abstract on purpose: camel-quarkus routes discovery instantiates every concrete public
+ * {@code RouteBuilder} it finds reflectively, and an abstract base is skipped by construction.
+ */
+abstract class IngestPipelineRouteBuilder extends RouteBuilder {
+
+ /**
+ * The built-in register capacity, sized above Camel's 1000-entry default so eviction does not re-ingest large
+ * directories during normal operation; in-memory, so lost on restart.
+ */
+ static final int DEFAULT_REGISTER_CAPACITY = 100_000;
+
+ /**
+ * The pipelines to build routes for; supplied by the subclass assembling its definitions from another source — a
+ * configuration model, say.
+ */
+ protected abstract List<IngestPipelineDefinition> pipelines();
+
+ @Override
+ public void configure() {
+ Set<String> names = new HashSet<>();
+ for (IngestPipelineDefinition pipeline : pipelines()) {
+ if (!names.add(pipeline.name())) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline.name() + "' is defined twice. Use one name per pipeline.");
+ }
+ configurePipeline(pipeline);
+ }
+ }
+
+ private void configurePipeline(IngestPipelineDefinition pipeline) {
+ requireParserComponent(pipeline);
+ String storeRef = bindInstance(pipeline.embeddingStore(), pipeline.embeddingStoreRef(), pipeline, "store");
+ String modelRef = bindInstance(pipeline.embeddingModel(), pipeline.embeddingModelRef(), pipeline, "model");
+ String splitterRef = pipeline.documentSplitterRef();
+
+ if (pipeline.directory() != null) {
+ directoryRoute(pipeline, storeRef, modelRef, splitterRef);
+ log.info("Ingestion pipeline '{}': source=file:{}", pipeline.name(), pipeline.directory());
+ } else {
+ consumerRoute(pipeline, storeRef, modelRef, splitterRef);
+ log.info("Ingestion pipeline '{}': source={}", pipeline.name(), URISupport.sanitizeUri(pipeline.uri()));
+ }
+ }
+
+ // ********************************************************************************
+ // The two route topologies
+ // ********************************************************************************
+
+ /**
+ * The route: watch the directory → resolve the id → (parse) → split, embed, store. The register in the file
+ * endpoint keeps unchanged files from re-ingesting; an edited file gets a new key and re-ingests, its old segments
+ * remain.
+ */
+ private void directoryRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) {
+ String registerRef = repositoryRef(pipeline, true);
+ Expression documentId = documentIdExpression(pipeline.documentId(), Exchange.FILE_NAME);
+
+ ProcessorDefinition<?> route = from(fileEndpointUri(pipeline, registerRef))
+ .routeId(routeId(pipeline))
+ .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId);
+ // no duplicate pre-check: the endpoint register already filtered duplicates out
+ route = parseSteps(route, pipeline, null);
+ // the file consumer discards the reply, so no repository is passed to the producer: the
+ // endpoint register already keeps the same file version from being ingested twice
+ ProcessorDefinition<?> tail = route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef, null, null));
+ if (pipeline.parser() != null) {
+ // the discarded reply would otherwise hide it: a parse to nothing typically means a
+ // missing Tika parser module or an image-only document, and the file's register key
+ // is committed, so it is not retried until the file changes
+ tail.process(exchange -> {
+ IngestResult result = exchange.getMessage().getBody(IngestResult.class);
+ if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) {
+ log.warn("Ingestion pipeline '{}': document '{}' parsed to no text and was skipped; its key is"
+ + " committed, so it is not retried until the file changes (missing parser module?"
+ + " image-only document?)",
+ pipeline.name(), result.documentId());
+ }
+ });
+ } else {
+ // the discarded reply would otherwise hide even the trace of a blank file
+ tail.process(exchange -> {
+ IngestResult result = exchange.getMessage().getBody(IngestResult.class);
+ if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) {
+ log.debug("Ingestion pipeline '{}': document '{}' contained no text, nothing was written",
+ pipeline.name(), result.documentId());
+ }
+ });
+ }
+ }
+
+ /** The route: consume → resolve the id → (parse) → split, embed, store; the reply is the result. */
+ private void consumerRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) {
+ String scheme = StringHelper.before(pipeline.uri(), ":");
+ requireComponent(scheme, pipeline, "its source");
+ String registerRef = repositoryRef(pipeline, false);
+ Expression documentId = documentIdExpression(pipeline.documentId(), LangChain4jIngestHeaders.DOCUMENT_ID);
+ IdempotentRepository register = registerRef == null
+ ? null
+ : getContext().getRegistry().lookupByNameAndType(registerRef, IdempotentRepository.class);
+
+ ProcessorDefinition<?> route = from(pipeline.uri())
+ .routeId(routeId(pipeline))
+ .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId);
+ route = parseSteps(route, pipeline, register);
+ // deduplication by document id happens inside the producer, so a duplicate is answered
+ // SKIPPED, a blank delivery releases its claim and a failed write can be retried
+ // when the id comes from a plain header, the producer is told its name: the property set
+ // above always wins, but the producer's missing-id error then names the header the route
+ // actually reads instead of the component default
+ String configured = pipeline.documentId();
+ String documentIdHeader = configured != null && !configured.contains("${") && !configured.contains("$simple{")
+ ? configured
+ : null;
+ route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef, registerRef, documentIdHeader));
+ }
+
+ /**
+ * The optional parse stage; the route is returned unchanged when the pipeline has no parser. Tika parses in-process
+ * to its plain-text output — body subtree only, block boundaries newline-separated by Tika's own handlers, the
+ * encoding pinned — and {@link TikaTextDecode} then decodes those bytes as the pinned charset, dodging the exchange
+ * charset heuristic that the parsed document's own Content-Type header could steer. Docling hands the payload to a
+ * Docling Serve instance and answers finished markdown; the body is pinned to bytes and the {@code CamelDocling*}
+ * control headers are swept, because none of them may be decided by a consumer-delivered payload.
+ */
+ private ProcessorDefinition<?> parseSteps(ProcessorDefinition<?> route, IngestPipelineDefinition pipeline,
+ IdempotentRepository register) {
+ if (pipeline.parser() == null) {
+ return route;
+ }
+ // a parser must never run without a captured id: the property would be unset, and the
+ // producer's header fallback would read headers a parsed document may have written -
+ // the previous engine failed such a delivery, and so does this guard
+ route = route.process(exchange -> {
+ String id = exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, String.class);
+ if (id == null || id.isBlank()) {
+ throw new IllegalArgumentException("Ingestion pipeline '" + pipeline.name()
+ + "': the document id resolved to nothing, and a parser pipeline requires it before"
+ + " the parse - a parsed document must not supply its own identity");
+ }
+ });
+ if (register != null) {
+ // advisory: a known duplicate is answered SKIPPED before the (possibly remote) parse
+ // is paid for; the producer's eager claim stays authoritative, so a duplicate racing
+ // this check is still caught there
+ route = route.choice()
+ .when(exchange -> register
+ .contains(exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, String.class)))
+ .process(exchange -> exchange.getMessage().setBody(new IngestResult(pipeline.name(),
+ exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, String.class), 0,
+ IngestResult.Outcome.SKIPPED)))
+ .stop()
+ .end();
+ }
+ if (pipeline.maxDocumentSize() > 0) {
+ // the endpoint's own cap counts extracted characters, which protects the splitter and
+ // the model but not the parse: this guard rejects the raw payload first, before tika
+ // or docling materialize it
+ route = route.process(rawSizeGuard(pipeline));
+ }
+ return switch (pipeline.parser()) {
+ case TIKA -> route.to("tika:parse?tikaParseOutputFormat=text&tikaParseOutputEncoding=UTF-8")
+ .process(new TikaTextDecode());
+ case DOCLING -> route.convertBodyTo(byte[].class)
+ .removeHeaders("CamelDocling*")
+ .to("docling:convert?operation=CONVERT_TO_MARKDOWN&contentInBody=true");
+ };
+ }
+
+ private static Processor rawSizeGuard(IngestPipelineDefinition pipeline) {
+ // the declared-length header spares even the read, but only the directory pipeline's own
+ // file consumer is trusted to have set it: on a consumer pipeline every header may be
+ // attacker-supplied along with the payload, so its body is always measured - a forged
+ // CamelFileLength must not talk an oversized payload past the guard and into the parser
+ boolean trustDeclaredLength = pipeline.directory() != null;
+ return exchange -> {
+ Long declared = trustDeclaredLength
+ ? exchange.getMessage().getHeader(Exchange.FILE_LENGTH, Long.class)
+ : null;
+ long size;
+ byte[] bounded = null;
+ if (declared != null) {
+ size = declared;
+ } else {
+ // every body is measured through a stream - a GenericFile streams from disk, a
+ // byte[] merely wraps - so an attacker-sized payload is rejected after
+ // maxDocumentSize + 1 bytes instead of being materialized whole in the heap just
+ // to be measured; an accepted stream is consumed here, so the bytes replace it
+ // as the body. A null body carries no bytes to guard; it flows on and becomes
+ // the EMPTY outcome
+ InputStream stream = exchange.getMessage().getBody(InputStream.class);
+ if (stream == null) {
+ size = 0;
+ } else {
+ int limit = pipeline.maxDocumentSize() == Integer.MAX_VALUE
+ ? Integer.MAX_VALUE
+ : pipeline.maxDocumentSize() + 1;
+ bounded = stream.readNBytes(limit);
+ size = bounded.length;
+ }
+ }
+ if (size > pipeline.maxDocumentSize()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline.name() + "': document '"
+ + exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY,
+ String.class)
+ + "' exceeds maxDocumentSize (" + size + " > "
+ + pipeline.maxDocumentSize() + " bytes)");
+ }
+ if (bounded != null) {
+ exchange.getMessage().setBody(bounded);
+ }
+ };
+ }
+
+ // ********************************************************************************
+ // Helper methods
+ // ********************************************************************************
+
+ /**
+ * The directory consumer. Built through {@code createQueryString} rather than concatenated, so no option can be
+ * injected through a crafted value. {@code noop} leaves the documents where they are, the register keeps the same
+ * file version from being ingested twice, and the {@code changed} read lock waits for a file still being copied in.
+ */
+ private static String fileEndpointUri(IngestPipelineDefinition pipeline, String registerRef) {
+ Map<String, Object> options = new LinkedHashMap<>();
+ options.put("noop", "true");
+ options.put("idempotent", "true");
+ options.put("idempotentRepository", "#bean:" + registerRef);
+ // an edited file gets a new key and re-ingests; old segments remain (append)
+ options.put("idempotentKey", "${file:absolute.path}:${file:modified}:${file:size}");
+ options.put("recursive", String.valueOf(pipeline.recursive()));
+ options.put("readLock", "changed");
+ if (pipeline.parser() == null) {
+ // text is read as UTF-8; a parser receives the raw bytes instead - the format is its
+ // business, and a charset conversion would corrupt a binary document
+ options.put("charset", "UTF-8");
+ }
+ return "file:" + pipeline.directory() + "?" + queryString(options);
+ }
+
+ private static String ingestEndpointUri(
+ IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef, String registerRef,
+ String documentIdHeader) {
+ Map<String, Object> options = new LinkedHashMap<>();
+ if (documentIdHeader != null) {
+ options.put("documentIdHeader", documentIdHeader);
+ }
+ options.put("maxSegmentSize", String.valueOf(pipeline.maxSegmentSize()));
+ options.put("maxOverlapSize", String.valueOf(pipeline.maxOverlapSize()));
+ options.put("embeddingBatchSize", String.valueOf(pipeline.embeddingBatchSize()));
+ if (pipeline.maxDocumentSize() > 0) {
+ options.put("maxDocumentSize", String.valueOf(pipeline.maxDocumentSize()));
+ }
+ if (splitterRef != null) {
+ options.put("documentSplitter", "#bean:" + splitterRef);
+ }
+ if (storeRef != null) {
+ options.put("embeddingStore", "#bean:" + storeRef);
+ }
+ if (modelRef != null) {
+ options.put("embeddingModel", "#bean:" + modelRef);
+ }
+ if (registerRef != null) {
+ options.put("idempotentRepository", "#bean:" + registerRef);
+ }
+ return LangChain4jIngest.SCHEME + ":" + pipeline.name() + "?" + queryString(options);
+ }
+
+ private static String queryString(Map<String, Object> options) {
+ return URISupport.createQueryString(options);
+ }
+
+ /**
+ * A bare header name is read as a header directly rather than parsed: a dotted header name would send the simple
+ * parser into OGNL. A value containing {@code ${} or {@code $simple{} is a simple-language expression. The
+ * expression is initialised here, at route build time — left to reify lazily it would race on the first concurrent
+ * exchanges.
+ */
+ private Expression documentIdExpression(String configured, String defaultHeader) {
+ Expression expression = configured == null
+ ? ExpressionBuilder.headerExpression(defaultHeader)
+ : configured.contains("${") || configured.contains("$simple{")
+ ? ExpressionBuilder.simpleExpression(configured)
+ : ExpressionBuilder.headerExpression(configured);
+ expression.init(getContext());
+ return expression;
+ }
+
+ /**
+ * Resolves the pipeline's duplicate register to a registry reference: a named bean, an auto-created in-memory
+ * register — the existing bean wins — or, when a directory pipeline names none, the built-in one.
+ */
+ private String repositoryRef(IngestPipelineDefinition pipeline, boolean required) {
+ String ref = pipeline.idempotentRepositoryRef();
+ if (pipeline.idempotentRepositoryAutoCreate()) {
+ if (ref == null) {
+ throw new IllegalStateException(
+ "Ingestion pipeline '" + pipeline.name() + "' sets idempotentRepositoryAutoCreate but no"
+ + " idempotentRepository name to create the register under.");
+ }
+ if (getContext().getRegistry().lookupByNameAndType(ref, IdempotentRepository.class) == null) {
+ getContext().getRegistry().bind(ref,
+ MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY));
+ }
+ }
+ if (ref == null && required) {
+ ref = generatedName(pipeline, "register");
+ if (getContext().getRegistry().lookupByNameAndType(ref, IdempotentRepository.class) == null) {
+ getContext().getRegistry().bind(ref,
+ MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_REGISTER_CAPACITY));
+ }
+ }
+ return ref;
+ }
+
+ /**
+ * Binds a definition-supplied instance to the registry, so the endpoint can reference it;
+ * called from {@code configure()}, which Camel runs single-threaded during startup.
+ */
+ private String bindInstance(Object instance, String ref, IngestPipelineDefinition pipeline, String what) {
+ if (instance == null) {
+ return ref;
+ }
+ String name = ref != null ? ref : generatedName(pipeline, what);
+ getContext().getRegistry().bind(name, instance);
+ return name;
+ }
+
+ private void requireParserComponent(IngestPipelineDefinition pipeline) {
+ if (pipeline.parser() == null) {
+ return;
+ }
+ switch (pipeline.parser()) {
+ case TIKA -> requireComponent("tika", pipeline, "its parser");
+ case DOCLING -> requireComponent("docling", pipeline, "its parser");
+ }
+ }
+
+ /**
+ * Fails a pipeline whose component is not on the classpath with the artifact to add, instead of Camel's bare
+ * failed-to-resolve message much later.
+ */
+ private void requireComponent(String scheme, IngestPipelineDefinition pipeline, String purpose) {
+ Component component = null;
+ try {
+ component = getContext().getComponent(scheme, true, false);
+ } catch (Exception e) {
+ // fall through to the actionable message
+ }
+ if (component == null) {
+ throw new IllegalStateException(
+ "Ingestion pipeline '" + pipeline.name() + "' needs the '" + scheme + "' component for " + purpose
+ + ", which is not on the classpath. Add the extension that provides it, e.g. "
+ + "org.apache.camel.quarkus:camel-quarkus-" + scheme);
+ }
+ }
+
+ private static String generatedName(IngestPipelineDefinition pipeline, String what) {
+ return LangChain4jIngest.SCHEME + "-" + pipeline.name() + "-" + what;
+ }
+
+ private static String routeId(IngestPipelineDefinition pipeline) {
+ return LangChain4jIngest.SCHEME + "-" + pipeline.name();
+ }
+}
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java
index 1461e16f52..dff173110d 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java
+++ b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRoutes.java
@@ -16,7 +16,7 @@
*/
package org.apache.camel.quarkus.component.langchain4j.ingest;
-import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
@@ -32,25 +32,18 @@ import jakarta.enterprise.inject.Instance;
import jakarta.enterprise.inject.literal.NamedLiteral;
import jakarta.inject.Inject;
import org.apache.camel.CamelContextAware;
-import org.apache.camel.Exchange;
-import org.apache.camel.Expression;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestResult;
-import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
import org.apache.camel.spi.IdempotentRepository;
-import org.apache.camel.support.builder.ExpressionBuilder;
-import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
-import org.apache.camel.util.URISupport;
import org.jboss.logging.Logger;
-import static org.apache.camel.builder.endpoint.StaticEndpointBuilders.file;
-
/**
- * Generates one Camel route per configured ingestion pipeline. Users never see these routes —
- * they are the implementation of the configuration.
+ * Translates the extension's configuration model — build-time and runtime properties and
+ * {@code @Ingest} builder methods — into upstream {@link IngestPipelineDefinition}s; the routes
+ * themselves are generated by the inherited {@link IngestPipelineRouteBuilder}, and the engine
+ * lives in the {@code camel-langchain4j-ingest} component. What stays here is the Quarkus DX:
+ * CDI bean resolution with its actionable messages, and the configuration-level validations.
*/
@ApplicationScoped
-public class IngestRoutes extends RouteBuilder {
+public class IngestRoutes extends IngestPipelineRouteBuilder {
private static final Logger LOG = Logger.getLogger(IngestRoutes.class);
@@ -73,11 +66,10 @@ public class IngestRoutes extends RouteBuilder {
@Any
Instance<EmbeddingModel> modelCandidates;
- /** Exchange property carrying the resolved document id. */
- private static final String DOCUMENT_ID_PROPERTY = "CamelQuarkusIngestDocumentId";
-
@Override
- public void configure() {
+ protected List<IngestPipelineDefinition> pipelines() {
+ List<IngestPipelineDefinition> definitions = new ArrayList<>();
+
// a pipeline may be declared entirely through runtime properties - the documented
// minimum is a directory and nothing else - so the two config roots are unioned. Keying
// off the build-time map alone would make that configuration a silent no-op, since
@@ -98,13 +90,6 @@ public class IngestRoutes extends RouteBuilder {
continue;
}
- IngestService service = new IngestService(
- name,
- resolveStore(name, pipeline == null ? null : pipeline.embeddingStore().orElse(null)),
- resolveModel(name, pipeline == null ? null : pipeline.embeddingModel().orElse(null)),
- pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_SEGMENT_SIZE : pipeline.maxSegmentSize(),
- pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE : pipeline.maxOverlapSize());
-
// a consumer URI says "consume from this"; its absence says "read that directory"
String uri = pipeline == null ? null : pipeline.source().uri().orElse(null);
if (uri != null && runtime != null && runtime.source().directory().isPresent()) {
@@ -112,29 +97,36 @@ public class IngestRoutes extends RouteBuilder {
+ "') and source.directory ('" + runtime.source().directory().get() + "'). A pipeline "
+ "reads one source: keep the URI, or drop it to read the directory.");
}
- if (uri == null) {
- configureFileSource(name, runtime, service);
- LOG.infof("Ingestion pipeline '%s': source=file", name);
- } else {
- configureEndpointSource(name, uri, runtime, service);
- LOG.infof("Ingestion pipeline '%s': source=%s", name, URISupport.sanitizeUri(uri));
- }
+
+ definitions.add(definition(name, uri, runtime,
+ resolveStore(name, pipeline == null ? null : pipeline.embeddingStore().orElse(null)),
+ resolveModel(name, pipeline == null ? null : pipeline.embeddingModel().orElse(null)),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_SEGMENT_SIZE : pipeline.maxSegmentSize(),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_OVERLAP_SIZE : pipeline.maxOverlapSize(),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_EMBEDDING_BATCH_SIZE : pipeline.embeddingBatchSize(),
+ pipeline == null ? IngestBuildTimeConfig.DEFAULT_MAX_DOCUMENT_SIZE : pipeline.maxDocumentSize(),
+ pipeline == null ? null : pipeline.documentSplitter().orElse(null),
+ pipeline == null ? null : pipeline.parser().orElse(null)));
}
for (IngestBuilderPipelines.Entry entry : builderPipelines.entries()) {
- configureBuilderPipeline(entry);
+ IngestPipelineDefinition definition = builderPipeline(entry);
+ if (definition != null) {
+ definitions.add(definition);
+ }
}
+ return definitions;
}
/** An {@code @Ingest}-declared pipeline: the builder twin of the configuration path. */
- private void configureBuilderPipeline(IngestBuilderPipelines.Entry entry) {
+ private IngestPipelineDefinition builderPipeline(IngestBuilderPipelines.Entry entry) {
String name = entry.name();
// configuration can still switch a builder-declared pipeline off, and the check precedes
// the invocation so a disabled pipeline's method never runs
IngestRunTimeConfig.PipelineRunTimeConfig external = runTimeConfig.pipelines().get(name);
if (external != null && !external.enabled()) {
LOG.infof("Ingestion pipeline '%s' (builder) is disabled", name);
- return;
+ return null;
}
// enabled is the one thing configuration may say about a builder pipeline; anything about
// its source would be quietly overruled by the @Ingest method, so it is an error instead
@@ -150,150 +142,87 @@ public class IngestRoutes extends RouteBuilder {
}
IngestPipeline definition = builderPipelines.definition(entry);
- IngestRunTimeConfig.PipelineRunTimeConfig runtime = definition.asRunTimeConfig();
-
- IngestService service = new IngestService(
- name,
+ String uri = "file".equals(definition.sourceType()) ? null : definition.sourceUri();
+ return definition(name, uri, definition.asRunTimeConfig(),
resolveStore(name, definition.embeddingStoreName().orElse(null)),
resolveModel(name, definition.embeddingModelName().orElse(null)),
definition.maxSegmentSize(),
- definition.maxOverlapSize());
-
- switch (definition.sourceType()) {
- case "file" -> configureFileSource(name, runtime, service);
- case "endpoint" -> configureEndpointSource(name, definition.sourceUri(), runtime, service);
- default -> throw new IllegalStateException("Unknown source type " + definition.sourceType());
- }
-
- LOG.infof("Ingestion pipeline '%s' (builder): source=%s", name,
- URISupport.sanitizeUri(definition.sourceUri() == null ? definition.sourceType() : definition.sourceUri()));
- }
-
- private void configureFileSource(String name, IngestRunTimeConfig.PipelineRunTimeConfig runtime,
- IngestService service) {
- String directory = required(name, runtime == null ? null : runtime.source().directory().orElse(null),
- "source.directory");
- // built with the Endpoint DSL rather than concatenated: a directory containing ? # & or a
- // space would otherwise mis-parse, and a crafted one could inject options - delete=true
- // is honoured ahead of noop and would delete the user's documents after reading them.
- // noop leaves the documents where they are (a knowledge base reads its source, it does
- // not consume it), idempotent keeps the same file from being ingested twice, and the
- // changed read lock waits for a file still being copied in rather than embedding half
- // of it
- Expression documentId = documentIdExpression(runtime, Exchange.FILE_NAME);
- maybeAutoCreateRepository(name, runtime);
- String repositoryName = runtime == null ? null : runtime.source().idempotentRepository().orElse(null);
- var endpoint = file(directory)
- .noop(true)
- .idempotent(true);
- if (repositoryName != null) {
- endpoint.idempotentRepository(resolveRepository(name, repositoryName));
- } else {
- // in-memory default, sized above Camel's 1000-entry cap so eviction does not
- // re-ingest large directories; lost on restart
- endpoint.idempotentRepository(MemoryIdempotentRepository.memoryIdempotentRepository(100_000));
- }
- from(endpoint
- // an edited file gets a new key and re-ingests; old segments remain (append)
- .idempotentKey("${file:absolute.path}:${file:modified}:${file:size}")
- .recursive(runtime.source().recursive())
- .readLock("changed")
- .charset(StandardCharsets.UTF_8.name()))
- .routeId(routeId(name))
- .process(exchange -> {
- IngestResult result = service.ingest(documentId.evaluate(exchange, String.class),
- exchange.getIn().getBody(String.class));
- // the file consumer discards the result, so an EMPTY outcome would
- // otherwise leave no trace at all
- if (result.outcome() == IngestResult.Outcome.EMPTY) {
- LOG.debugf("Ingestion pipeline '%s': document '%s' contained no text, nothing was written",
- name, result.documentId());
- }
- });
+ definition.maxOverlapSize(),
+ definition.embeddingBatchSize(),
+ definition.maxDocumentSize(),
+ definition.documentSplitterName().orElse(null),
+ definition.parser().orElse(null));
}
/**
- * The escape hatch: any Camel consumer feeds the pipeline. Which part of the exchange
- * identifies the document is the consumer's business, so {@code source.document-id} says it —
- * {@code ${header.CamelAwsS3Key}} for an S3 consumer, the message header otherwise.
+ * One creation path for both declaration styles, mirroring how a builder pipeline provides
+ * the same runtime-configuration view the configuration path reads.
*/
- private void configureEndpointSource(String name, String uri,
- IngestRunTimeConfig.PipelineRunTimeConfig runtime, IngestService service) {
- Expression documentId = documentIdExpression(runtime, IngestHeaders.DOCUMENT_ID);
- maybeAutoCreateRepository(name, runtime);
- String repositoryName = runtime == null ? null : runtime.source().idempotentRepository().orElse(null);
- if (repositoryName == null) {
- from(uri)
- .routeId(routeId(name))
- .process(exchange -> {
- String id = requireDocumentId(name, documentId, exchange);
- exchange.getIn().setBody(service.ingest(id, exchange.getIn().getBody(String.class)));
- });
- return;
+ private IngestPipelineDefinition definition(String name, String uri,
+ IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ EmbeddingStore<TextSegment> store, EmbeddingModel model,
+ int maxSegmentSize, int maxOverlapSize, int embeddingBatchSize, int maxDocumentSize,
+ String documentSplitterName, String parser) {
+
+ IngestPipelineDefinition definition;
+ if (uri == null) {
+ String directory = required(name, runtime == null ? null : runtime.source().directory().orElse(null),
+ "source.directory");
+ definition = IngestPipelineDefinition.directory(name, directory)
+ .recursive(runtime.source().recursive());
+ } else {
+ definition = IngestPipelineDefinition.consumer(name, uri);
}
- // duplicates skip the block and the tail processor answers SKIPPED; first write wins
- // per id. The EIP keys on the validated id property, evaluating the expression once
- IdempotentRepository repository = resolveRepository(name, repositoryName);
- from(uri)
- .routeId(routeId(name))
- .process(exchange -> exchange.setProperty(DOCUMENT_ID_PROPERTY,
- requireDocumentId(name, documentId, exchange)))
- .idempotentConsumer(exchangeProperty(DOCUMENT_ID_PROPERTY), repository)
- .process(exchange -> {
- String id = (String) exchange.getProperty(DOCUMENT_ID_PROPERTY);
- IngestResult result = service.ingest(id, exchange.getIn().getBody(String.class));
- if (result.outcome() == IngestResult.Outcome.EMPTY) {
- // a blank document wrote nothing, so it must not keep the eager claim
- // on the id - a later, populated delivery under the same id would be
- // answered SKIPPED. The completion-time confirm() of the removed key
- // is a no-op in the memory, file and JDBC repositories alike
- repository.remove(id);
- }
- exchange.getIn().setBody(result);
- })
- .end()
- .process(exchange -> {
- if (exchange.getProperty(Exchange.DUPLICATE_MESSAGE, false, Boolean.class)) {
- exchange.getIn().setBody(new IngestResult(name,
- (String) exchange.getProperty(DOCUMENT_ID_PROPERTY), 0,
- IngestResult.Outcome.SKIPPED));
- }
- });
- }
-
- private static String requireDocumentId(String name, Expression documentId, Exchange exchange) {
- String id = documentId.evaluate(exchange, String.class);
- if (id == null) {
- throw new IllegalArgumentException("Ingestion pipeline '" + name + "': no document id. "
- + "Set the " + IngestHeaders.DOCUMENT_ID + " header, or point "
- + "quarkus.camel.langchain4j.ingest." + name + ".source.document-id at where the "
- + "consumer puts it.");
+ definition.embeddingStore(store)
+ .embeddingModel(model)
+ .splitter(maxSegmentSize, maxOverlapSize)
+ .embeddingBatchSize(embeddingBatchSize);
+ if (maxDocumentSize > 0) {
+ definition.maxDocumentSize(maxDocumentSize);
+ }
+ if (documentSplitterName != null) {
+ definition.documentSplitter(documentSplitterName);
+ }
+ if (parser != null) {
+ definition.parser(parser);
}
- return id;
- }
- private Expression documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime,
- String defaultHeader) {
- String configured = runtime == null ? null : runtime.source().documentId().orElse(null);
- if (configured == null) {
- return ExpressionBuilder.headerExpression(defaultHeader);
+ String documentId = runtime == null ? null : runtime.source().documentId().orElse(null);
+ if (documentId != null) {
+ definition.documentId(documentId);
+ } else if (uri != null) {
+ // the upstream default header follows the design/headers.adoc convention
+ // (CamelLangChain4jIngestDocumentId); this extension released the contract as
+ // CamelIngestDocumentId in 3.39.0, so the legacy name is pinned here - the
+ // compatibility burden stays downstream, where it was created
+ definition.documentId(IngestHeaders.DOCUMENT_ID);
}
- // a bare header name is read as a header directly rather than parsed: a dotted header
- // name would send the simple parser into OGNL. $simple{...} is the form for a properties
- // file, where MicroProfile Config would consume a ${...} before Camel ever saw it; the
- // ${...} form still serves the Java builder. Both tokens are matched with contains, the
- // same way Camel's own LanguageSupport detects a simple function. The expression is
- // initialised here, at route build time - left to reify lazily it would race on the
- // first concurrent exchanges
- Expression expression = configured.contains("${") || configured.contains("$simple{")
- ? ExpressionBuilder.simpleExpression(configured)
- : ExpressionBuilder.headerExpression(configured);
- expression.init(getContext());
- return expression;
- }
- private static String routeId(String name) {
- return "camel-quarkus-langchain4j-ingest-" + name;
+ String repositoryName = runtime == null ? null : runtime.source().idempotentRepository().orElse(null);
+ boolean autoCreate = runtime != null && runtime.source().idempotentRepositoryAutoCreate();
+ if (autoCreate && repositoryName == null) {
+ throw new IllegalStateException("Ingestion pipeline '" + name
+ + "' sets source.idempotent-repository-auto-create but no "
+ + "source.idempotent-repository name to create the register under.");
+ }
+ if (repositoryName != null) {
+ IdempotentRepository repository = getContext().getRegistry().lookupByNameAndType(repositoryName,
+ IdempotentRepository.class);
+ if (repository == null && !autoCreate) {
+ // the existence check up front, with the configuration-level message
+ throw new IllegalStateException("Ingestion pipeline '" + name
+ + "' references idempotent repository '" + repositoryName + "' but no such bean exists");
+ }
+ if (repository != null) {
+ // whether auto-create is set or not, a pre-existing CDI-produced repository wins
+ // and does not pass through the registry's bind hook, so a CamelContextAware
+ // implementation would otherwise run contextless
+ CamelContextAware.trySetCamelContext(repository, getContext());
+ }
+ definition.idempotentRepository(repositoryName)
+ .idempotentRepositoryAutoCreate(autoCreate);
+ }
+ return definition;
}
private static String required(String name, String value, String property) {
@@ -319,48 +248,6 @@ public class IngestRoutes extends RouteBuilder {
* discovered first. A raw-typed registry search cannot serve here: it never matches a bean
* typed {@code EmbeddingStore<TextSegment>}.
*/
- /**
- * Binds an in-memory register under the configured name, unless a bean with that name
- * already exists — {@code camel.beans.*} beans are bound before route builders run, so both
- * they and CDI beans are visible here and win.
- */
- private void maybeAutoCreateRepository(String name, IngestRunTimeConfig.PipelineRunTimeConfig runtime) {
- if (runtime == null || !runtime.source().idempotentRepositoryAutoCreate()) {
- return;
- }
- String repositoryName = runtime.source().idempotentRepository().orElse(null);
- if (repositoryName == null) {
- throw new IllegalStateException("Ingestion pipeline '" + name
- + "' sets source.idempotent-repository-auto-create but no "
- + "source.idempotent-repository name to create the register under.");
- }
- if (getContext().getRegistry().lookupByNameAndType(repositoryName, IdempotentRepository.class) != null) {
- LOG.infof("Ingestion pipeline '%s': idempotent repository '%s' already exists, auto-create skipped",
- name, repositoryName);
- return;
- }
- getContext().getRegistry().bind(repositoryName,
- MemoryIdempotentRepository.memoryIdempotentRepository(100_000));
- }
-
- /**
- * Resolves the configured register from the Camel registry: CDI producers, camel.beans
- * definitions and auto-created registers alike. By name only — the application may hold
- * unrelated idempotent repositories.
- */
- private IdempotentRepository resolveRepository(String name, String repositoryName) {
- IdempotentRepository repository = getContext().getRegistry().lookupByNameAndType(repositoryName,
- IdempotentRepository.class);
- if (repository == null) {
- throw new IllegalStateException("Ingestion pipeline '" + name + "' references idempotent repository '"
- + repositoryName + "' but no such bean exists");
- }
- // a CDI-produced repository does not pass through the registry's bind hook, so a
- // CamelContextAware implementation would otherwise run contextless
- CamelContextAware.trySetCamelContext(repository, getContext());
- return repository;
- }
-
private <T> T resolve(String name, Instance<T> candidates, String configured, String what, String property) {
if (configured != null) {
Instance<T> named = candidates.select(NamedLiteral.of(configured));
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestService.java b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestService.java
deleted file mode 100644
index c2061e2714..0000000000
--- a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestService.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.quarkus.component.langchain4j.ingest.core;
-
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
-import dev.langchain4j.data.document.Document;
-import dev.langchain4j.data.document.DocumentSplitter;
-import dev.langchain4j.data.document.Metadata;
-import dev.langchain4j.data.document.splitter.DocumentSplitters;
-import dev.langchain4j.data.segment.TextSegment;
-import dev.langchain4j.model.embedding.EmbeddingModel;
-import dev.langchain4j.store.embedding.EmbeddingStore;
-import org.jboss.logging.Logger;
-
-/**
- * Splits a document, embeds the segments and writes them to the store.
- *
- * <p>
- * Deliberately naive: it writes whatever it is given and remembers nothing, so ingesting a
- * document twice leaves two copies. Keeping a store in step with a changing source — skipping
- * unchanged documents, replacing changed ones, removing deleted ones — needs a record of what was
- * written, and that engine replaces this class.
- */
-public class IngestService {
-
- private static final Logger LOG = Logger.getLogger(IngestService.class);
-
- // the runtime-neutral names of the camel-langchain4j-ingest component (Camel 4.23), which
- // this engine will delegate to - stores written from 3.40.0 on are compatible with it as-is
- public static final String METADATA_PIPELINE = "camel_ingest_pipeline";
- public static final String METADATA_DOCUMENT_ID = "camel_ingest_document_id";
-
- private static final int EMBEDDING_BATCH_SIZE = 32;
-
- private final String pipeline;
- private final EmbeddingStore<TextSegment> store;
- private final EmbeddingModel model;
- private final DocumentSplitter splitter;
-
- public IngestService(String pipeline, EmbeddingStore<TextSegment> store, EmbeddingModel model,
- int maxSegmentSize, int maxOverlapSize) {
- this.pipeline = pipeline;
- this.store = store;
- this.model = model;
- this.splitter = DocumentSplitters.recursive(maxSegmentSize, maxOverlapSize);
- }
-
- public IngestResult ingest(String documentId, String text) {
- if (documentId == null || documentId.isBlank()) {
- throw new IllegalArgumentException("Ingestion pipeline '" + pipeline + "': documentId is required");
- }
- if (text == null || text.isBlank()) {
- return new IngestResult(pipeline, documentId, 0, IngestResult.Outcome.EMPTY);
- }
-
- // the document id travels with every segment: retrieval can cite it, and the engine that
- // replaces this one needs it to find a document's vectors again
- Map<String, Object> metadata = new LinkedHashMap<>();
- metadata.put(METADATA_PIPELINE, pipeline);
- metadata.put(METADATA_DOCUMENT_ID, documentId);
-
- List<TextSegment> segments = splitter.split(Document.from(text, Metadata.from(metadata)));
- // embedded in batches: a single embedAll over a large document's full segment list can
- // exceed an embedding provider's per-request limits
- for (int from = 0; from < segments.size(); from += EMBEDDING_BATCH_SIZE) {
- List<TextSegment> batch = segments.subList(from, Math.min(from + EMBEDDING_BATCH_SIZE, segments.size()));
- store.addAll(model.embedAll(batch).content(), batch);
- }
-
- LOG.debugf("Ingestion pipeline '%s': wrote %d segment(s) of document '%s'", pipeline, segments.size(),
- documentId);
- return new IngestResult(pipeline, documentId, segments.size(), IngestResult.Outcome.INGESTED);
- }
-
- public String pipeline() {
- return pipeline;
- }
-}
diff --git a/integration-tests/langchain4j-ingest/pom.xml b/integration-tests/langchain4j-ingest/pom.xml
index 125b77664e..f0027af7a3 100644
--- a/integration-tests/langchain4j-ingest/pom.xml
+++ b/integration-tests/langchain4j-ingest/pom.xml
@@ -59,10 +59,24 @@
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-kafka</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-docling</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-sql</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-tika</artifactId>
+ </dependency>
+ <!-- tika parses only the formats whose parser modules are on the classpath; the reports
+ pipeline ingests PDFs -->
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-parser-pdf-module</artifactId>
+ </dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-jdbc-h2</artifactId>
@@ -89,6 +103,11 @@
<artifactId>camel-quarkus-integration-tests-support-kafka</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-integration-wiremock-support</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
@@ -185,6 +204,19 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-docling-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-kafka-deployment</artifactId>
@@ -224,6 +256,19 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-tika-deployment</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
</profile>
</profiles>
diff --git a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestItProducers.java b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestItProducers.java
index c61430afbc..c93e1fae3f 100644
--- a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestItProducers.java
+++ b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestItProducers.java
@@ -48,6 +48,34 @@ public class IngestItProducers {
return new InMemoryEmbeddingStore<>();
}
+ @Produces
+ @Singleton
+ @Named("htmlfeed-store")
+ EmbeddingStore<TextSegment> htmlfeedStore() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
+ @Produces
+ @Singleton
+ @Named("reports-store")
+ EmbeddingStore<TextSegment> reportsStore() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
+ @Produces
+ @Singleton
+ @Named("scans-store")
+ EmbeddingStore<TextSegment> scansStore() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
+ @Produces
+ @Singleton
+ @Named("capped-store")
+ EmbeddingStore<TextSegment> cappedStore() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
@Produces
@Singleton
@Named("datasheets-store")
diff --git a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java
index 3d8c3b8fad..ca2add6199 100644
--- a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java
+++ b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestResource.java
@@ -35,9 +35,9 @@ import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.MediaType;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.langchain4j.ingest.IngestResult;
+import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest;
import org.apache.camel.quarkus.component.langchain4j.ingest.IngestHeaders;
-import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestResult;
-import org.apache.camel.quarkus.component.langchain4j.ingest.core.IngestService;
import org.apache.camel.spi.IdempotentRepository;
import org.eclipse.microprofile.config.inject.ConfigProperty;
@@ -56,6 +56,10 @@ public class IngestResource {
@Named("custom-store")
EmbeddingStore<TextSegment> customStore;
+ @Inject
+ @Named("htmlfeed-store")
+ EmbeddingStore<TextSegment> htmlfeedStore;
+
@Inject
@Named("datasheets-store")
EmbeddingStore<TextSegment> datasheetsStore;
@@ -68,6 +72,18 @@ public class IngestResource {
@Named("events-store")
EmbeddingStore<TextSegment> eventsStore;
+ @Inject
+ @Named("reports-store")
+ EmbeddingStore<TextSegment> reportsStore;
+
+ @Inject
+ @Named("scans-store")
+ EmbeddingStore<TextSegment> scansStore;
+
+ @Inject
+ @Named("capped-store")
+ EmbeddingStore<TextSegment> cappedStore;
+
@Inject
ProducerTemplate producerTemplate;
@@ -77,6 +93,15 @@ public class IngestResource {
@ConfigProperty(name = "ingest.test.directory")
String directory;
+ @ConfigProperty(name = "ingest.reports.directory")
+ String reportsDirectory;
+
+ @ConfigProperty(name = "ingest.scans.directory")
+ String scansDirectory;
+
+ @ConfigProperty(name = "ingest.capped.directory")
+ String cappedDirectory;
+
/** Asserts a key was committed; registry lookup by name, the same way the pipelines resolve. */
@GET
@jakarta.ws.rs.Path("/register-contains")
@@ -97,15 +122,34 @@ public class IngestResource {
Files.writeString(dir.resolve(name), content);
}
+ /** Writes a binary document into a parser pipeline's watched directory. */
+ @POST
+ @jakarta.ws.rs.Path("/binary/{pipeline}/{name}")
+ @Consumes(MediaType.APPLICATION_OCTET_STREAM)
+ public void writeBinary(@PathParam("pipeline") String pipeline, @PathParam("name") String name,
+ byte[] content) throws Exception {
+ Path dir = Path.of(switch (pipeline) {
+ case "scans" -> scansDirectory;
+ case "capped" -> cappedDirectory;
+ default -> reportsDirectory;
+ });
+ Files.createDirectories(dir);
+ Files.write(dir.resolve(name), content);
+ }
+
@GET
@jakarta.ws.rs.Path("/search")
@Produces(MediaType.APPLICATION_JSON)
public List<SearchHit> search(@QueryParam("q") String query, @QueryParam("store") String storeName) {
EmbeddingStore<TextSegment> store = switch (storeName == null ? "products" : storeName) {
case "custom" -> customStore;
+ case "htmlfeed" -> htmlfeedStore;
case "datasheets" -> datasheetsStore;
case "s3" -> s3Store;
case "events" -> eventsStore;
+ case "reports" -> reportsStore;
+ case "capped" -> cappedStore;
+ case "scans" -> scansStore;
default -> productsStore;
};
// the deterministic test model gives a query no semantic pull towards any document, so
@@ -119,8 +163,8 @@ public class IngestResource {
return result.matches().stream()
.map(match -> new SearchHit(
match.embedded().text(),
- match.embedded().metadata().getString(IngestService.METADATA_PIPELINE),
- match.embedded().metadata().getString(IngestService.METADATA_DOCUMENT_ID)))
+ match.embedded().metadata().getString(LangChain4jIngest.METADATA_PIPELINE),
+ match.embedded().metadata().getString(LangChain4jIngest.METADATA_DOCUMENT_ID)))
.toList();
}
@@ -142,4 +186,18 @@ public class IngestResource {
return result.outcome().label();
}
+ /** Feeds a pipeline without any document id, so tests can assert the pre-parse id guard. */
+ @POST
+ @jakarta.ws.rs.Path("/feed-anonymous/{pipeline}")
+ @Consumes(MediaType.TEXT_PLAIN)
+ @Produces(MediaType.TEXT_PLAIN)
+ public String feedAnonymous(@PathParam("pipeline") String pipeline, String content) {
+ try {
+ producerTemplate.requestBody("direct:" + pipeline + "-feed", content, IngestResult.class);
+ return "ingested";
+ } catch (Exception e) {
+ return "failed";
+ }
+ }
+
}
diff --git a/integration-tests/langchain4j-ingest/src/main/resources/application.properties b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
index f8caf3a942..17eaaf0b7e 100644
--- a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
+++ b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
@@ -32,6 +32,48 @@ camel.beans.propertiesRegister=\#class\:org.apache.camel.support.processor.idemp
camel.beans.propertiesRegister.cacheSize=5000
quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=propertiesRegister
+# parser pipelines: reports extracts PDF text in-process (tika); scans converts through a
+# Docling Serve instance (docling), which the tests stub with WireMock - see
+# IngestDoclingTestResource. scans stays off until that resource provides the serve URL
+ingest.reports.directory=target/ingest-reports
+quarkus.camel.langchain4j.ingest.reports.source.directory=${ingest.reports.directory}
+quarkus.camel.langchain4j.ingest.reports.parser=tika
+quarkus.camel.langchain4j.ingest.reports.embedding-store=reports-store
+quarkus.camel.langchain4j.ingest.reports.embedding-model=test-model
+quarkus.camel.langchain4j.ingest.reports.max-segment-size=120
+quarkus.camel.langchain4j.ingest.reports.max-overlap-size=20
+
+# parser pipeline with a tight raw-byte cap: exercises the pre-parse size guard
+ingest.capped.directory=target/ingest-capped
+quarkus.camel.langchain4j.ingest.capped.source.directory=${ingest.capped.directory}
+quarkus.camel.langchain4j.ingest.capped.parser=tika
+quarkus.camel.langchain4j.ingest.capped.max-document-size=200
+quarkus.camel.langchain4j.ingest.capped.embedding-store=capped-store
+quarkus.camel.langchain4j.ingest.capped.embedding-model=test-model
+quarkus.camel.langchain4j.ingest.capped.max-segment-size=120
+quarkus.camel.langchain4j.ingest.capped.max-overlap-size=20
+
+ingest.scans.directory=target/ingest-scans
+quarkus.camel.langchain4j.ingest.scans.source.directory=${ingest.scans.directory}
+quarkus.camel.langchain4j.ingest.scans.parser=docling
+quarkus.camel.langchain4j.ingest.scans.embedding-store=scans-store
+quarkus.camel.langchain4j.ingest.scans.embedding-model=test-model
+quarkus.camel.langchain4j.ingest.scans.max-segment-size=120
+quarkus.camel.langchain4j.ingest.scans.max-overlap-size=20
+quarkus.camel.langchain4j.ingest.scans.enabled=false
+
+camel.component.docling.use-docling-serve=true
+camel.component.docling.docling-serve-url=${docling.serve.url:http://localhost:5001}
+
+# consumer-fed pipeline with a parser: the endpoint+parser combination, and the pre-parse
+# document-id capture (a parsed document cannot forge its own id through Tika metadata headers)
+quarkus.camel.langchain4j.ingest.htmlfeed.source.uri=direct:htmlfeed-feed
+quarkus.camel.langchain4j.ingest.htmlfeed.parser=tika
+quarkus.camel.langchain4j.ingest.htmlfeed.embedding-store=htmlfeed-store
+quarkus.camel.langchain4j.ingest.htmlfeed.embedding-model=test-model
+quarkus.camel.langchain4j.ingest.htmlfeed.max-segment-size=120
+quarkus.camel.langchain4j.ingest.htmlfeed.max-overlap-size=20
+
# consumer-fed pipeline; the document-id exercises the $simple{...} form and the register gives
# first-write-wins dedup by document id
quarkus.camel.langchain4j.ingest.custom.source.uri=direct:custom-feed
diff --git a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestDoclingTestResource.java b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestDoclingTestResource.java
new file mode 100644
index 0000000000..b86d60442b
--- /dev/null
+++ b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/IngestDoclingTestResource.java
@@ -0,0 +1,87 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.it;
+
+import java.util.Map;
+
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
+import org.apache.camel.quarkus.test.mock.backend.MockBackendUtils;
+import org.eclipse.microprofile.config.ConfigProvider;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+
+/**
+ * Provides the Docling Serve endpoint for the {@code scans} pipeline and switches the pipeline
+ * on. By default the docling-serve convert API is stubbed with an in-process WireMock server —
+ * the real container image is a multi-gigabyte download, about 5 GB compressed for the pinned
+ * v1.29.0 — and answers every conversion with the sentence
+ * {@link Langchain4jIngestDoclingTest} ingests. The real container starts only with
+ * {@code -Dcamel.quarkus.start-mock-backend=false}, the same convention the docling integration
+ * tests use; the test document carries the same sentence, so the assertions hold there too.
+ */
+public class IngestDoclingTestResource implements QuarkusTestResourceLifecycleManager {
+
+ static final String CONVERTED_MARKDOWN = "The EPSILON-2 valve seals at 80 bar.";
+
+ private static final int CONTAINER_PORT = 5001;
+
+ private WireMockServer server;
+ private GenericContainer<?> container;
+
+ @Override
+ public Map<String, String> start() {
+ String doclingServeUrl;
+ if (MockBackendUtils.startMockBackend()) {
+ server = new WireMockServer(WireMockConfiguration.options().dynamicPort());
+ server.stubFor(post(urlEqualTo("/v1/convert/source"))
+ .willReturn(aResponse()
+ .withHeader("Content-Type", "application/json")
+ .withBody("{\"document\":{\"filename\":\"valve.pdf\",\"md_content\":\""
+ + CONVERTED_MARKDOWN + "\",\"json_content\":null,\"html_content\":null,"
+ + "\"text_content\":null,\"doctags_content\":null},\"status\":\"success\","
+ + "\"errors\":[],\"processing_time\":0.01,\"timings\":{}}")));
+ server.start();
+ doclingServeUrl = "http://localhost:" + server.port();
+ } else {
+ container = new GenericContainer<>(
+ ConfigProvider.getConfig().getValue("docling.container.image", String.class))
+ .withExposedPorts(CONTAINER_PORT)
+ .waitingFor(Wait.forListeningPort());
+ container.start();
+ doclingServeUrl = "http://" + container.getHost() + ":" + container.getMappedPort(CONTAINER_PORT);
+ }
+ return Map.of(
+ "docling.serve.url", doclingServeUrl,
+ "quarkus.camel.langchain4j.ingest.scans.enabled", "true");
+ }
+
+ @Override
+ public void stop() {
+ if (server != null) {
+ server.stop();
+ }
+ if (container != null) {
+ container.stop();
+ }
+ }
+}
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestDoclingIT.java
similarity index 57%
rename from extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java
rename to integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestDoclingIT.java
index 45e4d86799..40b4af8759 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java
+++ b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestDoclingIT.java
@@ -14,26 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.quarkus.component.langchain4j.ingest.core;
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
-import java.util.Locale;
+import io.quarkus.test.junit.QuarkusIntegrationTest;
-/**
- * Outcome of one ingestion.
- */
-public record IngestResult(String pipeline, String documentId, int segmentsWritten, Outcome outcome) {
-
- public enum Outcome {
- /** Segments written. */
- INGESTED,
- /** Blank document, nothing written. */
- EMPTY,
- /** Already ingested under the same key, nothing written. */
- SKIPPED;
-
- /** The stable wire/log form. */
- public String label() {
- return name().toLowerCase(Locale.ROOT);
- }
- }
+@QuarkusIntegrationTest
+class Langchain4jIngestDoclingIT extends Langchain4jIngestDoclingTest {
}
diff --git a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestDoclingTest.java b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestDoclingTest.java
new file mode 100644
index 0000000000..33546585e1
--- /dev/null
+++ b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestDoclingTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.it;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * The {@code docling} parser: a PDF dropped into the watched directory is converted to markdown
+ * by docling-serve — stubbed with WireMock by default, see {@link IngestDoclingTestResource} —
+ * and the markdown is ingested. The checked-in fixture carries the same sentence the stub
+ * returns, so the test also holds against a real docling-serve
+ * ({@code -Dcamel.quarkus.start-mock-backend=false}).
+ */
+@QuarkusTest
+@QuarkusTestResource(value = IngestDoclingTestResource.class, restrictToAnnotatedClass = true)
+class Langchain4jIngestDoclingTest {
+
+ @Test
+ void documentIsConvertedAndIngested() throws Exception {
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body(getClass().getResourceAsStream("/valve.pdf").readAllBytes())
+ .post("/langchain4j-ingest/binary/scans/valve.pdf")
+ .then().statusCode(204);
+
+ Awaitility.await().atMost(60, TimeUnit.SECONDS).pollInterval(500, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ Map<String, String> hit = Langchain4jIngestTest.hit(
+ "At what pressure does the valve seal?", "scans", "EPSILON-2");
+ assertNotNull(hit, "the document must be converted and its markdown ingested");
+ assertEquals("scans", hit.get("pipeline"));
+ assertEquals("valve.pdf", hit.get("documentId"));
+ });
+ }
+}
diff --git a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestHeaders.java b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaIT.java
similarity index 58%
copy from extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestHeaders.java
copy to integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaIT.java
index d2834bce41..4e3bdcc787 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestHeaders.java
+++ b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaIT.java
@@ -14,20 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.camel.quarkus.component.langchain4j.ingest;
+package org.apache.camel.quarkus.component.langchain4j.ingest.it;
-/**
- * Headers an ingestion pipeline reads from the exchanges its consumer delivers.
- */
-public final class IngestHeaders {
-
- /**
- * The stable document id, when {@code source.document-id} does not name another header.
- * Identity is what future update and delete semantics will be built on, so a generated
- * fallback would silently break replacement — an absent id is an error.
- */
- public static final String DOCUMENT_ID = "CamelIngestDocumentId";
+import io.quarkus.test.junit.QuarkusIntegrationTest;
- private IngestHeaders() {
- }
+@QuarkusIntegrationTest
+class Langchain4jIngestTikaIT extends Langchain4jIngestTikaTest {
}
diff --git a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaTest.java b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaTest.java
new file mode 100644
index 0000000000..a660a90e9d
--- /dev/null
+++ b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTikaTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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.quarkus.component.langchain4j.ingest.it;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The {@code tika} parser: a PDF dropped into the watched directory is parsed to text in-process
+ * and ingested like any text file, the file name as the document id. The fixture is a checked-in
+ * one-page PDF carrying the asserted sentence, the same pattern the tika and docling test
+ * modules use.
+ */
+@QuarkusTest
+class Langchain4jIngestTikaTest {
+
+ @Test
+ void pdfIsParsedAndIngested() throws Exception {
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body(getClass().getResourceAsStream("/pump.pdf").readAllBytes())
+ .post("/langchain4j-ingest/binary/reports/pump.pdf")
+ .then().statusCode(204);
+
+ // generous on purpose: the first PDF parse on a fresh machine builds PDFBox's on-disk
+ // font cache (the fixture's non-embedded Helvetica consults the system font provider),
+ // which alone exceeds the usual timeout on Windows CI runners
+ Awaitility.await().atMost(3, TimeUnit.MINUTES).pollInterval(500, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ Map<String, String> hit = Langchain4jIngestTest.hit(
+ "What does the pump tolerate?", "reports", "DELTA-5");
+ assertNotNull(hit, "the PDF must be parsed and its text ingested");
+ assertEquals("reports", hit.get("pipeline"));
+ assertEquals("pump.pdf", hit.get("documentId"));
+ });
+ }
+
+ /**
+ * A structured format (HTML, from the parser module camel-tika itself ships): adjacent block
+ * elements must not glue into one word in the extracted text, and the {@code <head><title>}
+ * must not be ingested — Tika's text output keeps the body subtree only. The fixture keeps
+ * its two paragraphs adjacent with no whitespace between them, so any separation comes from
+ * the parse, not from the fixture's own formatting.
+ */
+ @Test
+ void htmlBlocksStaySeparatedAndTitleIsDropped() throws Exception {
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body(getClass().getResourceAsStream("/notice.html").readAllBytes())
+ .post("/langchain4j-ingest/binary/reports/notice.html")
+ .then().statusCode(204);
+
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).pollInterval(500, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> assertNotNull(
+ Langchain4jIngestTest.hit("What opens the second paragraph?", "reports", "OMEGA-3"),
+ "the HTML must be parsed and its text ingested"));
+
+ List<Map<String, String>> hits = Langchain4jIngestTest.hits("maintenance notice", "reports");
+ // both paragraphs fit one segment, so the glue check is exercised within a single text
+ assertTrue(hits.stream().anyMatch(
+ hit -> hit.get("text").contains("GAMMA-7") && hit.get("text").contains("OMEGA-3")),
+ "both paragraphs must land in one segment");
+ assertTrue(hits.stream().noneMatch(hit -> hit.get("text").contains("GAMMA-7OMEGA-3")),
+ "adjacent block elements must stay separated in the extracted text");
+ assertTrue(hits.stream().noneMatch(hit -> hit.get("text").contains("Should not be ingested")),
+ "the title must not be ingested");
+ }
+
+ /**
+ * The endpoint+parser combination, and the anti-spoofing contract with it: Tika copies a
+ * document's {@code <meta>} entries over the exchange headers, so the document id must have
+ * been captured before the parse — a document claiming another id through its own metadata
+ * ingests under the identity the consumer delivered, not the forged one.
+ */
+ @Test
+ void consumerFedParserCannotBeIdSpoofed() {
+ RestAssured.given().contentType(ContentType.TEXT)
+ .body("<html><head><meta name=\"CamelIngestDocumentId\" content=\"spoofed.txt\"/>"
+ + "<title>Should not be ingested</title></head>"
+ + "<body><p>The PSI-9 relay arrives over the feed.</p></body></html>")
+ .post("/langchain4j-ingest/feed/htmlfeed/real.html")
+ .then().statusCode(200).body(org.hamcrest.Matchers.is("ingested"));
+
+ Map<String, String> hit = Langchain4jIngestTest.hit("What arrives over the feed?", "htmlfeed", "PSI-9");
+ assertNotNull(hit, "the HTML must be parsed and its text ingested");
+ assertEquals("real.html", hit.get("documentId"),
+ "the id must be the one the consumer delivered, not the document's forgery");
+ assertTrue(Langchain4jIngestTest.hits("relay", "htmlfeed").stream()
+ .noneMatch(h -> "spoofed.txt".equals(h.get("documentId"))),
+ "nothing may be ingested under the forged id");
+ }
+
+ /**
+ * The pre-parse raw-byte guard: on the {@code capped} pipeline ({@code max-document-size=200})
+ * a small document ingests — proving the pipeline is live — while an oversized one is
+ * rejected before it reaches the parser and never lands in the store; its register key stays
+ * uncommitted, so the file is retried (and rejected again) on every poll.
+ */
+ @Test
+ void oversizedDocumentIsRejectedBeforeTheParser() throws Exception {
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body("<html><body><p>KAPPA-1 fits under the cap.</p></body></html>"
+ .getBytes(java.nio.charset.StandardCharsets.UTF_8))
+ .post("/langchain4j-ingest/binary/capped/small.html")
+ .then().statusCode(204);
+
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).pollInterval(500, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> assertNotNull(
+ Langchain4jIngestTest.hit("What fits under the cap?", "capped", "KAPPA-1"),
+ "the small document must be parsed and ingested"));
+
+ // pump.pdf is ~850 bytes, well over the 200-byte cap
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body(getClass().getResourceAsStream("/pump.pdf").readAllBytes())
+ .post("/langchain4j-ingest/binary/capped/pump.pdf")
+ .then().statusCode(204);
+
+ // the absence must hold across several poll cycles, not just at one instant
+ Awaitility.await().during(3, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS)
+ .until(() -> Langchain4jIngestTest.hits("What does the pump tolerate?", "capped").stream()
+ .noneMatch(hit -> hit.get("text").contains("DELTA-5")));
+ }
+
+ /**
+ * A delivery whose document id resolves to nothing must fail before the parse: with the
+ * exchange property unset, the producer's header fallback would read headers a parsed
+ * document may have written — the previous engine failed such a delivery too.
+ */
+ @Test
+ void consumerDeliveryWithoutIdFailsBeforeTheParse() {
+ RestAssured.given().contentType(ContentType.TEXT)
+ .body("<html><head><meta name=\"CamelIngestDocumentId\" content=\"anonymous-victim.txt\"/></head>"
+ + "<body><p>The RHO-4 module arrives without an id.</p></body></html>")
+ .post("/langchain4j-ingest/feed-anonymous/htmlfeed")
+ .then().statusCode(200).body(org.hamcrest.Matchers.is("failed"));
+
+ assertTrue(Langchain4jIngestTest.hits("module arrives", "htmlfeed").stream()
+ .noneMatch(hit -> hit.get("text").contains("RHO-4")),
+ "nothing may be ingested from an id-less delivery");
+ }
+
+ /** The same anti-spoofing contract on a directory pipeline: the file name wins. */
+ @Test
+ void directoryDocumentCannotSpoofItsId() {
+ RestAssured.given().contentType(ContentType.BINARY)
+ .body(("<html><head><meta name=\"CamelFileName\" content=\"forged.txt\"/></head>"
+ + "<body><p>The CHI-3 sensor sits in an honest file.</p></body></html>")
+ .getBytes(java.nio.charset.StandardCharsets.UTF_8))
+ .post("/langchain4j-ingest/binary/reports/honest.html")
+ .then().statusCode(204);
+
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).pollInterval(500, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ Map<String, String> hit = Langchain4jIngestTest.hit(
+ "Where does the sensor sit?", "reports", "CHI-3");
+ assertNotNull(hit, "the HTML must be parsed and its text ingested");
+ assertEquals("honest.html", hit.get("documentId"),
+ "the id must be the file name, not the document's forgery");
+ });
+ }
+}
diff --git a/integration-tests/langchain4j-ingest/src/test/resources/notice.html b/integration-tests/langchain4j-ingest/src/test/resources/notice.html
new file mode 100644
index 0000000000..3148175750
--- /dev/null
+++ b/integration-tests/langchain4j-ingest/src/test/resources/notice.html
@@ -0,0 +1,22 @@
+<!--
+
+ 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.
+
+-->
+<html>
+<head><title>Should not be ingested</title></head>
+<body><p>The maintenance notice ends its first paragraph with GAMMA-7</p><p>OMEGA-3 opens the second paragraph of the notice.</p></body>
+</html>
diff --git a/integration-tests/langchain4j-ingest/src/test/resources/pump.pdf b/integration-tests/langchain4j-ingest/src/test/resources/pump.pdf
new file mode 100644
index 0000000000..110d9e6093
Binary files /dev/null and b/integration-tests/langchain4j-ingest/src/test/resources/pump.pdf differ
diff --git a/integration-tests/langchain4j-ingest/src/test/resources/valve.pdf b/integration-tests/langchain4j-ingest/src/test/resources/valve.pdf
new file mode 100644
index 0000000000..dfc8aeefc6
Binary files /dev/null and b/integration-tests/langchain4j-ingest/src/test/resources/valve.pdf differ
diff --git a/pom.xml b/pom.xml
index 0f59b10908..d8b316f760 100644
--- a/pom.xml
+++ b/pom.xml
@@ -197,6 +197,7 @@
<microsoft.graph.auth.version>0.3.0</microsoft.graph.auth.version>
<pdfbox.version>${pdfbox-version}</pdfbox.version>
<sshd.version>2.19.0</sshd.version><!-- @sync io.quarkus:quarkus-bom:${quarkus.version} dep:org.apache.sshd:sshd-common -->
+ <tika.version>${tika-version}</tika.version>
<unboundid-ldapsdk.version>7.0.5</unboundid-ldapsdk.version>
<wiremock.version>3.13.2</wiremock.version>
<xmlunit.version>2.13.0</xmlunit.version>
diff --git a/poms/bom/pom.xml b/poms/bom/pom.xml
index 44dfb92a5c..550226523b 100644
--- a/poms/bom/pom.xml
+++ b/poms/bom/pom.xml
@@ -2191,6 +2191,11 @@
<artifactId>camel-langchain4j-embeddingstore-api</artifactId>
<version>${camel.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-langchain4j-ingest</artifactId>
+ <version>${camel.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-langchain4j-tokenizer</artifactId>
@@ -8179,6 +8184,17 @@
<artifactId>shiro-core</artifactId>
<version>${shiro-version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-parser-pdf-module</artifactId>
+ <version>${tika.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.xmlgraphics</groupId>
<artifactId>xmlgraphics-commons</artifactId>
@@ -8547,6 +8563,10 @@
<resolutionEntryPointInclude>com.ibm.icu:icu4j</resolutionEntryPointInclude>
<resolutionEntryPointInclude>com.sun.xml.fastinfoset:FastInfoset</resolutionEntryPointInclude>
<resolutionEntryPointInclude>org.mapstruct:mapstruct-processor</resolutionEntryPointInclude>
+ <!-- Tika parses only the formats whose parser modules are present; applications
+ using the langchain4j-ingest tika parser add the modules for their formats;
+ moves to camel-quarkus-tika together with the native build steps, see #9087 -->
+ <resolutionEntryPointInclude>org.apache.tika:tika-parser-pdf-module</resolutionEntryPointInclude>
</resolutionEntryPointIncludes>
<resolutionExcludes>
<!-- Exclude artifacts that we do not want in the flattened BOM -->
diff --git a/poms/bom/src/main/generated/flattened-full-pom.xml b/poms/bom/src/main/generated/flattened-full-pom.xml
index 2f0ba27106..5bdcc69c2d 100644
--- a/poms/bom/src/main/generated/flattened-full-pom.xml
+++ b/poms/bom/src/main/generated/flattened-full-pom.xml
@@ -2104,6 +2104,11 @@
<artifactId>camel-langchain4j-embeddingstore-api</artifactId>
<version>4.23.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-langchain4j-ingest</artifactId>
+ <version>4.23.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-langchain4j-tokenizer</artifactId>
@@ -8061,6 +8066,17 @@
<artifactId>shiro-core</artifactId>
<version>3.0.1</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-parser-pdf-module</artifactId>
+ <version>4.0.0</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.xmlgraphics</groupId>
<artifactId>xmlgraphics-commons</artifactId>
diff --git a/poms/bom/src/main/generated/flattened-reduced-pom.xml b/poms/bom/src/main/generated/flattened-reduced-pom.xml
index 3df1abab4c..8bb07594ad 100644
--- a/poms/bom/src/main/generated/flattened-reduced-pom.xml
+++ b/poms/bom/src/main/generated/flattened-reduced-pom.xml
@@ -3447,6 +3447,11 @@
<artifactId>camel-langchain4j-embeddingstore-api</artifactId>
<version>4.23.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-langchain4j-ingest</artifactId>
+ <version>4.23.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-langchain4j-tokenizer</artifactId>
@@ -8875,6 +8880,17 @@
<artifactId>shiro-core</artifactId>
<version>3.0.1</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-parser-pdf-module</artifactId>
+ <version>4.0.0</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.velocity</groupId>
<artifactId>velocity-engine-core</artifactId>
diff --git a/poms/bom/src/main/generated/flattened-reduced-verbose-pom.xml b/poms/bom/src/main/generated/flattened-reduced-verbose-pom.xml
index ef5c100617..f0cf827be0 100644
--- a/poms/bom/src/main/generated/flattened-reduced-verbose-pom.xml
+++ b/poms/bom/src/main/generated/flattened-reduced-verbose-pom.xml
@@ -3447,6 +3447,11 @@
<artifactId>camel-langchain4j-embeddingstore-api</artifactId>
<version>4.23.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-langchain4j-ingest</artifactId>
+ <version>4.23.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-langchain4j-tokenizer</artifactId>
@@ -8875,6 +8880,17 @@
<artifactId>shiro-core</artifactId>
<version>3.0.1</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-parser-pdf-module</artifactId>
+ <version>4.0.0</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.velocity</groupId>
<artifactId>velocity-engine-core</artifactId>