This is an automated email from the ASF dual-hosted git repository.
jamesnetherton pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
The following commit(s) were added to refs/heads/main by this push:
new c0ddf09b3f Fixes #9039. Add a configurable idempotent repository to
langchain4j-ingest
c0ddf09b3f is described below
commit c0ddf09b3fb83d18ce74f474e07aa073a5637704
Author: Jiří Ondrušek <[email protected]>
AuthorDate: Tue Sep 1 15:03:04 2026 +0200
Fixes #9039. Add a configurable idempotent repository to langchain4j-ingest
The per-pipeline option source.idempotent-repository names an
IdempotentRepository bean replacing the built-in in-memory register
(100k keys, lost on restart). The bean can be auto-created, defined via
camel.beans.* or produced by CDI (an existing bean wins); it resolves
eagerly by name, so a missing bean fails the start, and native mode
registers the camel-core repositories plus every indexed
IdempotentRepository implementation for reflection.
An edited file re-ingests under the path:modified:size key with its old
segments kept; a consumer-fed pipeline deduplicates by document id,
first write wins, answering the new outcome SKIPPED. Documented in
usage.adoc; tests cover the three definition styles, missing-bean
failures, key release on failed delivery and an H2-backed
JdbcMessageIdRepository through Quarkus Dev Services.
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../reference/extensions/langchain4j-ingest.adoc | 70 +++++++-
.../deployment/Langchain4jIngestProcessor.java | 23 +++
.../IngestAutoCreateWithoutNameTest.java | 43 +++++
.../IngestBuilderConfigIdempotentOverrideTest.java | 54 ++++++
...estMissingIdempotentRepositoryEndpointTest.java | 44 +++++
.../IngestMissingIdempotentRepositoryTest.java | 44 +++++
.../runtime/src/main/doc/usage.adoc | 52 +++++-
.../ingest/IngestComponentPresence.java | 4 +-
.../component/langchain4j/ingest/IngestRoutes.java | 126 ++++++++++++--
.../langchain4j/ingest/IngestRunTimeConfig.java | 16 ++
.../component/langchain4j/ingest/Source.java | 24 +++
.../langchain4j/ingest/core/IngestResult.java | 4 +-
integration-tests/langchain4j-ingest/pom.xml | 21 +++
.../ingest/it/DeterministicEmbeddingModel.java | 6 +
.../langchain4j/ingest/it/IngestItProducers.java | 32 +++-
.../langchain4j/ingest/it/IngestResource.java | 26 ++-
.../langchain4j/ingest/it/ItBuilderPipelines.java | 7 +-
.../src/main/resources/application.properties | 25 ++-
.../ingest/it/Langchain4jIngestIdempotentIT.java | 22 +--
.../ingest/it/Langchain4jIngestIdempotentTest.java | 181 +++++++++++++++++++++
.../ingest/it/Langchain4jIngestTest.java | 4 +-
21 files changed, 776 insertions(+), 52 deletions(-)
diff --git
a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
index 614bdc88fa..151e7a8c3a 100644
--- a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
+++ b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
@@ -55,9 +55,60 @@ Each file is read as UTF-8 text — there is no format
parsing, so convert a PDF
[NOTE]
====
-This experimental extension ingests what it is given and keeps no record of
it: a document ingested twice leaves two copies in the store, a restart
re-reads the whole directory (the duplicate register is in-memory, sized for
100,000 files), and a polled consumer re-reads its source on every poll.
Keeping the store in step with a changing source — skipping unchanged
documents, replacing edited ones, removing deleted ones — needs that record and
arrives with the synchronising engine in a l [...]
+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 [...]
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=productsRegister
+----
+
+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:
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=productsRegister
+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:
+
+[source,properties]
+----
+camel.beans.productsRegister=#class:org.apache.camel.support.processor.idempotent.FileIdempotentRepository
+camel.beans.productsRegister.fileStore=/var/data/ingest/register.dat
+camel.beans.productsRegister.cacheSize=100000
+# keys beyond this size are dropped oldest-first and their files re-ingest (32
MB here)
+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.
+
+*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]
+----
+@Produces @Singleton @Named("productsRegister")
+IdempotentRepository productsRegister(DataSource dataSource) {
+ return new JdbcMessageIdRepository(dataSource, "ingest-products"); //
camel-quarkus-sql
+}
+----
+
+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 [...]
+
+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 [...]
+
[id="extensions-langchain4j-ingest-usage-other-sources-declared-in-java"]
=== Other sources, declared in Java
@@ -190,6 +241,23 @@ Whether subdirectories are ingested too, when reading a
directory.
| `boolean`
| `true`
+a|
[[quarkus-camel-langchain4j-ingest-pipeline-name-source-idempotent-repository]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-source-idempotent-repository[quarkus.camel.langchain4j.ingest."pipeline-name".source.idempotent-repository]`
+
+Name of the `IdempotentRepository` bean remembering already ingested documents,
+instead of the built-in in-memory one (100 000 keys, lost on restart). Looked
up
+by name only. On a pipeline consuming from a component it deduplicates
deliveries
+by document id, first write wins.
+| `string`
+|
+
+a|
[[quarkus-camel-langchain4j-ingest-pipeline-name-source-idempotent-repository-auto-create]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-source-idempotent-repository-auto-create[quarkus.camel.langchain4j.ingest."pipeline-name".source.idempotent-repository-auto-create]`
+
+When `true`, an in-memory register (100 000 keys) is created and bound under
the
+`idempotent-repository` name, unless a bean with that name exists — the
existing
+bean wins.
+| `boolean`
+| `false`
+
a|
[[quarkus-camel-langchain4j-ingest-pipeline-name-source-document-id]]`link:#quarkus-camel-langchain4j-ingest-pipeline-name-source-document-id[quarkus.camel.langchain4j.ingest."pipeline-name".source.document-id]`
Where the document id lives in the exchange the consumer delivers: normally the
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 2e5cfc0f77..3f1c1148ba 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
@@ -73,6 +73,29 @@ class Langchain4jIngestProcessor {
.build();
}
+ /**
+ * {@code #class:} beans are instantiated reflectively, which native mode
allows only for
+ * registered classes: the two camel-core repositories the documentation
recommends, plus
+ * every {@code IdempotentRepository} implementation the Jandex index
knows — application
+ * classes always, third-party ones when their jar carries an index.
+ */
+ @BuildStep
+ void repositoryReflection(CombinedIndexBuildItem combinedIndex,
+ BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+ Set<String> repositories = new HashSet<>(Set.of(
+
"org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository",
+
"org.apache.camel.support.processor.idempotent.FileIdempotentRepository"));
+ combinedIndex.getIndex()
+
.getAllKnownImplementations(DotName.createSimple("org.apache.camel.spi.IdempotentRepository"))
+ .stream()
+ .filter(repository -> !Modifier.isAbstract(repository.flags()))
+ .map(repository -> repository.name().toString())
+ .forEach(repositories::add);
+
reflectiveClasses.produce(ReflectiveClassBuildItem.builder(repositories.toArray(new
String[0]))
+ .methods()
+ .build());
+ }
+
/**
* Discovers {@code @Ingest} builder methods: validated here (return type,
no parameters,
* unique names, no collision with configuration-declared pipelines),
invoked reflectively once
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestAutoCreateWithoutNameTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestAutoCreateWithoutNameTest.java
new file mode 100644
index 0000000000..750527574c
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestAutoCreateWithoutNameTest.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;
+
+/** Auto-create without a repository name to create it under fails the start.
*/
+class IngestAutoCreateWithoutNameTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar ->
jar.addClasses(TestEmbeddingBeans.class))
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-store",
"store")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-model",
"model")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.directory",
"target/auto-docs")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.idempotent-repository-auto-create",
+ "true")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "sets source.idempotent-repository-auto-create",
+ "no source.idempotent-repository name"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderConfigIdempotentOverrideTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderConfigIdempotentOverrideTest.java
new file mode 100644
index 0000000000..81245c7894
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestBuilderConfigIdempotentOverrideTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+/**
+ * {@code source.idempotent-repository} is owned by a builder-declared
pipeline; configuring it
+ * externally fails at startup like the rest of {@code source.*}.
+ */
+class IngestBuilderConfigIdempotentOverrideTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar -> jar.addClasses(Pipelines.class))
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.idempotent-repository",
"external-repo")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "declared in Java", "Remove
quarkus.camel.langchain4j.ingest.docs.source.*"));
+
+ @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/docs"));
+ }
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingIdempotentRepositoryEndpointTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingIdempotentRepositoryEndpointTest.java
new file mode 100644
index 0000000000..1d31c131e4
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingIdempotentRepositoryEndpointTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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 missing-repository failure on the consumer-fed path: resolution happens
while the route is
+ * built, so both source kinds fail the start with the same message naming the
bean.
+ */
+class IngestMissingIdempotentRepositoryEndpointTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar ->
jar.addClasses(TestEmbeddingBeans.class))
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-store",
"store")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-model",
"model")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.uri",
"file:target/endpoint-repo-missing")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.idempotent-repository",
"no-such-repo")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "references idempotent repository 'no-such-repo'", "no
such bean exists"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+}
diff --git
a/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingIdempotentRepositoryTest.java
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingIdempotentRepositoryTest.java
new file mode 100644
index 0000000000..00711ce5eb
--- /dev/null
+++
b/extensions/langchain4j-ingest/deployment/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/deployment/IngestMissingIdempotentRepositoryTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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 idempotent repository is resolved by name when the route starts — after
Camel Main has
+ * bound {@code camel.beans.*} definitions — and a missing bean fails the
start naming it.
+ */
+class IngestMissingIdempotentRepositoryTest {
+
+ @RegisterExtension
+ static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest()
+ .withApplicationRoot(jar ->
jar.addClasses(TestEmbeddingBeans.class))
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-store",
"store")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.embedding-model",
"model")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.directory",
"target/idem-docs")
+
.overrideConfigKey("quarkus.camel.langchain4j.ingest.docs.source.idempotent-repository",
"no-such-repo")
+ .assertException(t -> ValidationTestSupport.assertFailure(t,
+ "references idempotent repository 'no-such-repo'", "no
such bean exists"));
+
+ @Test
+ void startMustFail() {
+ Assertions.fail("The application start was expected to fail");
+ }
+}
diff --git a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
index 243fc58286..7af9811e88 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
+++ b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
@@ -13,9 +13,59 @@ Each file is read as UTF-8 text — there is no format
parsing, so convert a PDF
[NOTE]
====
-This experimental extension ingests what it is given and keeps no record of
it: a document ingested twice leaves two copies in the store, a restart
re-reads the whole directory (the duplicate register is in-memory, sized for
100,000 files), and a polled consumer re-reads its source on every poll.
Keeping the store in step with a changing source — skipping unchanged
documents, replacing edited ones, removing deleted ones — needs that record and
arrives with the synchronising engine in a l [...]
+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 [...]
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=productsRegister
+----
+
+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:
+
+[source,properties]
+----
+quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=productsRegister
+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:
+
+[source,properties]
+----
+camel.beans.productsRegister=#class:org.apache.camel.support.processor.idempotent.FileIdempotentRepository
+camel.beans.productsRegister.fileStore=/var/data/ingest/register.dat
+camel.beans.productsRegister.cacheSize=100000
+# keys beyond this size are dropped oldest-first and their files re-ingest (32
MB here)
+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.
+
+*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]
+----
+@Produces @Singleton @Named("productsRegister")
+IdempotentRepository productsRegister(DataSource dataSource) {
+ return new JdbcMessageIdRepository(dataSource, "ingest-products"); //
camel-quarkus-sql
+}
+----
+
+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 [...]
+
+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 [...]
+
=== 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:
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 1e7eb3c220..7fb10e8ae9 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
@@ -61,7 +61,9 @@ final class IngestComponentPresence {
continue;
}
if (external != null && (external.source().directory().isPresent()
- || external.source().documentId().isPresent())) {
+ || external.source().documentId().isPresent()
+ || external.source().idempotentRepository().isPresent()
+ || external.source().idempotentRepositoryAutoCreate())) {
// the route builder refuses this conflict with its own error;
invoking the
// method here first would change which failure the user sees
continue;
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 498a1ad499..1461e16f52 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
@@ -31,11 +31,13 @@ import jakarta.enterprise.inject.Any;
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;
@@ -71,6 +73,9 @@ 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() {
// a pipeline may be declared entirely through runtime properties -
the documented
@@ -136,7 +141,9 @@ public class IngestRoutes extends RouteBuilder {
// (source.recursive cannot be told apart from its default, so it
alone goes undetected -
// Source.recursive() is its builder twin)
if (external != null && (external.source().directory().isPresent()
- || external.source().documentId().isPresent())) {
+ || external.source().documentId().isPresent()
+ || external.source().idempotentRepository().isPresent()
+ || external.source().idempotentRepositoryAutoCreate())) {
throw new IllegalStateException("Ingestion pipeline '" + name + "'
is declared in Java, so its source "
+ "comes from the @Ingest method. Remove
quarkus.camel.langchain4j.ingest." + name + ".source.* , or "
+ "declare the pipeline in configuration instead.");
@@ -171,15 +178,24 @@ public class IngestRoutes extends RouteBuilder {
// 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 - the truncation would be permanent, since idempotent keys on
the path. The register
- // is sized explicitly: Camel's default caps at 1000 entries, and
beyond that eviction
- // would re-ingest a large directory steadily during normal operation,
not just on restart
+ // changed read lock waits for a file still being copied in rather
than embedding half
+ // of it
Expression documentId = documentIdExpression(runtime,
Exchange.FILE_NAME);
- from(file(directory)
+ maybeAutoCreateRepository(name, runtime);
+ String repositoryName = runtime == null ? null :
runtime.source().idempotentRepository().orElse(null);
+ var endpoint = file(directory)
.noop(true)
- .idempotent(true)
-
.idempotentRepository(MemoryIdempotentRepository.memoryIdempotentRepository(100_000))
+ .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()))
@@ -204,20 +220,58 @@ public class IngestRoutes extends RouteBuilder {
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;
+ }
+ // 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 -> {
- 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.");
+ 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));
}
- exchange.getIn().setBody(service.ingest(id,
exchange.getIn().getBody(String.class)));
});
}
+ 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.");
+ }
+ return id;
+ }
+
private Expression
documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime,
String defaultHeader) {
String configured = runtime == null ? null :
runtime.source().documentId().orElse(null);
@@ -265,6 +319,48 @@ 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/IngestRunTimeConfig.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java
index 17be745851..ce84df44c2 100644
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java
+++
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestRunTimeConfig.java
@@ -69,6 +69,22 @@ public interface IngestRunTimeConfig {
@WithDefault("true")
boolean recursive();
+ /**
+ * Name of the `IdempotentRepository` bean remembering already
ingested documents,
+ * instead of the built-in in-memory one (100 000 keys, lost on
restart). Looked up
+ * by name only. On a pipeline consuming from a component it
deduplicates deliveries
+ * by document id, first write wins.
+ */
+ Optional<String> idempotentRepository();
+
+ /**
+ * When `true`, an in-memory register (100 000 keys) is created
and bound under the
+ * `idempotent-repository` name, unless a bean with that name
exists — the existing
+ * bean wins.
+ */
+ @WithDefault("false")
+ boolean idempotentRepositoryAutoCreate();
+
/**
* Where the document id lives in the exchange the consumer
delivers: normally the
* name of a header, such as `CamelAwsS3Key` for an S3 consumer or
`CamelKafkaKey` for
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/Source.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/Source.java
index dfac49b82b..72d757758a 100644
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/Source.java
+++
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/Source.java
@@ -49,6 +49,8 @@ public final class Source {
private String directory;
private String uri;
private String documentId;
+ private String idempotentRepository;
+ private boolean idempotentRepositoryAutoCreate;
private boolean recursive = true;
private Source(String type) {
@@ -109,6 +111,18 @@ public final class Source {
return this;
}
+ /** Name of the {@code IdempotentRepository} bean; twin of {@code
source.idempotent-repository}. */
+ public Source idempotentRepository(String beanName) {
+ this.idempotentRepository = requireText(beanName,
"idempotentRepository");
+ return this;
+ }
+
+ /** Twin of {@code source.idempotent-repository-auto-create}. */
+ public Source idempotentRepositoryAutoCreate(boolean autoCreate) {
+ this.idempotentRepositoryAutoCreate = autoCreate;
+ return this;
+ }
+
/** A null or blank value here would surface much later as an obscure
Camel error. */
private static String requireText(String value, String what) {
if (value == null || value.isBlank()) {
@@ -143,6 +157,16 @@ public final class Source {
public Optional<String> documentId() {
return Optional.ofNullable(documentId);
}
+
+ @Override
+ public Optional<String> idempotentRepository() {
+ return Optional.ofNullable(idempotentRepository);
+ }
+
+ @Override
+ public boolean idempotentRepositoryAutoCreate() {
+ return idempotentRepositoryAutoCreate;
+ }
};
}
}
diff --git
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java
index 13ebbcfc07..45e4d86799 100644
---
a/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java
+++
b/extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java
@@ -27,7 +27,9 @@ public record IngestResult(String pipeline, String
documentId, int segmentsWritt
/** Segments written. */
INGESTED,
/** Blank document, nothing written. */
- EMPTY;
+ EMPTY,
+ /** Already ingested under the same key, nothing written. */
+ SKIPPED;
/** The stable wire/log form. */
public String label() {
diff --git a/integration-tests/langchain4j-ingest/pom.xml
b/integration-tests/langchain4j-ingest/pom.xml
index 5cd7bb4ef5..125b77664e 100644
--- a/integration-tests/langchain4j-ingest/pom.xml
+++ b/integration-tests/langchain4j-ingest/pom.xml
@@ -59,6 +59,14 @@
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-kafka</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-sql</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-jdbc-h2</artifactId>
+ </dependency>
<!-- test dependencies -->
<dependency>
@@ -203,6 +211,19 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel.quarkus</groupId>
+ <artifactId>camel-quarkus-sql-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/DeterministicEmbeddingModel.java
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/DeterministicEmbeddingModel.java
index bee18f2e12..0625978775 100644
---
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/DeterministicEmbeddingModel.java
+++
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/DeterministicEmbeddingModel.java
@@ -33,6 +33,9 @@ import dev.langchain4j.model.output.Response;
*/
public class DeterministicEmbeddingModel implements EmbeddingModel {
+ /** Test hook: a segment carrying this marker fails the embedding, so the
exchange fails. */
+ public static final String POISON = "POISON-PILL";
+
private final int dimension;
public DeterministicEmbeddingModel(int dimension) {
@@ -43,6 +46,9 @@ public class DeterministicEmbeddingModel implements
EmbeddingModel {
public Response<List<Embedding>> embedAll(List<TextSegment> segments) {
List<Embedding> embeddings = new ArrayList<>(segments.size());
for (TextSegment segment : segments) {
+ if (segment.text().contains(POISON)) {
+ throw new IllegalStateException("test-induced embedding
failure: " + POISON);
+ }
embeddings.add(embeddingFor(segment.text()));
}
return Response.from(embeddings);
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 6ba00152ed..c61430afbc 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
@@ -16,6 +16,8 @@
*/
package org.apache.camel.quarkus.component.langchain4j.ingest.it;
+import javax.sql.DataSource;
+
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.store.embedding.EmbeddingStore;
@@ -24,6 +26,9 @@ import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Produces;
import jakarta.inject.Named;
import jakarta.inject.Singleton;
+import org.apache.camel.processor.idempotent.jdbc.JdbcMessageIdRepository;
+import org.apache.camel.spi.IdempotentRepository;
+import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
@ApplicationScoped
public class IngestItProducers {
@@ -45,8 +50,8 @@ public class IngestItProducers {
@Produces
@Singleton
- @Named("built-store")
- EmbeddingStore<TextSegment> builtStore() {
+ @Named("datasheets-store")
+ EmbeddingStore<TextSegment> datasheetsStore() {
return new InMemoryEmbeddingStore<>();
}
@@ -64,10 +69,33 @@ public class IngestItProducers {
return new InMemoryEmbeddingStore<>();
}
+ @Produces
+ @Singleton
+ @Named("jdbc-store")
+ EmbeddingStore<TextSegment> jdbcStore() {
+ return new InMemoryEmbeddingStore<>();
+ }
+
@Produces
@Singleton
@Named("test-model")
EmbeddingModel embeddingModel() {
return new DeterministicEmbeddingModel(64);
}
+
+ // the custom pipeline's register; auto-create is also set, so this
existing bean must win
+ @Produces
+ @Singleton
+ @Named("test-register")
+ IdempotentRepository testRegister() {
+ return MemoryIdempotentRepository.memoryIdempotentRepository(1000);
+ }
+
+ // the JDBC register from the documentation's CDI example, over the Dev
Services H2 datasource
+ @Produces
+ @Singleton
+ @Named("jdbcRegister")
+ IdempotentRepository jdbcRegister(DataSource dataSource) {
+ return new JdbcMessageIdRepository(dataSource, "ingest-jdbc");
+ }
}
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 f6f7e4dee7..3d8c3b8fad 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
@@ -33,10 +33,12 @@ import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
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.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;
@jakarta.ws.rs.Path("/langchain4j-ingest")
@@ -55,8 +57,8 @@ public class IngestResource {
EmbeddingStore<TextSegment> customStore;
@Inject
- @Named("built-store")
- EmbeddingStore<TextSegment> builtStore;
+ @Named("datasheets-store")
+ EmbeddingStore<TextSegment> datasheetsStore;
@Inject
@Named("s3-store")
@@ -69,9 +71,22 @@ public class IngestResource {
@Inject
ProducerTemplate producerTemplate;
+ @Inject
+ CamelContext camelContext;
+
@ConfigProperty(name = "ingest.test.directory")
String directory;
+ /** Asserts a key was committed; registry lookup by name, the same way the
pipelines resolve. */
+ @GET
+ @jakarta.ws.rs.Path("/register-contains")
+ @Produces(MediaType.TEXT_PLAIN)
+ public boolean registerContains(@QueryParam("repo") String repo,
@QueryParam("key") String key) {
+ IdempotentRepository repository =
camelContext.getRegistry().lookupByNameAndType(repo,
+ IdempotentRepository.class);
+ return repository != null && repository.contains(key);
+ }
+
/** Writes a document into the watched directory — app-side, so native
mode shares the path. */
@POST
@jakarta.ws.rs.Path("/file/{name}")
@@ -88,7 +103,7 @@ public class IngestResource {
public List<SearchHit> search(@QueryParam("q") String query,
@QueryParam("store") String storeName) {
EmbeddingStore<TextSegment> store = switch (storeName == null ?
"products" : storeName) {
case "custom" -> customStore;
- case "built" -> builtStore;
+ case "datasheets" -> datasheetsStore;
case "s3" -> s3Store;
case "events" -> eventsStore;
default -> productsStore;
@@ -113,14 +128,15 @@ public class IngestResource {
public record SearchHit(String text, String pipeline, String documentId) {
}
- /** Feeds a push pipeline through its Camel consumer URI. */
+ /** Feeds a pipeline synchronously; the reply carries the outcome, so
tests can assert skipped and failures. */
@POST
@jakarta.ws.rs.Path("/feed/{pipeline}/{documentId:.+}")
@Consumes(MediaType.TEXT_PLAIN)
@Produces(MediaType.TEXT_PLAIN)
public String feed(@PathParam("pipeline") String pipeline,
@PathParam("documentId") String documentId,
String content) {
- String uri = "built".equals(pipeline) ? "direct:built-source" :
"direct:custom-source";
+ // every consumer-fed test pipeline reads direct:<pipeline>-feed
+ String uri = "direct:" + pipeline + "-feed";
IngestResult result = producerTemplate.requestBodyAndHeader(uri,
content, IngestHeaders.DOCUMENT_ID,
documentId, IngestResult.class);
return result.outcome().label();
diff --git
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/ItBuilderPipelines.java
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/ItBuilderPipelines.java
index 0124e61b6c..aa29ee04d6 100644
---
a/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/ItBuilderPipelines.java
+++
b/integration-tests/langchain4j-ingest/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/ItBuilderPipelines.java
@@ -48,8 +48,11 @@ public class ItBuilderPipelines {
@Ingest("datasheets")
IngestPipeline datasheets() {
- return IngestPipeline.from(Source.endpoint(direct("built-source")))
- .embeddingStore("built-store")
+ // the auto-created register: named, but no bean is defined anywhere
for it
+ return IngestPipeline.from(Source.endpoint(direct("datasheets-feed"))
+ .idempotentRepository("datasheetsRegister")
+ .idempotentRepositoryAutoCreate(true))
+ .embeddingStore("datasheets-store")
.embeddingModel("test-model")
.splitter(120, 20);
}
diff --git
a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
index e01734c18b..f8caf3a942 100644
---
a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
+++
b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
@@ -26,18 +26,35 @@
quarkus.camel.langchain4j.ingest.products.embedding-store=products-store
quarkus.camel.langchain4j.ingest.products.embedding-model=test-model
quarkus.camel.langchain4j.ingest.products.max-segment-size=120
quarkus.camel.langchain4j.ingest.products.max-overlap-size=20
+# register defined purely in properties via camel.beans; the name has no
dashes because Camel
+# normalises dashed camel.beans keys to camelCase
+camel.beans.propertiesRegister=\#class\:org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository
+camel.beans.propertiesRegister.cacheSize=5000
+quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=propertiesRegister
-# endpoint source: the escape hatch — any Camel consumer URI feeds the
pipeline. The document-id
-# exercises the $simple{...} form, the one MicroProfile Config's own ${...}
expansion leaves
-# untouched; it reads the same header the pipeline would fall back to anyway
-quarkus.camel.langchain4j.ingest.custom.source.uri=direct:custom-source
+# 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
quarkus.camel.langchain4j.ingest.custom.source.document-id=$simple{header.CamelIngestDocumentId}
+quarkus.camel.langchain4j.ingest.custom.source.idempotent-repository=test-register
+# auto-create plus the CDI-produced test-register bean: the existing bean wins
over auto-creation
+quarkus.camel.langchain4j.ingest.custom.source.idempotent-repository-auto-create=true
quarkus.camel.langchain4j.ingest.custom.embedding-store=custom-store
quarkus.camel.langchain4j.ingest.custom.embedding-model=test-model
quarkus.camel.langchain4j.ingest.custom.max-segment-size=120
quarkus.camel.langchain4j.ingest.custom.max-overlap-size=20
+# H2 through Dev Services (a separate process, no container) backs the JDBC
register
+quarkus.datasource.db-kind=h2
+
+quarkus.camel.langchain4j.ingest.jdbcdocs.source.uri=direct:jdbcdocs-feed
+quarkus.camel.langchain4j.ingest.jdbcdocs.source.idempotent-repository=jdbcRegister
+quarkus.camel.langchain4j.ingest.jdbcdocs.embedding-store=jdbc-store
+quarkus.camel.langchain4j.ingest.jdbcdocs.embedding-model=test-model
+quarkus.camel.langchain4j.ingest.jdbcdocs.max-segment-size=120
+quarkus.camel.langchain4j.ingest.jdbcdocs.max-overlap-size=20
+
# s3docs and events are declared in Java (ItBuilderPipelines) through the
Endpoint DSL; they stay
# off until MinioTestResource / IngestKafkaTestResource provide a store and a
broker
quarkus.camel.langchain4j.ingest.s3docs.enabled=false
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/Langchain4jIngestIdempotentIT.java
similarity index 60%
copy from
extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/core/IngestResult.java
copy to
integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestIdempotentIT.java
index 13ebbcfc07..94c9a7041c 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/Langchain4jIngestIdempotentIT.java
@@ -14,24 +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;
-
- /** The stable wire/log form. */
- public String label() {
- return name().toLowerCase(Locale.ROOT);
- }
- }
+@QuarkusIntegrationTest
+class Langchain4jIngestIdempotentIT extends Langchain4jIngestIdempotentTest {
}
diff --git
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestIdempotentTest.java
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestIdempotentTest.java
new file mode 100644
index 0000000000..74a98d5c20
--- /dev/null
+++
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestIdempotentTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.concurrent.TimeUnit;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import org.awaitility.Awaitility;
+import org.hamcrest.Matchers;
+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 idempotent repository: an edited file re-ingests under a new key (path,
mtime, size) with the
+ * previous segments kept, and a consumer-fed pipeline deduplicates by
document id, first write
+ * wins.
+ */
+@QuarkusTest
+class Langchain4jIngestIdempotentTest {
+
+ @Test
+ void editedFileReIngestsAndAppends() {
+ Langchain4jIngestTest.write("versioned.txt", "The RHO-1 relay ships as
rev ALPHA.");
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).pollInterval(500,
TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> assertNotNull(
+ Langchain4jIngestTest.hit("Which rev ships?", null,
"rev ALPHA"),
+ "the first version must be ingested"));
+
+ // different length guarantees a new key even within the mtime
granularity
+ Langchain4jIngestTest.write("versioned.txt", "The RHO-1 relay now
ships as rev BRAVO, improved.");
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).pollInterval(500,
TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> assertNotNull(
+ Langchain4jIngestTest.hit("Which rev ships?", null,
"rev BRAVO"),
+ "the edited version must be re-ingested under its new
key"));
+
+ // append mode: the first version's segments were not replaced
+ assertNotNull(Langchain4jIngestTest.hit("Which rev ships?", null, "rev
ALPHA"),
+ "append mode keeps the previous version's segments");
+ }
+
+ @Test
+ void duplicateDocumentIdIsSkipped() {
+ String first = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The KAPPA-4 sensor reads humidity.")
+ .post("/langchain4j-ingest/feed/custom/dedup/kappa.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("ingested", first);
+
+ // same document id again: the register skips it, first write wins
+ String second = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The KAPPA-4 sensor allegedly reads pressure now.")
+ .post("/langchain4j-ingest/feed/custom/dedup/kappa.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("skipped", second);
+
+ // the store holds the first write only, and the register committed
the key
+ assertNotNull(Langchain4jIngestTest.hit("What does the sensor read?",
"custom", "humidity"));
+ assertTrue(Langchain4jIngestTest.hits("What does the sensor read?",
"custom").stream()
+ .noneMatch(hit -> hit.get("text").contains("pressure")),
+ "the duplicate delivery must not have been ingested");
+ RestAssured.given()
+ .queryParam("repo", "test-register")
+ .queryParam("key", "dedup/kappa.txt")
+ .get("/langchain4j-ingest/register-contains")
+ .then()
+ .statusCode(200)
+ .body(Matchers.is("true"));
+ }
+
+ /** A blank document releases its claim: only a delivery that wrote
segments occupies the id. */
+ @Test
+ void emptyDeliveryDoesNotClaimTheId() {
+ String first = RestAssured.given().contentType(ContentType.TEXT)
+ .body(" ")
+ .post("/langchain4j-ingest/feed/custom/dedup/lambda.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("empty", first);
+
+ // the claim was released, so the id is free again
+ RestAssured.given()
+ .queryParam("repo", "test-register")
+ .queryParam("key", "dedup/lambda.txt")
+ .get("/langchain4j-ingest/register-contains")
+ .then()
+ .statusCode(200)
+ .body(Matchers.is("false"));
+
+ RestAssured.given().contentType(ContentType.TEXT)
+ .body("The LAMBDA-2 valve regulates coolant flow.")
+ .post("/langchain4j-ingest/feed/custom/dedup/lambda.txt")
+ .then().statusCode(200)
+ .body(Matchers.is("ingested"));
+
+ assertNotNull(Langchain4jIngestTest.hit("What does the valve
regulate?", "custom", "coolant"),
+ "the populated delivery must be ingested despite the earlier
empty one");
+ }
+
+ /** A failed delivery releases the key ({@code removeOnFailure}), so the
same id can retry. */
+ @Test
+ void failedDeliveryReleasesTheKey() {
+ RestAssured.given().contentType(ContentType.TEXT)
+ .body("The SIGMA-9 valve " +
DeterministicEmbeddingModel.POISON + " fails to embed.")
+ .post("/langchain4j-ingest/feed/custom/release/sigma9.txt")
+ .then().statusCode(500);
+
+ String retry = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The SIGMA-9 valve seals reliably.")
+ .post("/langchain4j-ingest/feed/custom/release/sigma9.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("ingested", retry);
+ }
+
+ /** A JDBC register (H2-backed {@code JdbcMessageIdRepository}): dedup
through real SQL. */
+ @Test
+ void jdbcRegisterDeduplicates() {
+ String first = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The THETA-2 pump moves coolant.")
+ .post("/langchain4j-ingest/feed/jdbcdocs/dedup/theta.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("ingested", first);
+
+ String second = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The THETA-2 pump allegedly moves lava now.")
+ .post("/langchain4j-ingest/feed/jdbcdocs/dedup/theta.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("skipped", second);
+
+ // contains() issues a SELECT against the H2 store, proving the key
survived in SQL
+ String contains = RestAssured.given()
+ .queryParam("repo", "jdbcRegister")
+ .queryParam("key", "dedup/theta.txt")
+ .get("/langchain4j-ingest/register-contains")
+ .then().statusCode(200).extract().asString();
+ assertEquals("true", contains);
+ }
+
+ /**
+ * The {@code datasheets} builder pipeline names {@code
datasheetsRegister} with auto-create and no
+ * bean defined anywhere: the auto-created register deduplicates too.
+ */
+ @Test
+ void autoCreatedRegisterDeduplicates() {
+ String first = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The OMICRON-8 filter removes particles.")
+ .post("/langchain4j-ingest/feed/datasheets/dedup/omicron.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("ingested", first);
+
+ String second = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The OMICRON-8 filter allegedly removes odors now.")
+ .post("/langchain4j-ingest/feed/datasheets/dedup/omicron.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("skipped", second);
+
+ String contains = RestAssured.given()
+ .queryParam("repo", "datasheetsRegister")
+ .queryParam("key", "dedup/omicron.txt")
+ .get("/langchain4j-ingest/register-contains")
+ .then().statusCode(200).extract().asString();
+ assertEquals("true", contains);
+ }
+}
diff --git
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTest.java
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTest.java
index da1b0a6d86..05d1ecbe4f 100644
---
a/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTest.java
+++
b/integration-tests/langchain4j-ingest/src/test/java/org/apache/camel/quarkus/component/langchain4j/ingest/it/Langchain4jIngestTest.java
@@ -86,10 +86,10 @@ class Langchain4jIngestTest {
void builderDeclaredPipelineIngests() {
RestAssured.given().contentType(ContentType.TEXT)
.body("The builder-declared pipeline handles the SIGMA-3
datasheet.")
- .post("/langchain4j-ingest/feed/built/datasheets/sigma.txt")
+
.post("/langchain4j-ingest/feed/datasheets/datasheets/sigma.txt")
.then().statusCode(200).body(org.hamcrest.Matchers.is("ingested"));
- Map<String, String> hit = hit("Which datasheet is handled?", "built",
"SIGMA-3");
+ Map<String, String> hit = hit("Which datasheet is handled?",
"datasheets", "SIGMA-3");
assertNotNull(hit, "the fed document must be ingested");
// the metadata names the pipeline (@Ingest("datasheets")), not the
store it writes to
assertEquals("datasheets", hit.get("pipeline"));
