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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
