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 4367b04b9e Fixes #9162. Read the document id from the
CamelLangChain4jIngestDocumentId header
4367b04b9e is described below
commit 4367b04b9eff6ded181c4e1d7cdd2b5075fb0def
Author: Jiří Ondrušek <[email protected]>
AuthorDate: Tue Sep 15 08:34:09 2026 +0200
Fixes #9162. Read the document id from the CamelLangChain4jIngestDocumentId
header
The default id header of consumer-fed pipelines becomes the
camel-langchain4j-ingest component's CamelLangChain4jIngestDocumentId,
so the identity contract taught in 3.40 LTS survives the engine
delegation (#9142). The 3.39 name CamelIngestDocumentId is still read
as a deprecated fallback when the new header is absent - the new one
wins when both are set - and IngestHeaders keeps it as
LEGACY_DOCUMENT_ID. Documented in the migration guide.
Verified: 23 deployment test suites, 13 JVM integration tests green.
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../modules/ROOT/pages/migration-guide/3.40.0.adoc | 6 ++++
.../reference/extensions/langchain4j-ingest.adoc | 5 +--
.../runtime/src/main/doc/usage.adoc | 2 +-
.../langchain4j/ingest/IngestHeaders.java | 14 ++++++++-
.../component/langchain4j/ingest/IngestRoutes.java | 36 ++++++++++++++++++++--
.../langchain4j/ingest/IngestRunTimeConfig.java | 3 +-
.../langchain4j/ingest/it/IngestResource.java | 33 ++++++++++++++++++++
.../src/main/resources/application.properties | 2 +-
.../ingest/it/Langchain4jIngestIdempotentTest.java | 33 ++++++++++++++++++++
9 files changed, 125 insertions(+), 9 deletions(-)
diff --git a/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc
b/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc
index e62955d7a0..27e367f7ba 100644
--- a/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc
+++ b/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc
@@ -49,6 +49,12 @@ Nothing else about the metadata changed: the values and the
segment text are wri
Newly ingested segments carry the new keys, while segments already in a store
keep the old ones — retrieval that filters or cites by these keys sees the
mismatch.
Either re-ingest the affected documents (with the default in-memory register a
restart re-ingests the directory; clear a persistent register first) so
everything carries the new keys, or make retrieval read both key pairs during
the transition.
+=== The default document id header is now CamelLangChain4jIngestDocumentId
+
+A consumer-fed pipeline without `source.document-id` now reads its id from the
`CamelLangChain4jIngestDocumentId` header — the `camel-langchain4j-ingest`
component's convention.
+The 3.39 name `CamelIngestDocumentId` is still read as a deprecated fallback
(the new header wins when both are set), so existing routes keep working; it
will be removed in a future release.
+`IngestHeaders.DOCUMENT_ID` carries the new name, the old one stays as the
deprecated `LEGACY_DOCUMENT_ID`.
+
== LangChain4j EmbeddingStore extension changes
=== The RAG augmentor bridge moved to the EmbeddingStore extension
diff --git
a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
index d481cc2adf..f76dc2865b 100644
--- a/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
+++ b/docs/modules/ROOT/pages/reference/extensions/langchain4j-ingest.adoc
@@ -126,7 +126,7 @@ 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.
-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 `CamelLangChain4jIngestDocumentId` header — the
`camel-langchain4j-ingest` component's convention; the 3.39 name
`CamelIngestDocumentId` is still read as a fallback, see the migration guide —
and fails the exchange when both are abs [...]
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.
@@ -267,7 +267,8 @@ expression in the `+$simple{...}+` form — MicroProfile
Config passes it throug
untouched, while a `+${...}+` in a properties file would be consumed as a
config
expansion before Camel ever saw it. When not set, a pipeline reading a
directory
uses the file name, and one consuming from a component uses the
-`CamelIngestDocumentId` header.
+`CamelLangChain4jIngestDocumentId` header (the deprecated 3.39 name
+`CamelIngestDocumentId` is still read as a fallback).
| `string`
|
|===
diff --git a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
index 62663db5e1..3eb73506ca 100644
--- a/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
+++ b/extensions/langchain4j-ingest/runtime/src/main/doc/usage.adoc
@@ -82,7 +82,7 @@ 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.
-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 `CamelLangChain4jIngestDocumentId` header — the
`camel-langchain4j-ingest` component's convention; the 3.39 name
`CamelIngestDocumentId` is still read as a fallback, see the migration guide —
and fails the exchange when both are abs [...]
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.
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..1ad316d6fc 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,8 +25,20 @@ 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>
+ * The {@code camel-langchain4j-ingest} component's name; the 3.39 name is
still read, see
+ * {@link #LEGACY_DOCUMENT_ID}.
*/
- public static final String DOCUMENT_ID = "CamelIngestDocumentId";
+ public static final String DOCUMENT_ID =
"CamelLangChain4jIngestDocumentId";
+
+ /**
+ * The 3.39 name, read as a fallback when {@link #DOCUMENT_ID} is absent.
+ *
+ * @deprecated set {@link #DOCUMENT_ID} instead
+ */
+ @Deprecated(since = "3.40.0", forRemoval = true)
+ public static final String LEGACY_DOCUMENT_ID = "CamelIngestDocumentId";
private IngestHeaders() {
}
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..9155f9db91 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
@@ -20,6 +20,7 @@ import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
@@ -38,6 +39,7 @@ 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.ExpressionAdapter;
import org.apache.camel.support.builder.ExpressionBuilder;
import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
import org.apache.camel.util.URISupport;
@@ -180,7 +182,7 @@ public class IngestRoutes extends RouteBuilder {
// 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);
+ Expression documentId = documentIdExpression(name, runtime,
Exchange.FILE_NAME);
maybeAutoCreateRepository(name, runtime);
String repositoryName = runtime == null ? null :
runtime.source().idempotentRepository().orElse(null);
var endpoint = file(directory)
@@ -219,7 +221,7 @@ public class IngestRoutes extends RouteBuilder {
*/
private void configureEndpointSource(String name, String uri,
IngestRunTimeConfig.PipelineRunTimeConfig runtime, IngestService
service) {
- Expression documentId = documentIdExpression(runtime,
IngestHeaders.DOCUMENT_ID);
+ Expression documentId = documentIdExpression(name, runtime,
IngestHeaders.DOCUMENT_ID);
maybeAutoCreateRepository(name, runtime);
String repositoryName = runtime == null ? null :
runtime.source().idempotentRepository().orElse(null);
if (repositoryName == null) {
@@ -272,10 +274,38 @@ public class IngestRoutes extends RouteBuilder {
return id;
}
- private Expression
documentIdExpression(IngestRunTimeConfig.PipelineRunTimeConfig runtime,
+ @SuppressWarnings("deprecation")
+ private Expression documentIdExpression(String name,
IngestRunTimeConfig.PipelineRunTimeConfig runtime,
String defaultHeader) {
String configured = runtime == null ? null :
runtime.source().documentId().orElse(null);
if (configured == null) {
+ if (IngestHeaders.DOCUMENT_ID.equals(defaultHeader)) {
+ // the current name first, the 3.39 name as fallback; each
name keeps Camel's
+ // header-then-exchange-property lookup. The deprecation is
warned once per
+ // pipeline - literal-setting producers get no compiler signal
+ Expression current =
ExpressionBuilder.headerExpression(IngestHeaders.DOCUMENT_ID);
+ Expression legacy =
ExpressionBuilder.headerExpression(IngestHeaders.LEGACY_DOCUMENT_ID);
+ current.init(getContext());
+ legacy.init(getContext());
+ return new ExpressionAdapter() {
+ private final AtomicBoolean warned = new AtomicBoolean();
+
+ @Override
+ public Object evaluate(Exchange exchange) {
+ Object id = current.evaluate(exchange, Object.class);
+ if (id != null) {
+ return id;
+ }
+ id = legacy.evaluate(exchange, Object.class);
+ if (id != null && warned.compareAndSet(false, true)) {
+ LOG.warnf("Ingestion pipeline '%s': document id
read from the deprecated %s name"
+ + " - switch the producer to %s",
+ name, IngestHeaders.LEGACY_DOCUMENT_ID,
IngestHeaders.DOCUMENT_ID);
+ }
+ return id;
+ }
+ };
+ }
return ExpressionBuilder.headerExpression(defaultHeader);
}
// a bare header name is read as a header directly rather than parsed:
a dotted header
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 ce84df44c2..f6320321d5 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
@@ -93,7 +93,8 @@ public interface IngestRunTimeConfig {
* untouched, while a `+${...}+` in a properties file would be
consumed as a config
* expansion before Camel ever saw it. When not set, a pipeline
reading a directory
* uses the file name, and one consuming from a component uses the
- * `CamelIngestDocumentId` header.
+ * `CamelLangChain4jIngestDocumentId` header (the deprecated 3.39
name
+ * `CamelIngestDocumentId` is still read as a fallback).
*/
Optional<String> documentId();
}
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..d496f4f474 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
@@ -19,6 +19,7 @@ package
org.apache.camel.quarkus.component.langchain4j.ingest.it;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
+import java.util.Map;
import dev.langchain4j.data.segment.TextSegment;
import dev.langchain4j.model.embedding.EmbeddingModel;
@@ -68,6 +69,10 @@ public class IngestResource {
@Named("events-store")
EmbeddingStore<TextSegment> eventsStore;
+ @Inject
+ @Named("jdbc-store")
+ EmbeddingStore<TextSegment> jdbcStore;
+
@Inject
ProducerTemplate producerTemplate;
@@ -106,6 +111,7 @@ public class IngestResource {
case "datasheets" -> datasheetsStore;
case "s3" -> s3Store;
case "events" -> eventsStore;
+ case "jdbc" -> jdbcStore;
default -> productsStore;
};
// the deterministic test model gives a query no semantic pull towards
any document, so
@@ -142,4 +148,31 @@ public class IngestResource {
return result.outcome().label();
}
+ /** Feeds a pipeline carrying different ids in the current and the
deprecated header, so tests can assert precedence. */
+ @POST
+
@jakarta.ws.rs.Path("/feed-both/{pipeline}/{documentId}/{legacyDocumentId}")
+ @Consumes(MediaType.TEXT_PLAIN)
+ @Produces(MediaType.TEXT_PLAIN)
+ @SuppressWarnings("deprecation")
+ public String feedBoth(@PathParam("pipeline") String pipeline,
@PathParam("documentId") String documentId,
+ @PathParam("legacyDocumentId") String legacyDocumentId, String
content) {
+ IngestResult result = producerTemplate.requestBodyAndHeaders("direct:"
+ pipeline + "-feed", content,
+ Map.of(IngestHeaders.DOCUMENT_ID, documentId,
IngestHeaders.LEGACY_DOCUMENT_ID, legacyDocumentId),
+ IngestResult.class);
+ return result.outcome().label();
+ }
+
+ /** Feeds a pipeline carrying the id in the deprecated 3.39 header, so
tests can assert the fallback. */
+ @POST
+ @jakarta.ws.rs.Path("/feed-legacy/{pipeline}/{documentId:.+}")
+ @Consumes(MediaType.TEXT_PLAIN)
+ @Produces(MediaType.TEXT_PLAIN)
+ @SuppressWarnings("deprecation")
+ public String feedLegacy(@PathParam("pipeline") String pipeline,
@PathParam("documentId") String documentId,
+ String content) {
+ IngestResult result = producerTemplate.requestBodyAndHeader("direct:"
+ pipeline + "-feed", content,
+ IngestHeaders.LEGACY_DOCUMENT_ID, documentId,
IngestResult.class);
+ return result.outcome().label();
+ }
+
}
diff --git
a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
index f8caf3a942..91ce3bd0b6 100644
---
a/integration-tests/langchain4j-ingest/src/main/resources/application.properties
+++
b/integration-tests/langchain4j-ingest/src/main/resources/application.properties
@@ -35,7 +35,7 @@
quarkus.camel.langchain4j.ingest.products.source.idempotent-repository=propertie
# 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.document-id=$simple{header.CamelLangChain4jIngestDocumentId}
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
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
index 74a98d5c20..7a67fea8b0 100644
---
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
@@ -178,4 +178,37 @@ class Langchain4jIngestIdempotentTest {
.then().statusCode(200).extract().asString();
assertEquals("true", contains);
}
+
+ /** When both headers are set, the current one wins. */
+ @Test
+ void currentDocumentIdHeaderWinsOverLegacy() {
+ String outcome = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The SIGMA-8 relay carries two ids.")
+
.post("/langchain4j-ingest/feed-both/jdbcdocs/current-sigma.txt/legacy-sigma.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("ingested", outcome);
+
+ assertEquals("current-sigma.txt",
+ Langchain4jIngestTest.hit("Which relay carries two ids?",
"jdbc", "SIGMA-8").get("documentId"));
+ }
+
+ /**
+ * The deprecated 3.39 header still works and addresses the same identity:
a legacy delivery
+ * is deduplicated against a current-header one.
+ */
+ @Test
+ void legacyDocumentIdHeaderStillWorks() {
+ String first = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The LAMBDA-6 valve arrives under the legacy header.")
+
.post("/langchain4j-ingest/feed-legacy/jdbcdocs/legacy/lambda.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("ingested", first);
+
+ // the same id through the current header: one identity, so the
register answers skipped
+ String second = RestAssured.given().contentType(ContentType.TEXT)
+ .body("The LAMBDA-6 valve tries again under the current
header.")
+ .post("/langchain4j-ingest/feed/jdbcdocs/legacy/lambda.txt")
+ .then().statusCode(200).extract().asString();
+ assertEquals("skipped", second);
+ }
}
