You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "JiriOndrusek (via GitHub)" <gi...@apache.org> on 2026/09/08 08:43:37 UTC
[PR] CAMEL-24621: camel-langchain4j-ingest - new document ingestion component [camel]
JiriOndrusek opened a new pull request, #26190:
URL: https://github.com/apache/camel/pull/26190
fixes https://issues.apache.org/jira/browse/CAMEL-24621
# What this brings
New component camel-langchain4j-ingest: the ingestion half of a RAG setup. Documents flowing through a Camel route are split into segments, embedded and written into a LangChain4j EmbeddingStore — turning any Camel consumer (a directory, S3, Kafka, …) into a feed for a knowledge base, with optional deduplication by document id.
The feature originated in Camel Quarkus, where it shipped as the camel-quarkus-langchain4j-ingest extension (CQ 3.39.0). This PR ports the engine to Apache Camel so that the Quarkus extension can delegate to it and other runtimes (Camel Main, Spring Boot, JBang) get the same capability, with Kamelets to follow. Support level: Preview.
Follow-ups planned separately: a set of Kamelets in camel-kamelets wrapping this endpoint (file source + tika/docling parsing sinks), and the Camel Quarkus delegation.
# Technical details
A producer-only endpoint, langchain4j-ingest:pipelineName, backed by LangChain4j's own EmbeddingStoreIngestor (fully explicit builder — no ServiceLoader defaults can leak in):
```
exchange
│
├─ resolve document id
│ CamelLangChain4jIngestDocumentId exchange property (set pre-parse by the route,
│ else <documentIdHeader> header so a parsed document cannot
│ else FAIL forge its own identity)
│
├─ idempotentRepository configured?
│ yes: add(exchange,id)? ──false──> reply SKIPPED, done (eager claim, before
│ the body is read)
├─ blank text ────────────────────────> reply EMPTY (claim released)
├─ text > maxDocumentSize ────────────> FAIL (claim released)
│
├─ split ── default recursive splitter (maxSegmentSize/maxOverlapSize)
│ or a DocumentSplitter bean (documentSplitter=#bean:...)
├─ embed ── batched (embeddingBatchSize, default 32) ──> EmbeddingModel
├─ store ── ONE addAll after all segments are embedded ──> EmbeddingStore
│ (a mid-embedding failure writes nothing)
│
└─ reply IngestResult(pipeline, documentId, segmentsWritten,
INGESTED | EMPTY | SKIPPED) (claim confirmed on
INGESTED only)
```
- Every segment carries camel_ingest_pipeline and camel_ingest_document_id metadata, so retrieval can cite its source and a future synchronising engine can find a document's vectors again.
- embeddingStore/embeddingModel are autowired from the registry when unambiguous; zero or several candidates fail the start with the option to set. The idempotent repository is referenced by name on purpose and used through the Exchange-aware overloads, matching the idempotent consumer EIP; claim semantics (eager claim, release on EMPTY/failure) are documented including the at-least-once caveat.
- TikaTextDecode ships as a helper for the documented parse-route shape: it decodes Tika's pinned-encoding text output without consulting the exchange charset heuristic (CamelCharsetName header outranks the property, so convertBodyTo(String, "UTF-8") would be no substitute).
- Tests: 28 unit tests (engine, dedup, bean resolution, validation, real Tika HTML+PDF and a stubbed Docling Serve behind hand-rolled routes) plus an integration test against a real Qdrant container with a real in-process embedding model, asserting semantic retrieval with deliberately no shared vocabulary.
- Docs cover the directory and consumer recipes, dedup limits, the size cap (whole-document-in-memory by design) and parser hygiene; catalog, DSL builder factories, bom/coverage wiring included.
# Description
<!--
- Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
-->
# Target
- [ ] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch)
# Tracking
- [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it).
<!--
# *Note*: trivial changes like, typos, minor documentation fixes and other small items do not require a JIRA issue. In this case your pull request should address just this issue, without pulling in other changes.
-->
# Apache Camel coding standards and style
- [ ] I checked that each commit in the pull request has a meaningful subject line and body.
<!--
If you're unsure, you can format the pull request title like `[CAMEL-XXX] Fixes bug in camel-file component`, where you replace `CAMEL-XXX` with the appropriate JIRA issue.
-->
- [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes.
<!--
You can run the aforementioned command in your module so that the build auto-formats your code. This will also be verified as part of the checks and your PR may be rejected if if there are uncommited changes after running `mvn clean install -DskipTests`.
You can learn more about the contribution guidelines at https://github.com/apache/camel/blob/main/CONTRIBUTING.md
-->
# AI-assisted contributions
- [ ] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24621: camel-langchain4j-ingest - new document ingestion component [camel]
Posted by "gnodet (via GitHub)" <gi...@apache.org>.
gnodet commented on code in PR #26190:
URL: https://github.com/apache/camel/pull/26190#discussion_r3956597532
##########
components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.langchain4j.ingest;
+
+import java.util.Set;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.service.ServiceHelper;
+
+/**
+ * Splits the message body into segments, embeds them in batches and writes them to the embedding store; the message
+ * body is replaced with the {@link IngestResult}.
+ *
+ * <p>
+ * With an {@code idempotentRepository} configured, the producer claims the document id before writing: a duplicate
+ * delivery is answered {@code SKIPPED} without touching the store, a blank document releases its claim so a later,
+ * populated delivery under the same id still ingests, and a failed write releases it so the delivery can be retried.
+ * The eager claim has a documented consequence: a duplicate racing an in-flight first delivery is answered
+ * {@code SKIPPED} even if that delivery then fails and releases the id — with an at-least-once source the skipped
+ * duplicate is acknowledged, so the failed original must be redelivered by its own source or the document is in neither
+ * the store nor a queue. Claim-on-success semantics are the eventual alternative for sources that cannot redeliver.
+ */
+public class LangChain4jIngestProducer extends DefaultProducer {
+
+ private final LangChain4jIngestEndpoint endpoint;
+ private final LangChain4jIngestConfiguration configuration;
+ private IngestService service;
+
+ public LangChain4jIngestProducer(LangChain4jIngestEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ String pipeline = endpoint.getPipelineName();
+ if (pipeline == null || pipeline.isBlank()) {
+ throw new IllegalArgumentException(
+ "The ingestion pipeline name is missing. Use langchain4j-ingest:pipelineName in the endpoint URI.");
+ }
+ if (configuration.getDocumentIdHeader() == null || configuration.getDocumentIdHeader().isBlank()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': documentIdHeader must not be blank. Omit the option for"
+ + " the default, or name the header carrying the document id.");
+ }
+ if (configuration.getMaxSegmentSize() <= 0 || configuration.getMaxOverlapSize() < 0
+ || configuration.getMaxOverlapSize() >= configuration.getMaxSegmentSize()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxSegmentSize must be positive and maxOverlapSize must"
+ + " be non-negative and smaller than it (got "
+ + configuration.getMaxSegmentSize()
+ + " / " + configuration.getMaxOverlapSize() + ")");
+ }
Review Comment:
🐛 **Bug: `maxSegmentSize`/`maxOverlapSize` validation fires even when `documentSplitter` is set**
The `documentSplitter` option's Javadoc (in `LangChain4jIngestConfiguration`) explicitly says: _"When set, `maxSegmentSize` and `maxOverlapSize` are ignored (they parameterise the default splitter only)."_ But this validation block runs unconditionally — if a user sets `documentSplitter=#bean:mySplitter` AND explicitly passes `maxSegmentSize=0` (thinking it's irrelevant), the endpoint fails to start with a confusing error about an option the docs say is ignored.
The defaults (500/50) mean this is only triggered when a user explicitly sets these options alongside a custom splitter, but the contradiction between documentation and behaviour is still a real bug.
```suggestion
if (configuration.getDocumentSplitter() == null
&& (configuration.getMaxSegmentSize() <= 0 || configuration.getMaxOverlapSize() < 0
|| configuration.getMaxOverlapSize() >= configuration.getMaxSegmentSize())) {
throw new IllegalArgumentException(
"Ingestion pipeline '" + pipeline + "': maxSegmentSize must be positive and maxOverlapSize must"
+ " be non-negative and smaller than it (got "
+ configuration.getMaxSegmentSize()
+ " / " + configuration.getMaxOverlapSize() + ")");
}
```
##########
components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.langchain4j.ingest;
+
+import java.util.Set;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.service.ServiceHelper;
+
+/**
+ * Splits the message body into segments, embeds them in batches and writes them to the embedding store; the message
+ * body is replaced with the {@link IngestResult}.
+ *
+ * <p>
+ * With an {@code idempotentRepository} configured, the producer claims the document id before writing: a duplicate
+ * delivery is answered {@code SKIPPED} without touching the store, a blank document releases its claim so a later,
+ * populated delivery under the same id still ingests, and a failed write releases it so the delivery can be retried.
+ * The eager claim has a documented consequence: a duplicate racing an in-flight first delivery is answered
+ * {@code SKIPPED} even if that delivery then fails and releases the id — with an at-least-once source the skipped
+ * duplicate is acknowledged, so the failed original must be redelivered by its own source or the document is in neither
+ * the store nor a queue. Claim-on-success semantics are the eventual alternative for sources that cannot redeliver.
+ */
+public class LangChain4jIngestProducer extends DefaultProducer {
+
+ private final LangChain4jIngestEndpoint endpoint;
+ private final LangChain4jIngestConfiguration configuration;
+ private IngestService service;
+
+ public LangChain4jIngestProducer(LangChain4jIngestEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ String pipeline = endpoint.getPipelineName();
+ if (pipeline == null || pipeline.isBlank()) {
+ throw new IllegalArgumentException(
+ "The ingestion pipeline name is missing. Use langchain4j-ingest:pipelineName in the endpoint URI.");
+ }
+ if (configuration.getDocumentIdHeader() == null || configuration.getDocumentIdHeader().isBlank()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': documentIdHeader must not be blank. Omit the option for"
+ + " the default, or name the header carrying the document id.");
+ }
+ if (configuration.getMaxSegmentSize() <= 0 || configuration.getMaxOverlapSize() < 0
+ || configuration.getMaxOverlapSize() >= configuration.getMaxSegmentSize()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxSegmentSize must be positive and maxOverlapSize must"
+ + " be non-negative and smaller than it (got "
+ + configuration.getMaxSegmentSize()
+ + " / " + configuration.getMaxOverlapSize() + ")");
+ }
+
+ if (configuration.getEmbeddingBatchSize() < 1) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': embeddingBatchSize must be positive (got "
+ + configuration.getEmbeddingBatchSize() + ")");
+ }
+
+ if (configuration.getMaxDocumentSize() < 0) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxDocumentSize must not be negative (got "
+ + configuration.getMaxDocumentSize() + ")");
+ }
+
+ EmbeddingStore<TextSegment> store
+ = resolve(EmbeddingStore.class, configuration.getEmbeddingStore(), "embedding store", "embeddingStore");
+ EmbeddingModel model
+ = resolve(EmbeddingModel.class, configuration.getEmbeddingModel(), "embedding model", "embeddingModel");
+ service = configuration.getDocumentSplitter() != null
+ ? new IngestService(
+ pipeline, store, model, configuration.getDocumentSplitter(),
+ configuration.getEmbeddingBatchSize(), configuration.getMaxDocumentSize())
+ : new IngestService(
+ pipeline, store, model, configuration.getMaxSegmentSize(),
+ configuration.getMaxOverlapSize(), configuration.getEmbeddingBatchSize(),
+ configuration.getMaxDocumentSize());
+
+ IdempotentRepository repository = configuration.getIdempotentRepository();
+ if (repository != null) {
+ // a repository bound via configuration does not pass through the registry's bind
+ // hook, so a CamelContextAware implementation would otherwise run contextless. It is
+ // started but never stopped here: the bean may be shared, and stopping the in-memory
+ // repository would clear it
+ CamelContextAware.trySetCamelContext(repository, getEndpoint().getCamelContext());
+ ServiceHelper.startService(repository);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ String documentId = resolveDocumentId(exchange);
+
+ IdempotentRepository repository = configuration.getIdempotentRepository();
+ if (repository == null) {
+ exchange.getMessage().setBody(service.ingest(documentId, exchange.getMessage().getBody(String.class)));
+ return;
+ }
+
+ // the claim is eager: a duplicate racing an in-flight first delivery is answered SKIPPED
+ // even if that delivery then fails and releases the id. The claim also precedes the body
+ // read, so a skipped duplicate never pays for materialising a large payload. The
+ // Exchange-aware repository overloads are used throughout, matching the idempotent
+ // consumer EIP - a repository overriding only those variants is not bypassed
+ if (!repository.add(exchange, documentId)) {
+ exchange.getMessage().setBody(
+ new IngestResult(service.pipeline(), documentId, 0, IngestResult.Outcome.SKIPPED));
+ return;
+ }
+ IngestResult result;
+ try {
+ result = service.ingest(documentId, exchange.getMessage().getBody(String.class));
+ } catch (Exception e) {
+ // a failed write must not keep the claim, or the delivery could never be retried.
+ // An Error (an OutOfMemoryError, say) is deliberately not caught: under a VM-level
+ // failure the release itself could not be trusted, so the id stays claimed and later
+ // deliveries are answered SKIPPED - clear it from the repository to re-ingest
+ try {
+ repository.remove(exchange, documentId);
+ } catch (Exception rollback) {
+ // the release often fails from the same root cause; the original failure is
+ // the one worth reporting
+ e.addSuppressed(rollback);
+ }
+ throw e;
+ }
+ if (result.outcome() == IngestResult.Outcome.EMPTY) {
+ // a blank document wrote nothing, so it must not keep the claim - a later, populated
+ // delivery under the same id would be answered SKIPPED
+ repository.remove(exchange, documentId);
+ } else {
+ repository.confirm(exchange, documentId);
+ }
+ exchange.getMessage().setBody(result);
+ }
+
+ /**
+ * The exchange property wins over the header: a route that parses documents captures the id into the property
+ * before the parse, and a parser (Tika) copies document metadata over the headers, so a header read here could be
+ * spoofed by the document itself. A property that is present but blank fails the exchange on purpose, without
+ * falling back to the header: the route deliberately captured the id, so a blank capture is a broken expression to
+ * surface loudly, not a case to paper over with a value of weaker provenance.
+ */
+ private String resolveDocumentId(Exchange exchange) {
+ String documentId = exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, String.class);
+ if (documentId == null) {
+ documentId = exchange.getMessage().getHeader(configuration.getDocumentIdHeader(), String.class);
+ }
+ if (documentId == null || documentId.isBlank()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + endpoint.getPipelineName() + "': no document id. Set the "
+ + configuration.getDocumentIdHeader()
+ + " header (or the " + LangChain4jIngest.DOCUMENT_ID_PROPERTY
+ + " exchange property), or point the documentIdHeader endpoint option"
+ + " at where the consumer puts it.");
+ }
+ return documentId;
+ }
+
+ /**
+ * Resolves a configured bean, or the single registry bean of the type. Picking one of several silently would bind
+ * the pipeline to whichever bean happened to be found first, so zero and several candidates each fail with the fix
+ * in the message. Runs after autowiring: with exactly one candidate the configuration already carries it.
+ */
+ @SuppressWarnings("unchecked")
+ private <T> T resolve(Class<?> type, T configured, String what, String option) {
Review Comment:
💡 **Design note: `resolve()` takes `Class<?>` instead of `Class<T>`, breaking the type link at the call site**
The signature:
```java
private <T> T resolve(Class<?> type, T configured, String what, String option)
```
uses `Class<?>` rather than `Class<T>`, so the compiler cannot enforce that `type` and `T` match. This compiles without error:
```java
EmbeddingStore<TextSegment> store = resolve(EmbeddingModel.class, configuration.getEmbeddingStore(), ...);
```
— the wrong class is passed, `findByType(EmbeddingModel.class)` returns candidates of the wrong type, and the unchecked cast silently produces a `ClassCastException` at use time.
No caller today passes a mismatched class, but the safety net is gone. Consider:
```suggestion
@SuppressWarnings("unchecked")
private <T> T resolve(Class<T> type, T configured, String what, String option) {
```
This requires the two callers to adjust the `EmbeddingStore.class` invocation to pass the raw type through an unchecked cast at the call site — or simply accept that the generic bound is only partially expressible here and leave a comment to that effect.
##########
components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.langchain4j.ingest;
+
+import java.util.Set;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.service.ServiceHelper;
+
+/**
+ * Splits the message body into segments, embeds them in batches and writes them to the embedding store; the message
+ * body is replaced with the {@link IngestResult}.
+ *
+ * <p>
+ * With an {@code idempotentRepository} configured, the producer claims the document id before writing: a duplicate
+ * delivery is answered {@code SKIPPED} without touching the store, a blank document releases its claim so a later,
+ * populated delivery under the same id still ingests, and a failed write releases it so the delivery can be retried.
+ * The eager claim has a documented consequence: a duplicate racing an in-flight first delivery is answered
+ * {@code SKIPPED} even if that delivery then fails and releases the id — with an at-least-once source the skipped
+ * duplicate is acknowledged, so the failed original must be redelivered by its own source or the document is in neither
+ * the store nor a queue. Claim-on-success semantics are the eventual alternative for sources that cannot redeliver.
+ */
+public class LangChain4jIngestProducer extends DefaultProducer {
+
+ private final LangChain4jIngestEndpoint endpoint;
+ private final LangChain4jIngestConfiguration configuration;
+ private IngestService service;
+
+ public LangChain4jIngestProducer(LangChain4jIngestEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ String pipeline = endpoint.getPipelineName();
+ if (pipeline == null || pipeline.isBlank()) {
+ throw new IllegalArgumentException(
+ "The ingestion pipeline name is missing. Use langchain4j-ingest:pipelineName in the endpoint URI.");
+ }
+ if (configuration.getDocumentIdHeader() == null || configuration.getDocumentIdHeader().isBlank()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': documentIdHeader must not be blank. Omit the option for"
+ + " the default, or name the header carrying the document id.");
+ }
+ if (configuration.getMaxSegmentSize() <= 0 || configuration.getMaxOverlapSize() < 0
+ || configuration.getMaxOverlapSize() >= configuration.getMaxSegmentSize()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxSegmentSize must be positive and maxOverlapSize must"
+ + " be non-negative and smaller than it (got "
+ + configuration.getMaxSegmentSize()
+ + " / " + configuration.getMaxOverlapSize() + ")");
+ }
+
+ if (configuration.getEmbeddingBatchSize() < 1) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': embeddingBatchSize must be positive (got "
+ + configuration.getEmbeddingBatchSize() + ")");
+ }
+
+ if (configuration.getMaxDocumentSize() < 0) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxDocumentSize must not be negative (got "
+ + configuration.getMaxDocumentSize() + ")");
+ }
+
+ EmbeddingStore<TextSegment> store
+ = resolve(EmbeddingStore.class, configuration.getEmbeddingStore(), "embedding store", "embeddingStore");
+ EmbeddingModel model
+ = resolve(EmbeddingModel.class, configuration.getEmbeddingModel(), "embedding model", "embeddingModel");
+ service = configuration.getDocumentSplitter() != null
+ ? new IngestService(
+ pipeline, store, model, configuration.getDocumentSplitter(),
+ configuration.getEmbeddingBatchSize(), configuration.getMaxDocumentSize())
+ : new IngestService(
+ pipeline, store, model, configuration.getMaxSegmentSize(),
+ configuration.getMaxOverlapSize(), configuration.getEmbeddingBatchSize(),
+ configuration.getMaxDocumentSize());
+
+ IdempotentRepository repository = configuration.getIdempotentRepository();
+ if (repository != null) {
+ // a repository bound via configuration does not pass through the registry's bind
+ // hook, so a CamelContextAware implementation would otherwise run contextless. It is
+ // started but never stopped here: the bean may be shared, and stopping the in-memory
+ // repository would clear it
+ CamelContextAware.trySetCamelContext(repository, getEndpoint().getCamelContext());
+ ServiceHelper.startService(repository);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ String documentId = resolveDocumentId(exchange);
+
+ IdempotentRepository repository = configuration.getIdempotentRepository();
+ if (repository == null) {
+ exchange.getMessage().setBody(service.ingest(documentId, exchange.getMessage().getBody(String.class)));
+ return;
+ }
+
+ // the claim is eager: a duplicate racing an in-flight first delivery is answered SKIPPED
+ // even if that delivery then fails and releases the id. The claim also precedes the body
+ // read, so a skipped duplicate never pays for materialising a large payload. The
+ // Exchange-aware repository overloads are used throughout, matching the idempotent
+ // consumer EIP - a repository overriding only those variants is not bypassed
+ if (!repository.add(exchange, documentId)) {
+ exchange.getMessage().setBody(
+ new IngestResult(service.pipeline(), documentId, 0, IngestResult.Outcome.SKIPPED));
+ return;
+ }
+ IngestResult result;
+ try {
+ result = service.ingest(documentId, exchange.getMessage().getBody(String.class));
+ } catch (Exception e) {
+ // a failed write must not keep the claim, or the delivery could never be retried.
+ // An Error (an OutOfMemoryError, say) is deliberately not caught: under a VM-level
+ // failure the release itself could not be trusted, so the id stays claimed and later
+ // deliveries are answered SKIPPED - clear it from the repository to re-ingest
+ try {
+ repository.remove(exchange, documentId);
+ } catch (Exception rollback) {
+ // the release often fails from the same root cause; the original failure is
+ // the one worth reporting
+ e.addSuppressed(rollback);
+ }
+ throw e;
+ }
Review Comment:
✅ **`catch (Exception)` here is intentional — not a bug (ast-grep `broad-exception-catch` flag is a false positive)**
ast-grep flagged this block, but the pattern is correct and well-commented: the catch is intentional dedup-claim cleanup before re-throw, and `Error` is deliberately excluded per the inline comment (an OOM during embedding should keep the claim poisoned so a later delivery sees SKIPPED rather than re-attempting an already partially-written ingest). The `catch (Exception rollback)` nested block (line 144) is equally intentional — a repository failure during claim release is suppressed onto the original exception so the root cause is preserved. Both catches re-throw, so no exception is swallowed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24621: camel-langchain4j-ingest - new document ingestion component [camel]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on PR #26190:
URL: https://github.com/apache/camel/pull/26190#issuecomment-5581997084
FYI @jamesnetherton , @zbendhiba
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24621: camel-langchain4j-ingest - new document ingestion component [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26190:
URL: https://github.com/apache/camel/pull/26190#issuecomment-5583121989
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24621: camel-langchain4j-ingest - new document ingestion component [camel]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on PR #26190:
URL: https://github.com/apache/camel/pull/26190#issuecomment-5584462636
Thanks for the review! Both findings are addressed in the amended commit (58d7d97a3c2):
- the splitter bounds are now validated only when no `documentSplitter` is configured (your suggestion verbatim), with a test proving a custom-splitter endpoint starts despite explicit nonsensical sizes;
- `resolve` is now `resolve(Class<? super T> type, T configured, ...)` — both call sites are cast-free and a swapped class literal fails to compile at either one.
Details in the inline threads. CI is green.
_Claude Code on behalf of @JiriOndrusek_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24621: camel-langchain4j-ingest - new document ingestion component [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #26190:
URL: https://github.com/apache/camel/pull/26190
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24621: camel-langchain4j-ingest - new document ingestion component [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26190:
URL: https://github.com/apache/camel/pull/26190#issuecomment-5583292564
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `bom/camel-bom`
- `catalog/camel-allcomponents`
- `catalog/camel-catalog`
- `components/camel-ai`
- `components/camel-ai/camel-langchain4j-ingest`
- `core/camel-main`
- `coverage`
- `docs`
- `dsl/camel-componentdsl`
- `dsl/camel-endpointdsl`
- `dsl/camel-kamelet-main`
- `parent`
- `tooling/maven/camel-package-maven-plugin`
> :information_source: Dependent modules were not tested because the total number of affected modules exceeded the threshold (50). Use the `test-dependents` label to force testing all dependents.
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 590 tested, 20 compile-only — current: 70 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **610 affected modules** (current approach: 70).
<details><summary>:warning: Modules only in Scalpel (610)</summary>
- `apache-camel`
- `archetypes`
- `camel-a2a`
- `camel-activemq`
- `camel-activemq6`
- `camel-ai-observability`
- `camel-ai-observability-api`
- `camel-ai-parent`
- `camel-ai-resource`
- `camel-ai-tool`
- `camel-alibaba-common`
- `camel-alibaba-eventbridge`
- `camel-alibaba-fc`
- `camel-alibaba-kms`
- `camel-alibaba-mns`
- `camel-alibaba-oss`
- `camel-alibaba-ots`
- `camel-alibaba-parent`
- `camel-alibaba-sls`
- `camel-alibaba-sms`
- `camel-allcomponents`
- `camel-amqp`
- `camel-api`
- `camel-api-component-maven-plugin`
- `camel-arangodb`
- `camel-archetype-api-component`
- `camel-archetype-component`
- `camel-archetype-dataformat`
- `camel-archetype-java`
- `camel-archetype-main`
- `camel-as2`
- `camel-as2-api`
- `camel-as2-parent`
- `camel-asn1`
- `camel-asterisk`
- `camel-atmosphere-websocket`
- `camel-atom`
- `camel-attachments`
- `camel-avro`
- `camel-avro-rpc`
- `camel-avro-rpc-jetty`
- `camel-avro-rpc-parent`
- `camel-avro-rpc-spi`
- `camel-aws-bedrock`
- `camel-aws-cloudtrail`
- `camel-aws-common`
- `camel-aws-config`
- `camel-aws-parameter-store`
- `camel-aws-parent`
- `camel-aws-secrets-manager`
- `camel-aws-security-hub`
- `camel-aws2-athena`
- `camel-aws2-comprehend`
- `camel-aws2-cw`
- `camel-aws2-ddb`
- `camel-aws2-ec2`
- `camel-aws2-ecs`
- `camel-aws2-eks`
- `camel-aws2-eventbridge`
- `camel-aws2-iam`
- `camel-aws2-kinesis`
- `camel-aws2-kms`
- `camel-aws2-lambda`
- `camel-aws2-mq`
- `camel-aws2-msk`
- `camel-aws2-polly`
- `camel-aws2-redshift`
- `camel-aws2-rekognition`
- `camel-aws2-s3`
- `camel-aws2-s3-vectors`
- `camel-aws2-ses`
- `camel-aws2-sns`
- `camel-aws2-sqs`
- `camel-aws2-step-functions`
- `camel-aws2-sts`
- `camel-aws2-textract`
- `camel-aws2-timestream`
- `camel-aws2-transcribe`
- `camel-aws2-translate`
- `camel-azure-common`
- `camel-azure-cosmosdb`
- `camel-azure-eventgrid`
- `camel-azure-eventhubs`
- `camel-azure-files`
- `camel-azure-functions`
- `camel-azure-key-vault`
- `camel-azure-parent`
- `camel-azure-schema-registry`
- `camel-azure-servicebus`
- `camel-azure-storage-blob`
- `camel-azure-storage-datalake`
- `camel-azure-storage-queue`
- `camel-barcode`
- `camel-base`
- `camel-base-engine`
- `camel-base64`
- `camel-bean`
- `camel-bean-validator`
- `camel-beanio`
- `camel-bindy`
- `camel-bom`
- `camel-bonita`
- `camel-box`
- `camel-box-api`
- `camel-box-parent`
- `camel-braintree`
- `camel-browse`
- `camel-caffeine`
- `camel-camunda`
- `camel-cassandraql`
- `camel-catalog`
- `camel-catalog-common`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-cbor`
- `camel-chatscript`
- `camel-chunk`
- `camel-cli-connector`
- `camel-cli-debug`
- `camel-clickhouse`
- `camel-clickup`
- `camel-cloudevents`
- `camel-cluster`
- `camel-cm-sms`
- `camel-coap`
- `camel-cometd`
- `camel-component-maven-plugin`
- `camel-componentdsl`
- `camel-console`
- `camel-consul`
- `camel-controlbus`
- `camel-core`
- `camel-core-all`
- `camel-core-catalog`
- `camel-core-engine`
- `camel-core-languages`
- `camel-core-model`
- `camel-core-processor`
- `camel-core-reifier`
- `camel-core-xml`
- `camel-couchbase`
- `camel-couchdb`
- `camel-cron`
- `camel-crypto`
- `camel-crypto-pgp`
- `camel-csv`
- `camel-cxf-common`
- `camel-cxf-parent`
- `camel-cxf-rest`
- `camel-cxf-soap`
- `camel-cxf-spring-common`
- `camel-cxf-spring-rest`
- `camel-cxf-spring-soap`
- `camel-cxf-spring-transport`
- `camel-cxf-transport`
- `camel-cyberark-vault`
- `camel-dapr`
- `camel-dataformat`
- `camel-dataset`
- `camel-datasonnet`
- `camel-dataweave`
- `camel-debezium-common`
- `camel-debezium-common-parent`
- `camel-debezium-db2`
- `camel-debezium-maven-plugin`
- `camel-debezium-mongodb`
- `camel-debezium-mysql`
- `camel-debezium-oracle`
- `camel-debezium-parent`
- `camel-debezium-postgres`
- `camel-debezium-sqlserver`
- `camel-debug`
- `camel-dependencies`
- `camel-dfdl`
- `camel-dhis2`
- `camel-dhis2-api`
- `camel-dhis2-parent`
- `camel-diagram`
- `camel-direct`
- `camel-disruptor`
- `camel-djl`
- `camel-dns`
- `camel-docker`
- `camel-docling`
- `camel-drill`
- `camel-dropbox`
- `camel-dsl-modeline`
- `camel-dsl-support`
- `camel-duckdb`
- `camel-dynamic-router`
- `camel-ehcache`
- `camel-eip-documentation-enricher-maven-plugin`
- `camel-elasticsearch`
- `camel-elasticsearch-rest-client`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-event`
- `camel-exec`
- `camel-fastjson`
- `camel-fhir`
- `camel-fhir-api`
- `camel-fhir-parent`
- `camel-file`
- `camel-file-watch`
- `camel-flatpack`
- `camel-flink`
- `camel-flowable`
- `camel-fop`
- `camel-fory`
- `camel-freemarker`
- `camel-ftp`
- `camel-ftp-common`
- `camel-geocoder`
- `camel-git`
- `camel-github2`
- `camel-google-bigquery`
- `camel-google-calendar`
- `camel-google-common`
- `camel-google-drive`
- `camel-google-firestore`
- `camel-google-functions`
- `camel-google-mail`
- `camel-google-parent`
- `camel-google-pubsub`
- `camel-google-secret-manager`
- `camel-google-sheets`
- `camel-google-speech-to-text`
- `camel-google-storage`
- `camel-google-text-to-speech`
- `camel-google-vertexai`
- `camel-google-vision`
- `camel-graphql`
- `camel-grok`
- `camel-groovy`
- `camel-grpc`
- `camel-gson`
- `camel-hashicorp-vault`
- `camel-hazelcast`
- `camel-health`
- `camel-hivemq`
- `camel-hl7`
- `camel-http`
- `camel-http-base`
- `camel-http-common`
- `camel-huawei-parent`
- `camel-huaweicloud-common`
- `camel-huaweicloud-dms`
- `camel-huaweicloud-frs`
- `camel-huaweicloud-functiongraph`
- `camel-huaweicloud-iam`
- `camel-huaweicloud-imagerecognition`
- `camel-huaweicloud-obs`
- `camel-huaweicloud-smn`
- `camel-huggingface`
- `camel-ibm-cos`
- `camel-ibm-parent`
- `camel-ibm-secrets-manager`
- `camel-ibm-watson-discovery`
- `camel-ibm-watson-language`
- `camel-ibm-watson-speech-to-text`
- `camel-ibm-watson-text-to-speech`
- `camel-ibm-watsonx-ai`
- `camel-ibm-watsonx-data`
- `camel-ical`
- `camel-iggy`
- `camel-ignite`
- `camel-infinispan`
- `camel-infinispan-common`
- `camel-infinispan-embedded`
- `camel-infinispan-parent`
- `camel-influxdb`
- `camel-influxdb2`
- `camel-iso8583`
- `camel-itest`
- `camel-jackson`
- `camel-jackson-avro`
- `camel-jackson-protobuf`
- `camel-jackson3`
- `camel-jackson3-avro`
- `camel-jackson3-protobuf`
- `camel-jackson3xml`
- `camel-jacksonxml`
- `camel-jactl`
- `camel-jandex`
- `camel-jasypt`
- `camel-java-io`
- `camel-java-joor-dsl`
- `camel-javascript`
- `camel-jaxb`
- `camel-jbang-console`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-mcp`
- `camel-jbang-parent`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-mcp`
- `camel-jbang-plugin-route-parser`
- `camel-jbang-plugin-test`
- `camel-jbang-plugin-tui`
- `camel-jbang-plugin-validate`
- `camel-jcache`
- `camel-jcr`
- `camel-jdbc`
- `camel-jetty`
- `camel-jetty-common`
- `camel-jfr`
- `camel-jgroups`
- `camel-jgroups-raft`
- `camel-jira`
- `camel-jms`
- `camel-jmx`
- `camel-jolt`
- `camel-jooq`
- `camel-joor`
- `camel-jpa`
- `camel-jq`
- `camel-jsch`
- `camel-jslt`
- `camel-json-validator`
- `camel-jsonapi`
- `camel-jsonata`
- `camel-jsonb`
- `camel-jsonpath`
- `camel-jsoup`
- `camel-jt400`
- `camel-jta`
- `camel-jte`
- `camel-kafka`
- `camel-kamelet`
- `camel-kamelet-main`
- `camel-kamelet-main-support`
- `camel-keycloak`
- `camel-knative`
- `camel-knative-api`
- `camel-knative-http`
- `camel-knative-parent`
- `camel-kserve`
- `camel-kubernetes`
- `camel-kudu`
- `camel-langchain4j-agent`
- `camel-langchain4j-agent-api`
- `camel-langchain4j-chat`
- `camel-langchain4j-core`
- `camel-langchain4j-embeddings`
- `camel-langchain4j-embeddingstore`
- `camel-langchain4j-embeddingstore-api`
- `camel-langchain4j-ingest`
- `camel-langchain4j-tokenizer`
- `camel-langchain4j-web-search`
- `camel-language`
- `camel-launcher`
- `camel-launcher-container`
- `camel-ldap`
- `camel-ldif`
- `camel-log`
- `camel-lra`
- `camel-lucene`
- `camel-lumberjack`
- `camel-lzf`
- `camel-mail`
- `camel-mail-microsoft-oauth`
- `camel-main`
- `camel-management`
- `camel-management-api`
- `camel-mapstruct`
- `camel-master`
- `camel-maven-plugin`
- `camel-mcp-server`
- `camel-mcp-server-api`
- `camel-mdc`
- `camel-metrics`
- `camel-micrometer`
- `camel-micrometer-observability`
- `camel-micrometer-prometheus`
- `camel-microprofile-config`
- `camel-microprofile-fault-tolerance`
- `camel-microprofile-health`
- `camel-microprofile-parent`
- `camel-milo`
- `camel-milvus`
- `camel-mina`
- `camel-mina-sftp`
- `camel-minio`
- `camel-mllp`
- `camel-mock`
- `camel-mongodb`
- `camel-mongodb-gridfs`
- `camel-mustache`
- `camel-mvel`
- `camel-mybatis`
- `camel-nats`
- `camel-neo4j`
- `camel-netty`
- `camel-netty-http`
- `camel-oaipmh`
- `camel-oauth`
- `camel-observability-services`
- `camel-observation`
- `camel-ocsf`
- `camel-ognl`
- `camel-olingo2`
- `camel-olingo2-api`
- `camel-olingo2-parent`
- `camel-olingo4`
- `camel-olingo4-api`
- `camel-olingo4-parent`
- `camel-once`
- `camel-opa`
- `camel-openai`
- `camel-openapi-java`
- `camel-openapi-rest-dsl-generator`
- `camel-openapi-validator`
- `camel-opensearch`
- `camel-openstack`
- `camel-opentelemetry`
- `camel-opentelemetry-metrics`
- `camel-opentelemetry2`
- `camel-optaplanner`
- `camel-package-maven-plugin`
- `camel-paho`
- `camel-paho-mqtt5`
- `camel-parquet-avro`
- `camel-pdf`
- `camel-pg-replication-slot`
- `camel-pgevent`
- `camel-pgvector`
- `camel-pinecone`
- `camel-platform-http`
- `camel-platform-http-jolokia`
- `camel-platform-http-main`
- `camel-platform-http-vertx`
- `camel-plc4x`
- `camel-pqc`
- `camel-printer`
- `camel-protobuf`
- `camel-pubnub`
- `camel-pulsar`
- `camel-python`
- `camel-python3`
- `camel-qdrant`
- `camel-quartz`
- `camel-quickfix`
- `camel-quickjs`
- `camel-reactive-streams`
- `camel-reactor`
- `camel-redis`
- `camel-ref`
- `camel-report-maven-plugin`
- `camel-resilience4j`
- `camel-resilience4j-micrometer`
- `camel-resourceresolver-github`
- `camel-rest`
- `camel-rest-openapi`
- `camel-rest-postman`
- `camel-restdsl-openapi-plugin`
- `camel-robotframework`
- `camel-rocketmq`
- `camel-route-parser`
- `camel-rss`
- `camel-rxjava`
- `camel-saga`
- `camel-salesforce`
- `camel-salesforce-codegen`
- `camel-salesforce-maven-plugin`
- `camel-salesforce-parent`
- `camel-sap-netweaver`
- `camel-saxon`
- `camel-scheduler`
- `camel-schematron`
- `camel-seda`
- `camel-servicenow`
- `camel-servicenow-maven-plugin`
- `camel-servicenow-parent`
- `camel-servlet`
- `camel-shell`
- `camel-shiro`
- `camel-sjms`
- `camel-sjms2`
- `camel-slack`
- `camel-smb`
- `camel-smooks`
- `camel-smpp`
- `camel-snakeyaml`
- `camel-snmp`
- `camel-soap`
- `camel-solr`
- `camel-spiffe`
- `camel-splunk-hec`
- `camel-spring`
- `camel-spring-ai-chat`
- `camel-spring-ai-embeddings`
- `camel-spring-ai-image`
- `camel-spring-ai-parent`
- `camel-spring-ai-vector-store`
- `camel-spring-batch`
- `camel-spring-cloud-config`
- `camel-spring-jdbc`
- `camel-spring-ldap`
- `camel-spring-main`
- `camel-spring-parent`
- `camel-spring-rabbitmq`
- `camel-spring-redis`
- `camel-spring-security`
- `camel-spring-ws`
- `camel-spring-xml`
- `camel-sql`
- `camel-ssh`
- `camel-state-store`
- `camel-state-store-parent`
- `camel-stax`
- `camel-stitch`
- `camel-stream`
- `camel-streamcaching-test`
- `camel-stringtemplate`
- `camel-stripe`
- `camel-stub`
- `camel-support`
- `camel-swift`
- `camel-syslog`
- `camel-tahu`
- `camel-tarfile`
- `camel-telegram`
- `camel-telemetry`
- `camel-telemetry-dev`
- `camel-tensorflow-serving`
- `camel-test-infra-all`
- `camel-test-infra-artemis`
- `camel-test-infra-cli`
- `camel-test-infra-core`
- `camel-test-infra-jetty`
- `camel-test-infra-qdrant`
- `camel-test-infra-smb`
- `camel-test-junit5`
- `camel-test-junit6`
- `camel-test-main-junit5`
- `camel-test-main-junit6`
- `camel-test-parent`
- `camel-test-spring-junit5`
- `camel-test-spring-junit6`
- `camel-thrift`
- `camel-thymeleaf`
- `camel-tika`
- `camel-timer`
- `camel-tooling-maven`
- `camel-toon`
- `camel-tracing`
- `camel-twilio`
- `camel-twitter`
- `camel-ubl`
- `camel-undertow`
- `camel-undertow-spring-security`
- `camel-univocity-parsers`
- `camel-util`
- `camel-validator`
- `camel-velocity`
- `camel-vertx`
- `camel-vertx-common`
- `camel-vertx-http`
- `camel-vertx-parent`
- `camel-vertx-websocket`
- `camel-wal`
- `camel-wasm`
- `camel-weather`
- `camel-weaviate`
- `camel-web3j`
- `camel-webhook`
- `camel-whatsapp`
- `camel-wordpress`
- `camel-workday`
- `camel-xchange`
- `camel-xj`
- `camel-xml-io`
- `camel-xml-io-dsl`
- `camel-xml-io-util`
- `camel-xml-jaxb`
- `camel-xml-jaxb-dsl`
- `camel-xml-jaxb-dsl-test`
- `camel-xml-jaxb-dsl-test-definition`
- `camel-xml-jaxb-dsl-test-spring`
- `camel-xml-jaxp`
- `camel-xml-jaxp-util`
- `camel-xmlsecurity`
- `camel-xmpp`
- `camel-xpath`
- `camel-xslt`
- `camel-xslt-saxon`
- `camel-yaml-dsl`
- `camel-yaml-dsl-common`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `camel-yaml-dsl-parent`
- `camel-yaml-dsl-validator`
- `camel-yaml-dsl-validator-maven-plugin`
- `camel-yaml-io`
- `camel-zendesk`
- `camel-zip-deflater`
- `camel-zipfile`
- `camel-zookeeper`
- `camel-zookeeper-master`
- `components`
- `core`
- `coverage`
- `docs`
- `dsl`
- `dummy-component`
- `sync-properties-maven-plugin`
</details>
<details><summary>Modules only in current approach (70)</summary>
- `[ERROR]`
- `[ERROR]`
- `[ERROR] Command execution failed.`
- `[ERROR] Failed to execute goal org.codehaus.mojo:exec-maven-plugin:3.6.3:exec (default-cli) on project camel-package-maven-plugin: Command execution failed. Process exited with an error: 127 (Exit value: 127) -> [Help 1]`
- `[ERROR] For more information about the errors and possible solutions, please read the following articles:`
- `[ERROR] Re-run Maven using the -X switch to enable full debug logging.`
- `[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.`
- `[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException`
- `at com.gradle.maven.cache.extension.c.a.a (SourceFile:46)`
- `at com.gradle.maven.cache.extension.c.a.a (SourceFile:46)`
- `at com.gradle.maven.cache.extension.c.b.a (SourceFile:62)`
- `at com.gradle.maven.cache.extension.c.b.c (SourceFile:118)`
- `at com.gradle.maven.cache.extension.c.c.a (SourceFile:26)`
- `at com.gradle.maven.cache.extension.c.d.a (SourceFile:27)`
- `at com.gradle.maven.cache.extension.c.g.a (SourceFile:27)`
- `at com.gradle.maven.cache.extension.c.h$1.run (SourceFile:35)`
- `at com.gradle.maven.cache.extension.c.h.a (SourceFile:31)`
- `at com.gradle.maven.cache.extension.c.j.a (SourceFile:28)`
- `at com.gradle.maven.cache.extension.c.m.a (SourceFile:80)`
- `at com.gradle.maven.cache.extension.c.n.a (SourceFile:53)`
- `at com.gradle.maven.cache.extension.c.o.a (SourceFile:18)`
- `at com.gradle.maven.cache.extension.c.p.a (SourceFile:27)`
- `at com.gradle.maven.cache.extension.c.q.a (SourceFile:23)`
- `at com.gradle.maven.cache.extension.g.b.lambda$createProxy$0 (SourceFile:77)`
- `at com.gradle.maven.scan.extension.internal.e.b.executeMojo (SourceFile:116)`
- `at java.lang.reflect.Method.invoke (Method.java:565)`
- `at java.lang.reflect.Method.invoke (Method.java:565)`
- `at java.lang.reflect.Method.invoke (Method.java:565)`
- `at jdk.internal.reflect.DirectMethodHandleAccessor.invoke (DirectMethodHandleAccessor.java:104)`
- `at jdk.internal.reflect.DirectMethodHandleAccessor.invoke (DirectMethodHandleAccessor.java:104)`
- `at jdk.internal.reflect.DirectMethodHandleAccessor.invoke (DirectMethodHandleAccessor.java:104)`
- `at jdk.proxy5.$Proxy69.execute (Unknown Source)`
- `at org.apache.commons.exec.DefaultExecutor.execute (DefaultExecutor.java:306)`
- `at org.apache.commons.exec.DefaultExecutor.executeInternal (DefaultExecutor.java:408)`
- `at org.apache.maven.DefaultMaven.doExecute (DefaultMaven.java:173)`
- `at org.apache.maven.DefaultMaven.doExecute (DefaultMaven.java:261)`
- `at org.apache.maven.DefaultMaven.execute (DefaultMaven.java:101)`
- `at org.apache.maven.cli.MavenCli.doMain (MavenCli.java:285)`
- `at org.apache.maven.cli.MavenCli.execute (MavenCli.java:919)`
- `at org.apache.maven.cli.MavenCli.main (MavenCli.java:207)`
- `at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject (LifecycleModuleBuilder.java:105)`
- `at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject (LifecycleModuleBuilder.java:73)`
- `at org.apache.maven.lifecycle.internal.LifecycleStarter.execute (LifecycleStarter.java:118)`
- `at org.apache.maven.lifecycle.internal.MojoExecutor$1.run (MojoExecutor.java:162)`
- `at org.apache.maven.lifecycle.internal.MojoExecutor.access$000 (MojoExecutor.java:75)`
- `at org.apache.maven.lifecycle.internal.MojoExecutor.doExecute (MojoExecutor.java:316)`
- `at org.apache.maven.lifecycle.internal.MojoExecutor.doExecute2 (MojoExecutor.java:328)`
- `at org.apache.maven.lifecycle.internal.MojoExecutor.execute (MojoExecutor.java:159)`
- `at org.apache.maven.lifecycle.internal.MojoExecutor.execute (MojoExecutor.java:174)`
- `at org.apache.maven.lifecycle.internal.MojoExecutor.execute (MojoExecutor.java:212)`
- `at org.apache.maven.lifecycle.internal.builder.singlethreaded.SingleThreadedBuilder.build (SingleThreadedBuilder.java:53)`
- `at org.apache.maven.plugin.DefaultMojosExecutionStrategy.execute (DefaultMojosExecutionStrategy.java:39)`
- `at org.apache.maven.wrapper.BootstrapMainStarter.start (BootstrapMainStarter.java:52)`
- `at org.apache.maven.wrapper.MavenWrapperMain.main (MavenWrapperMain.java:73)`
- `at org.apache.maven.wrapper.WrapperExecutor.execute (WrapperExecutor.java:161)`
- `at org.codehaus.mojo.exec.ExecMojo.execute (ExecMojo.java:464)`
- `at org.codehaus.mojo.exec.ExecMojo.executeCommandLine (ExecMojo.java:873)`
- `at org.codehaus.mojo.exec.ExecMojo.executeCommandLine (ExecMojo.java:913)`
- `at org.codehaus.plexus.classworlds.launcher.Launcher.launch (Launcher.java:201)`
- `at org.codehaus.plexus.classworlds.launcher.Launcher.launchEnhanced (Launcher.java:255)`
- `at org.codehaus.plexus.classworlds.launcher.Launcher.main (Launcher.java:314)`
- `at org.codehaus.plexus.classworlds.launcher.Launcher.mainWithExitCode (Launcher.java:362)`
- `at org.gradle.internal.operations.DefaultBuildOperationRunner$1.execute (SourceFile:27)`
- `at org.gradle.internal.operations.DefaultBuildOperationRunner$1.execute (SourceFile:30)`
- `at org.gradle.internal.operations.DefaultBuildOperationRunner$2.execute (SourceFile:60)`
- `at org.gradle.internal.operations.DefaultBuildOperationRunner$2.execute (SourceFile:67)`
- `at org.gradle.internal.operations.DefaultBuildOperationRunner.execute (SourceFile:167)`
- `at org.gradle.internal.operations.DefaultBuildOperationRunner.execute (SourceFile:60)`
- `at org.gradle.internal.operations.DefaultBuildOperationRunner.run (SourceFile:48)`
- `org.apache.commons.exec.ExecuteException: Process exited with an error: 127 (Exit value: 127)`
</details>
Changed managed dependencies: org.apache.camel:camel-langchain4j-ingest
**Skip-tests mode would test 590 modules** (11 direct + 579 downstream), **skip tests for 20** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (590)</summary>
- `archetypes`
- `camel-a2a`
- `camel-activemq`
- `camel-activemq6`
- `camel-ai-observability`
- `camel-ai-observability-api`
- `camel-ai-parent`
- `camel-ai-resource`
- `camel-ai-tool`
- `camel-alibaba-common`
- `camel-alibaba-eventbridge`
- `camel-alibaba-fc`
- `camel-alibaba-kms`
- `camel-alibaba-mns`
- `camel-alibaba-oss`
- `camel-alibaba-ots`
- `camel-alibaba-parent`
- `camel-alibaba-sls`
- `camel-alibaba-sms`
- `camel-allcomponents`
- `camel-amqp`
- `camel-api`
- `camel-api-component-maven-plugin`
- `camel-arangodb`
- `camel-archetype-api-component`
- `camel-archetype-component`
- `camel-archetype-dataformat`
- `camel-archetype-java`
- `camel-archetype-main`
- `camel-as2`
- `camel-as2-api`
- `camel-as2-parent`
- `camel-asn1`
- `camel-asterisk`
- `camel-atmosphere-websocket`
- `camel-atom`
- `camel-attachments`
- `camel-avro`
- `camel-avro-rpc`
- `camel-avro-rpc-jetty`
- `camel-avro-rpc-parent`
- `camel-avro-rpc-spi`
- `camel-aws-bedrock`
- `camel-aws-cloudtrail`
- `camel-aws-common`
- `camel-aws-config`
- `camel-aws-parameter-store`
- `camel-aws-parent`
- `camel-aws-secrets-manager`
- `camel-aws-security-hub`
- `camel-aws2-athena`
- `camel-aws2-comprehend`
- `camel-aws2-cw`
- `camel-aws2-ddb`
- `camel-aws2-ec2`
- `camel-aws2-ecs`
- `camel-aws2-eks`
- `camel-aws2-eventbridge`
- `camel-aws2-iam`
- `camel-aws2-kinesis`
- `camel-aws2-kms`
- `camel-aws2-lambda`
- `camel-aws2-mq`
- `camel-aws2-msk`
- `camel-aws2-polly`
- `camel-aws2-redshift`
- `camel-aws2-rekognition`
- `camel-aws2-s3`
- `camel-aws2-s3-vectors`
- `camel-aws2-ses`
- `camel-aws2-sns`
- `camel-aws2-sqs`
- `camel-aws2-step-functions`
- `camel-aws2-sts`
- `camel-aws2-textract`
- `camel-aws2-timestream`
- `camel-aws2-transcribe`
- `camel-aws2-translate`
- `camel-azure-common`
- `camel-azure-cosmosdb`
- `camel-azure-eventgrid`
- `camel-azure-eventhubs`
- `camel-azure-files`
- `camel-azure-functions`
- `camel-azure-key-vault`
- `camel-azure-parent`
- `camel-azure-schema-registry`
- `camel-azure-servicebus`
- `camel-azure-storage-blob`
- `camel-azure-storage-datalake`
- `camel-azure-storage-queue`
- `camel-barcode`
- `camel-base`
- `camel-base-engine`
- `camel-base64`
- `camel-bean`
- `camel-bean-validator`
- `camel-beanio`
- `camel-bindy`
- `camel-bom`
- `camel-bonita`
- `camel-box`
- `camel-box-api`
- `camel-box-parent`
- `camel-braintree`
- `camel-browse`
- `camel-caffeine`
- `camel-camunda`
- `camel-cassandraql`
- `camel-catalog`
- `camel-catalog-common`
- `camel-cbor`
- `camel-chatscript`
- `camel-chunk`
- `camel-cli-connector`
- `camel-cli-debug`
- `camel-clickhouse`
- `camel-clickup`
- `camel-cloudevents`
- `camel-cluster`
- `camel-cm-sms`
- `camel-coap`
- `camel-cometd`
- `camel-component-maven-plugin`
- `camel-componentdsl`
- `camel-console`
- `camel-consul`
- `camel-controlbus`
- `camel-core`
- `camel-core-all`
- `camel-core-catalog`
- `camel-core-engine`
- `camel-core-languages`
- `camel-core-model`
- `camel-core-processor`
- `camel-core-reifier`
- `camel-core-xml`
- `camel-couchbase`
- `camel-couchdb`
- `camel-cron`
- `camel-crypto`
- `camel-crypto-pgp`
- `camel-csv`
- `camel-cxf-common`
- `camel-cxf-parent`
- `camel-cxf-rest`
- `camel-cxf-soap`
- `camel-cxf-spring-common`
- `camel-cxf-spring-rest`
- `camel-cxf-spring-soap`
- `camel-cxf-spring-transport`
- `camel-cxf-transport`
- `camel-cyberark-vault`
- `camel-dapr`
- `camel-dataformat`
- `camel-dataset`
- `camel-datasonnet`
- `camel-dataweave`
- `camel-debezium-common`
- `camel-debezium-common-parent`
- `camel-debezium-db2`
- `camel-debezium-maven-plugin`
- `camel-debezium-mongodb`
- `camel-debezium-mysql`
- `camel-debezium-oracle`
- `camel-debezium-parent`
- `camel-debezium-postgres`
- `camel-debezium-sqlserver`
- `camel-debug`
- `camel-dependencies`
- `camel-dfdl`
- `camel-dhis2`
- `camel-dhis2-api`
- `camel-dhis2-parent`
- `camel-diagram`
- `camel-direct`
- `camel-disruptor`
- `camel-djl`
- `camel-dns`
- `camel-docker`
- `camel-docling`
- `camel-drill`
- `camel-dropbox`
- `camel-dsl-modeline`
- `camel-dsl-support`
- `camel-duckdb`
- `camel-dynamic-router`
- `camel-ehcache`
- `camel-eip-documentation-enricher-maven-plugin`
- `camel-elasticsearch`
- `camel-elasticsearch-rest-client`
- `camel-endpointdsl`
- `camel-event`
- `camel-exec`
- `camel-fastjson`
- `camel-fhir`
- `camel-fhir-api`
- `camel-fhir-parent`
- `camel-file`
- `camel-file-watch`
- `camel-flatpack`
- `camel-flink`
- `camel-flowable`
- `camel-fop`
- `camel-fory`
- `camel-freemarker`
- `camel-ftp`
- `camel-ftp-common`
- `camel-geocoder`
- `camel-git`
- `camel-github2`
- `camel-google-bigquery`
- `camel-google-calendar`
- `camel-google-common`
- `camel-google-drive`
- `camel-google-firestore`
- `camel-google-functions`
- `camel-google-mail`
- `camel-google-parent`
- `camel-google-pubsub`
- `camel-google-secret-manager`
- `camel-google-sheets`
- `camel-google-speech-to-text`
- `camel-google-storage`
- `camel-google-text-to-speech`
- `camel-google-vertexai`
- `camel-google-vision`
- `camel-graphql`
- `camel-grok`
- `camel-groovy`
- `camel-grpc`
- `camel-gson`
- `camel-hashicorp-vault`
- `camel-hazelcast`
- `camel-health`
- `camel-hivemq`
- `camel-hl7`
- `camel-http`
- `camel-http-base`
- `camel-http-common`
- `camel-huawei-parent`
- `camel-huaweicloud-common`
- `camel-huaweicloud-dms`
- `camel-huaweicloud-frs`
- `camel-huaweicloud-functiongraph`
- `camel-huaweicloud-iam`
- `camel-huaweicloud-imagerecognition`
- `camel-huaweicloud-obs`
- `camel-huaweicloud-smn`
- `camel-huggingface`
- `camel-ibm-cos`
- `camel-ibm-parent`
- `camel-ibm-secrets-manager`
- `camel-ibm-watson-discovery`
- `camel-ibm-watson-language`
- `camel-ibm-watson-speech-to-text`
- `camel-ibm-watson-text-to-speech`
- `camel-ibm-watsonx-ai`
- `camel-ibm-watsonx-data`
- `camel-ical`
- `camel-iggy`
- `camel-ignite`
- `camel-infinispan`
- `camel-infinispan-common`
- `camel-infinispan-embedded`
- `camel-infinispan-parent`
- `camel-influxdb`
- `camel-influxdb2`
- `camel-iso8583`
- `camel-jackson`
- `camel-jackson-avro`
- `camel-jackson-protobuf`
- `camel-jackson3`
- `camel-jackson3-avro`
- `camel-jackson3-protobuf`
- `camel-jackson3xml`
- `camel-jacksonxml`
- `camel-jactl`
- `camel-jandex`
- `camel-jasypt`
- `camel-java-io`
- `camel-java-joor-dsl`
- `camel-javascript`
- `camel-jaxb`
- `camel-jbang-console`
- `camel-jbang-mcp`
- `camel-jbang-parent`
- `camel-jbang-plugin-mcp`
- `camel-jbang-plugin-route-parser`
- `camel-jbang-plugin-tui`
- `camel-jbang-plugin-validate`
- `camel-jcache`
- `camel-jcr`
- `camel-jdbc`
- `camel-jetty`
- `camel-jetty-common`
- `camel-jfr`
- `camel-jgroups`
- `camel-jgroups-raft`
- `camel-jira`
- `camel-jms`
- `camel-jmx`
- `camel-jolt`
- `camel-jooq`
- `camel-joor`
- `camel-jpa`
- `camel-jq`
- `camel-jsch`
- `camel-jslt`
- `camel-json-validator`
- `camel-jsonapi`
- `camel-jsonata`
- `camel-jsonb`
- `camel-jsonpath`
- `camel-jsoup`
- `camel-jt400`
- `camel-jta`
- `camel-jte`
- `camel-kafka`
- `camel-kamelet`
- `camel-kamelet-main`
- `camel-kamelet-main-support`
- `camel-keycloak`
- `camel-knative`
- `camel-knative-api`
- `camel-knative-http`
- `camel-knative-parent`
- `camel-kserve`
- `camel-kubernetes`
- `camel-kudu`
- `camel-langchain4j-agent`
- `camel-langchain4j-agent-api`
- `camel-langchain4j-chat`
- `camel-langchain4j-core`
- `camel-langchain4j-embeddings`
- `camel-langchain4j-embeddingstore`
- `camel-langchain4j-embeddingstore-api`
- `camel-langchain4j-ingest`
- `camel-langchain4j-tokenizer`
- `camel-langchain4j-web-search`
- `camel-language`
- `camel-launcher-container`
- `camel-ldap`
- `camel-ldif`
- `camel-log`
- `camel-lra`
- `camel-lucene`
- `camel-lumberjack`
- `camel-lzf`
- `camel-mail`
- `camel-mail-microsoft-oauth`
- `camel-main`
- `camel-management`
- `camel-management-api`
- `camel-mapstruct`
- `camel-master`
- `camel-maven-plugin`
- `camel-mcp-server`
- `camel-mcp-server-api`
- `camel-mdc`
- `camel-metrics`
- `camel-micrometer`
- `camel-micrometer-observability`
- `camel-micrometer-prometheus`
- `camel-microprofile-config`
- `camel-microprofile-fault-tolerance`
- `camel-microprofile-health`
- `camel-microprofile-parent`
- `camel-milo`
- `camel-milvus`
- `camel-mina`
- `camel-mina-sftp`
- `camel-minio`
- `camel-mllp`
- `camel-mock`
- `camel-mongodb`
- `camel-mongodb-gridfs`
- `camel-mustache`
- `camel-mvel`
- `camel-mybatis`
- `camel-nats`
- `camel-neo4j`
- `camel-netty`
- `camel-netty-http`
- `camel-oaipmh`
- `camel-oauth`
- `camel-observability-services`
- `camel-observation`
- `camel-ocsf`
- `camel-ognl`
- `camel-olingo2`
- `camel-olingo2-api`
- `camel-olingo2-parent`
- `camel-olingo4`
- `camel-olingo4-api`
- `camel-olingo4-parent`
- `camel-once`
- `camel-opa`
- `camel-openai`
- `camel-openapi-java`
- `camel-openapi-rest-dsl-generator`
- `camel-openapi-validator`
- `camel-opensearch`
- `camel-openstack`
- `camel-opentelemetry`
- `camel-opentelemetry-metrics`
- `camel-opentelemetry2`
- `camel-optaplanner`
- `camel-package-maven-plugin`
- `camel-paho`
- `camel-paho-mqtt5`
- `camel-parquet-avro`
- `camel-pdf`
- `camel-pg-replication-slot`
- `camel-pgevent`
- `camel-pgvector`
- `camel-pinecone`
- `camel-platform-http`
- `camel-platform-http-jolokia`
- `camel-platform-http-main`
- `camel-platform-http-vertx`
- `camel-plc4x`
- `camel-pqc`
- `camel-printer`
- `camel-protobuf`
- `camel-pubnub`
- `camel-pulsar`
- `camel-python`
- `camel-python3`
- `camel-qdrant`
- `camel-quartz`
- `camel-quickfix`
- `camel-quickjs`
- `camel-reactive-streams`
- `camel-reactor`
- `camel-redis`
- `camel-ref`
- `camel-resilience4j`
- `camel-resilience4j-micrometer`
- `camel-resourceresolver-github`
- `camel-rest`
- `camel-rest-openapi`
- `camel-rest-postman`
- `camel-restdsl-openapi-plugin`
- `camel-robotframework`
- `camel-rocketmq`
- `camel-rss`
- `camel-rxjava`
- `camel-saga`
- `camel-salesforce`
- `camel-salesforce-codegen`
- `camel-salesforce-maven-plugin`
- `camel-salesforce-parent`
- `camel-sap-netweaver`
- `camel-saxon`
- `camel-scheduler`
- `camel-schematron`
- `camel-seda`
- `camel-servicenow`
- `camel-servicenow-maven-plugin`
- `camel-servicenow-parent`
- `camel-servlet`
- `camel-shell`
- `camel-shiro`
- `camel-sjms`
- `camel-sjms2`
- `camel-slack`
- `camel-smb`
- `camel-smooks`
- `camel-smpp`
- `camel-snakeyaml`
- `camel-snmp`
- `camel-soap`
- `camel-solr`
- `camel-spiffe`
- `camel-splunk-hec`
- `camel-spring`
- `camel-spring-ai-chat`
- `camel-spring-ai-embeddings`
- `camel-spring-ai-image`
- `camel-spring-ai-parent`
- `camel-spring-ai-vector-store`
- `camel-spring-batch`
- `camel-spring-cloud-config`
- `camel-spring-jdbc`
- `camel-spring-ldap`
- `camel-spring-main`
- `camel-spring-parent`
- `camel-spring-rabbitmq`
- `camel-spring-redis`
- `camel-spring-security`
- `camel-spring-ws`
- `camel-spring-xml`
- `camel-sql`
- `camel-ssh`
- `camel-state-store`
- `camel-state-store-parent`
- `camel-stax`
- `camel-stitch`
- `camel-stream`
- `camel-streamcaching-test`
- `camel-stringtemplate`
- `camel-stripe`
- `camel-stub`
- `camel-support`
- `camel-swift`
- `camel-syslog`
- `camel-tahu`
- `camel-tarfile`
- `camel-telegram`
- `camel-telemetry`
- `camel-telemetry-dev`
- `camel-tensorflow-serving`
- `camel-test-infra-all`
- `camel-test-infra-artemis`
- `camel-test-infra-cli`
- `camel-test-infra-core`
- `camel-test-infra-jetty`
- `camel-test-infra-qdrant`
- `camel-test-infra-smb`
- `camel-test-junit5`
- `camel-test-junit6`
- `camel-test-main-junit5`
- `camel-test-main-junit6`
- `camel-test-parent`
- `camel-test-spring-junit5`
- `camel-test-spring-junit6`
- `camel-thrift`
- `camel-thymeleaf`
- `camel-tika`
- `camel-timer`
- `camel-tooling-maven`
- `camel-toon`
- `camel-tracing`
- `camel-twilio`
- `camel-twitter`
- `camel-ubl`
- `camel-undertow`
- `camel-undertow-spring-security`
- `camel-univocity-parsers`
- `camel-util`
- `camel-validator`
- `camel-velocity`
- `camel-vertx`
- `camel-vertx-common`
- `camel-vertx-http`
- `camel-vertx-parent`
- `camel-vertx-websocket`
- `camel-wal`
- `camel-wasm`
- `camel-weather`
- `camel-weaviate`
- `camel-web3j`
- `camel-webhook`
- `camel-whatsapp`
- `camel-wordpress`
- `camel-workday`
- `camel-xchange`
- `camel-xj`
- `camel-xml-io`
- `camel-xml-io-dsl`
- `camel-xml-io-util`
- `camel-xml-jaxb`
- `camel-xml-jaxb-dsl`
- `camel-xml-jaxb-dsl-test`
- `camel-xml-jaxb-dsl-test-definition`
- `camel-xml-jaxb-dsl-test-spring`
- `camel-xml-jaxp`
- `camel-xml-jaxp-util`
- `camel-xmlsecurity`
- `camel-xmpp`
- `camel-xpath`
- `camel-xslt`
- `camel-xslt-saxon`
- `camel-yaml-dsl-common`
- `camel-yaml-dsl-parent`
- `camel-yaml-dsl-validator`
- `camel-yaml-dsl-validator-maven-plugin`
- `camel-yaml-io`
- `camel-zendesk`
- `camel-zip-deflater`
- `camel-zipfile`
- `camel-zookeeper`
- `camel-zookeeper-master`
- `components`
- `core`
- `coverage`
- `docs`
- `dsl`
- `sync-properties-maven-plugin`
</details>
<details><summary>Modules with tests skipped (20)</summary>
- `apache-camel`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `dummy-component`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
:warning: **Some tests are disabled on GitHub Actions** (`@DisabledIfSystemProperty(named = "ci.env.name")`) and require manual verification:
- `components/camel-ai`: 27 test(s) disabled on GitHub Actions
<details><summary>Build reactor — dependencies compiled but only changed modules were tested (13 modules, 2m 49s total)</summary>
**Total reactor time:** 2m 49s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Component DSL | 48.8s | SUCCESS |
| Camel :: AI :: LangChain4j :: Ingest | 32.9s | SUCCESS |
| Camel :: Main | 30.4s | SUCCESS |
| Camel :: Docs | 15.2s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 13.9s | SUCCESS |
| Camel :: Maven Plugins :: Camel Maven Package | 10.8s | SUCCESS |
| Camel :: Kamelet Main | 7.9s | SUCCESS |
| Camel :: All Components Sync point | 4.7s | SUCCESS |
| Camel :: Coverage | 1.6s | SUCCESS |
| Camel :: Parent | 1.2s | SUCCESS |
| Camel :: AI :: Parent | 1.0s | SUCCESS |
| Camel :: BOM | 0.2s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
**Top 20 slowest modules:**
- `Camel :: Component DSL` (48.8s)
- `Camel :: AI :: LangChain4j :: Ingest` (32.9s)
- `Camel :: Main` (30.4s)
- `Camel :: Docs` (15.2s)
- `Camel :: Catalog :: Camel Catalog` (13.9s)
- `Camel :: Maven Plugins :: Camel Maven Package` (10.8s)
- `Camel :: Kamelet Main` (7.9s)
- `Camel :: All Components Sync point` (4.7s)
- `Camel :: Coverage` (1.6s)
- `Camel :: Parent` (1.2s)
- `Camel :: AI :: Parent` (1.0s)
- `Camel :: BOM` (0.2s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/34206069150)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24621: camel-langchain4j-ingest - new document ingestion component [camel]
Posted by "JiriOndrusek (via GitHub)" <gi...@apache.org>.
JiriOndrusek commented on code in PR #26190:
URL: https://github.com/apache/camel/pull/26190#discussion_r3957404568
##########
components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.langchain4j.ingest;
+
+import java.util.Set;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.service.ServiceHelper;
+
+/**
+ * Splits the message body into segments, embeds them in batches and writes them to the embedding store; the message
+ * body is replaced with the {@link IngestResult}.
+ *
+ * <p>
+ * With an {@code idempotentRepository} configured, the producer claims the document id before writing: a duplicate
+ * delivery is answered {@code SKIPPED} without touching the store, a blank document releases its claim so a later,
+ * populated delivery under the same id still ingests, and a failed write releases it so the delivery can be retried.
+ * The eager claim has a documented consequence: a duplicate racing an in-flight first delivery is answered
+ * {@code SKIPPED} even if that delivery then fails and releases the id — with an at-least-once source the skipped
+ * duplicate is acknowledged, so the failed original must be redelivered by its own source or the document is in neither
+ * the store nor a queue. Claim-on-success semantics are the eventual alternative for sources that cannot redeliver.
+ */
+public class LangChain4jIngestProducer extends DefaultProducer {
+
+ private final LangChain4jIngestEndpoint endpoint;
+ private final LangChain4jIngestConfiguration configuration;
+ private IngestService service;
+
+ public LangChain4jIngestProducer(LangChain4jIngestEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ String pipeline = endpoint.getPipelineName();
+ if (pipeline == null || pipeline.isBlank()) {
+ throw new IllegalArgumentException(
+ "The ingestion pipeline name is missing. Use langchain4j-ingest:pipelineName in the endpoint URI.");
+ }
+ if (configuration.getDocumentIdHeader() == null || configuration.getDocumentIdHeader().isBlank()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': documentIdHeader must not be blank. Omit the option for"
+ + " the default, or name the header carrying the document id.");
+ }
+ if (configuration.getMaxSegmentSize() <= 0 || configuration.getMaxOverlapSize() < 0
+ || configuration.getMaxOverlapSize() >= configuration.getMaxSegmentSize()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxSegmentSize must be positive and maxOverlapSize must"
+ + " be non-negative and smaller than it (got "
+ + configuration.getMaxSegmentSize()
+ + " / " + configuration.getMaxOverlapSize() + ")");
+ }
Review Comment:
Good catch — the docs and the check genuinely contradicted each other. Applied your suggestion verbatim: the bounds validation now only runs when no `documentSplitter` is configured, and a new test (`customSplitterIgnoresExplicitSizeBounds`) proves an endpoint with a custom splitter and explicitly nonsensical `maxSegmentSize=0&maxOverlapSize=-1` starts and ingests fine.
_Claude Code on behalf of @JiriOndrusek_
##########
components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.langchain4j.ingest;
+
+import java.util.Set;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.service.ServiceHelper;
+
+/**
+ * Splits the message body into segments, embeds them in batches and writes them to the embedding store; the message
+ * body is replaced with the {@link IngestResult}.
+ *
+ * <p>
+ * With an {@code idempotentRepository} configured, the producer claims the document id before writing: a duplicate
+ * delivery is answered {@code SKIPPED} without touching the store, a blank document releases its claim so a later,
+ * populated delivery under the same id still ingests, and a failed write releases it so the delivery can be retried.
+ * The eager claim has a documented consequence: a duplicate racing an in-flight first delivery is answered
+ * {@code SKIPPED} even if that delivery then fails and releases the id — with an at-least-once source the skipped
+ * duplicate is acknowledged, so the failed original must be redelivered by its own source or the document is in neither
+ * the store nor a queue. Claim-on-success semantics are the eventual alternative for sources that cannot redeliver.
+ */
+public class LangChain4jIngestProducer extends DefaultProducer {
+
+ private final LangChain4jIngestEndpoint endpoint;
+ private final LangChain4jIngestConfiguration configuration;
+ private IngestService service;
+
+ public LangChain4jIngestProducer(LangChain4jIngestEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ String pipeline = endpoint.getPipelineName();
+ if (pipeline == null || pipeline.isBlank()) {
+ throw new IllegalArgumentException(
+ "The ingestion pipeline name is missing. Use langchain4j-ingest:pipelineName in the endpoint URI.");
+ }
+ if (configuration.getDocumentIdHeader() == null || configuration.getDocumentIdHeader().isBlank()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': documentIdHeader must not be blank. Omit the option for"
+ + " the default, or name the header carrying the document id.");
+ }
+ if (configuration.getMaxSegmentSize() <= 0 || configuration.getMaxOverlapSize() < 0
+ || configuration.getMaxOverlapSize() >= configuration.getMaxSegmentSize()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxSegmentSize must be positive and maxOverlapSize must"
+ + " be non-negative and smaller than it (got "
+ + configuration.getMaxSegmentSize()
+ + " / " + configuration.getMaxOverlapSize() + ")");
+ }
+
+ if (configuration.getEmbeddingBatchSize() < 1) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': embeddingBatchSize must be positive (got "
+ + configuration.getEmbeddingBatchSize() + ")");
+ }
+
+ if (configuration.getMaxDocumentSize() < 0) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxDocumentSize must not be negative (got "
+ + configuration.getMaxDocumentSize() + ")");
+ }
+
+ EmbeddingStore<TextSegment> store
+ = resolve(EmbeddingStore.class, configuration.getEmbeddingStore(), "embedding store", "embeddingStore");
+ EmbeddingModel model
+ = resolve(EmbeddingModel.class, configuration.getEmbeddingModel(), "embedding model", "embeddingModel");
+ service = configuration.getDocumentSplitter() != null
+ ? new IngestService(
+ pipeline, store, model, configuration.getDocumentSplitter(),
+ configuration.getEmbeddingBatchSize(), configuration.getMaxDocumentSize())
+ : new IngestService(
+ pipeline, store, model, configuration.getMaxSegmentSize(),
+ configuration.getMaxOverlapSize(), configuration.getEmbeddingBatchSize(),
+ configuration.getMaxDocumentSize());
+
+ IdempotentRepository repository = configuration.getIdempotentRepository();
+ if (repository != null) {
+ // a repository bound via configuration does not pass through the registry's bind
+ // hook, so a CamelContextAware implementation would otherwise run contextless. It is
+ // started but never stopped here: the bean may be shared, and stopping the in-memory
+ // repository would clear it
+ CamelContextAware.trySetCamelContext(repository, getEndpoint().getCamelContext());
+ ServiceHelper.startService(repository);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ String documentId = resolveDocumentId(exchange);
+
+ IdempotentRepository repository = configuration.getIdempotentRepository();
+ if (repository == null) {
+ exchange.getMessage().setBody(service.ingest(documentId, exchange.getMessage().getBody(String.class)));
+ return;
+ }
+
+ // the claim is eager: a duplicate racing an in-flight first delivery is answered SKIPPED
+ // even if that delivery then fails and releases the id. The claim also precedes the body
+ // read, so a skipped duplicate never pays for materialising a large payload. The
+ // Exchange-aware repository overloads are used throughout, matching the idempotent
+ // consumer EIP - a repository overriding only those variants is not bypassed
+ if (!repository.add(exchange, documentId)) {
+ exchange.getMessage().setBody(
+ new IngestResult(service.pipeline(), documentId, 0, IngestResult.Outcome.SKIPPED));
+ return;
+ }
+ IngestResult result;
+ try {
+ result = service.ingest(documentId, exchange.getMessage().getBody(String.class));
+ } catch (Exception e) {
+ // a failed write must not keep the claim, or the delivery could never be retried.
+ // An Error (an OutOfMemoryError, say) is deliberately not caught: under a VM-level
+ // failure the release itself could not be trusted, so the id stays claimed and later
+ // deliveries are answered SKIPPED - clear it from the repository to re-ingest
+ try {
+ repository.remove(exchange, documentId);
+ } catch (Exception rollback) {
+ // the release often fails from the same root cause; the original failure is
+ // the one worth reporting
+ e.addSuppressed(rollback);
+ }
+ throw e;
+ }
+ if (result.outcome() == IngestResult.Outcome.EMPTY) {
+ // a blank document wrote nothing, so it must not keep the claim - a later, populated
+ // delivery under the same id would be answered SKIPPED
+ repository.remove(exchange, documentId);
+ } else {
+ repository.confirm(exchange, documentId);
+ }
+ exchange.getMessage().setBody(result);
+ }
+
+ /**
+ * The exchange property wins over the header: a route that parses documents captures the id into the property
+ * before the parse, and a parser (Tika) copies document metadata over the headers, so a header read here could be
+ * spoofed by the document itself. A property that is present but blank fails the exchange on purpose, without
+ * falling back to the header: the route deliberately captured the id, so a blank capture is a broken expression to
+ * surface loudly, not a case to paper over with a value of weaker provenance.
+ */
+ private String resolveDocumentId(Exchange exchange) {
+ String documentId = exchange.getProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, String.class);
+ if (documentId == null) {
+ documentId = exchange.getMessage().getHeader(configuration.getDocumentIdHeader(), String.class);
+ }
+ if (documentId == null || documentId.isBlank()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + endpoint.getPipelineName() + "': no document id. Set the "
+ + configuration.getDocumentIdHeader()
+ + " header (or the " + LangChain4jIngest.DOCUMENT_ID_PROPERTY
+ + " exchange property), or point the documentIdHeader endpoint option"
+ + " at where the consumer puts it.");
+ }
+ return documentId;
+ }
+
+ /**
+ * Resolves a configured bean, or the single registry bean of the type. Picking one of several silently would bind
+ * the pipeline to whichever bean happened to be found first, so zero and several candidates each fail with the fix
+ * in the message. Runs after autowiring: with exactly one candidate the configuration already carries it.
+ */
+ @SuppressWarnings("unchecked")
+ private <T> T resolve(Class<?> type, T configured, String what, String option) {
Review Comment:
Went one step further than `Class<T>`: the signature is now `resolve(Class<? super T> type, T configured, ...)`. Because the parameterized store type is a subtype of its raw class, `EmbeddingStore.class` satisfies the bound with no cast at either call site — and a swapped class literal fails to compile at **both** sites (verified: planting `EmbeddingModel.class` at the store call site is rejected with "inference variable T has incompatible bounds"). The single remaining cast sits inside the method, where the registry's contract guarantees it, with a comment saying so.
_Claude Code on behalf of @JiriOndrusek_
##########
components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.langchain4j.ingest;
+
+import java.util.Set;
+
+import dev.langchain4j.data.segment.TextSegment;
+import dev.langchain4j.model.embedding.EmbeddingModel;
+import dev.langchain4j.store.embedding.EmbeddingStore;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.service.ServiceHelper;
+
+/**
+ * Splits the message body into segments, embeds them in batches and writes them to the embedding store; the message
+ * body is replaced with the {@link IngestResult}.
+ *
+ * <p>
+ * With an {@code idempotentRepository} configured, the producer claims the document id before writing: a duplicate
+ * delivery is answered {@code SKIPPED} without touching the store, a blank document releases its claim so a later,
+ * populated delivery under the same id still ingests, and a failed write releases it so the delivery can be retried.
+ * The eager claim has a documented consequence: a duplicate racing an in-flight first delivery is answered
+ * {@code SKIPPED} even if that delivery then fails and releases the id — with an at-least-once source the skipped
+ * duplicate is acknowledged, so the failed original must be redelivered by its own source or the document is in neither
+ * the store nor a queue. Claim-on-success semantics are the eventual alternative for sources that cannot redeliver.
+ */
+public class LangChain4jIngestProducer extends DefaultProducer {
+
+ private final LangChain4jIngestEndpoint endpoint;
+ private final LangChain4jIngestConfiguration configuration;
+ private IngestService service;
+
+ public LangChain4jIngestProducer(LangChain4jIngestEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ String pipeline = endpoint.getPipelineName();
+ if (pipeline == null || pipeline.isBlank()) {
+ throw new IllegalArgumentException(
+ "The ingestion pipeline name is missing. Use langchain4j-ingest:pipelineName in the endpoint URI.");
+ }
+ if (configuration.getDocumentIdHeader() == null || configuration.getDocumentIdHeader().isBlank()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': documentIdHeader must not be blank. Omit the option for"
+ + " the default, or name the header carrying the document id.");
+ }
+ if (configuration.getMaxSegmentSize() <= 0 || configuration.getMaxOverlapSize() < 0
+ || configuration.getMaxOverlapSize() >= configuration.getMaxSegmentSize()) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxSegmentSize must be positive and maxOverlapSize must"
+ + " be non-negative and smaller than it (got "
+ + configuration.getMaxSegmentSize()
+ + " / " + configuration.getMaxOverlapSize() + ")");
+ }
+
+ if (configuration.getEmbeddingBatchSize() < 1) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': embeddingBatchSize must be positive (got "
+ + configuration.getEmbeddingBatchSize() + ")");
+ }
+
+ if (configuration.getMaxDocumentSize() < 0) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': maxDocumentSize must not be negative (got "
+ + configuration.getMaxDocumentSize() + ")");
+ }
+
+ EmbeddingStore<TextSegment> store
+ = resolve(EmbeddingStore.class, configuration.getEmbeddingStore(), "embedding store", "embeddingStore");
+ EmbeddingModel model
+ = resolve(EmbeddingModel.class, configuration.getEmbeddingModel(), "embedding model", "embeddingModel");
+ service = configuration.getDocumentSplitter() != null
+ ? new IngestService(
+ pipeline, store, model, configuration.getDocumentSplitter(),
+ configuration.getEmbeddingBatchSize(), configuration.getMaxDocumentSize())
+ : new IngestService(
+ pipeline, store, model, configuration.getMaxSegmentSize(),
+ configuration.getMaxOverlapSize(), configuration.getEmbeddingBatchSize(),
+ configuration.getMaxDocumentSize());
+
+ IdempotentRepository repository = configuration.getIdempotentRepository();
+ if (repository != null) {
+ // a repository bound via configuration does not pass through the registry's bind
+ // hook, so a CamelContextAware implementation would otherwise run contextless. It is
+ // started but never stopped here: the bean may be shared, and stopping the in-memory
+ // repository would clear it
+ CamelContextAware.trySetCamelContext(repository, getEndpoint().getCamelContext());
+ ServiceHelper.startService(repository);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ String documentId = resolveDocumentId(exchange);
+
+ IdempotentRepository repository = configuration.getIdempotentRepository();
+ if (repository == null) {
+ exchange.getMessage().setBody(service.ingest(documentId, exchange.getMessage().getBody(String.class)));
+ return;
+ }
+
+ // the claim is eager: a duplicate racing an in-flight first delivery is answered SKIPPED
+ // even if that delivery then fails and releases the id. The claim also precedes the body
+ // read, so a skipped duplicate never pays for materialising a large payload. The
+ // Exchange-aware repository overloads are used throughout, matching the idempotent
+ // consumer EIP - a repository overriding only those variants is not bypassed
+ if (!repository.add(exchange, documentId)) {
+ exchange.getMessage().setBody(
+ new IngestResult(service.pipeline(), documentId, 0, IngestResult.Outcome.SKIPPED));
+ return;
+ }
+ IngestResult result;
+ try {
+ result = service.ingest(documentId, exchange.getMessage().getBody(String.class));
+ } catch (Exception e) {
+ // a failed write must not keep the claim, or the delivery could never be retried.
+ // An Error (an OutOfMemoryError, say) is deliberately not caught: under a VM-level
+ // failure the release itself could not be trusted, so the id stays claimed and later
+ // deliveries are answered SKIPPED - clear it from the repository to re-ingest
+ try {
+ repository.remove(exchange, documentId);
+ } catch (Exception rollback) {
+ // the release often fails from the same root cause; the original failure is
+ // the one worth reporting
+ e.addSuppressed(rollback);
+ }
+ throw e;
+ }
Review Comment:
Thanks for tracing that one through — exactly right: both catches re-throw, the nested one exists purely so a repository failure during claim release cannot mask the root cause, and `Error` staying uncaught (claim stays poisoned → later deliveries see SKIPPED) is the documented at-least-once trade-off.
_Claude Code on behalf of @JiriOndrusek_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org