jamesnetherton commented on code in PR #9143: URL: https://github.com/apache/camel-quarkus/pull/9143#discussion_r3989993267
########## extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java: ########## @@ -0,0 +1,390 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.langchain4j.ingest; + +import java.io.InputStream; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Component; +import org.apache.camel.Exchange; +import org.apache.camel.Expression; +import org.apache.camel.Processor; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.langchain4j.ingest.IngestResult; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders; +import org.apache.camel.component.langchain4j.ingest.TikaTextDecode; +import org.apache.camel.model.ProcessorDefinition; +import org.apache.camel.spi.IdempotentRepository; +import org.apache.camel.support.builder.ExpressionBuilder; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.apache.camel.util.StringHelper; +import org.apache.camel.util.URISupport; + +/** + * Generates one Camel route per {@link IngestPipelineDefinition}: consume, resolve the document id, optionally parse, + * then split, embed and store through the {@code langchain4j-ingest} producer. + * + * <p> + * A directory pipeline watches its folder with the safe file-consumer defaults — documents are left in place, unchanged + * files are remembered in a duplicate register keyed on path, modification time and size, and a file still being copied + * in is waited for. A consumer pipeline reads any component and deduplicates by document id when a repository is + * configured. The id is always captured into an exchange property <em>before</em> the parse stage: a parser copies + * document metadata over the headers, so a crafted document could otherwise forge its own identity. + * + * <p> + * An internal copy of the pipeline assembly proposed alongside the upstream component: the delegation to Apache Camel + * is engine-only, so the topology lives here, package-private — replaceable by an upstream artifact or kamelets if the + * community adopts one of them. + * + * <p> + * The class is abstract on purpose: camel-quarkus routes discovery instantiates every concrete public + * {@code RouteBuilder} it finds reflectively, and an abstract base is skipped by construction. + */ +abstract class IngestPipelineRouteBuilder extends RouteBuilder { + + /** + * The built-in register capacity, sized above Camel's 1000-entry default so eviction does not re-ingest large + * directories during normal operation; in-memory, so lost on restart. + */ + static final int DEFAULT_REGISTER_CAPACITY = 100_000; + + /** + * The pipelines to build routes for; supplied by the subclass assembling its definitions from another source — a + * configuration model, say. + */ + protected abstract List<IngestPipelineDefinition> pipelines(); + + @Override + public void configure() { + Set<String> names = new HashSet<>(); + for (IngestPipelineDefinition pipeline : pipelines()) { + if (!names.add(pipeline.name())) { + throw new IllegalArgumentException( + "Ingestion pipeline '" + pipeline.name() + "' is defined twice. Use one name per pipeline."); + } + configurePipeline(pipeline); + } + } + + private void configurePipeline(IngestPipelineDefinition pipeline) { + requireParserComponent(pipeline); + String storeRef = bindInstance(pipeline.embeddingStore(), pipeline.embeddingStoreRef(), pipeline, "store"); + String modelRef = bindInstance(pipeline.embeddingModel(), pipeline.embeddingModelRef(), pipeline, "model"); + String splitterRef = pipeline.documentSplitterRef(); + + if (pipeline.directory() != null) { + directoryRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source=file:{}", pipeline.name(), pipeline.directory()); + } else { + consumerRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source={}", pipeline.name(), URISupport.sanitizeUri(pipeline.uri())); + } + } + + // ******************************************************************************** + // The two route topologies + // ******************************************************************************** + + /** + * The route: watch the directory → resolve the id → (parse) → split, embed, store. The register in the file + * endpoint keeps unchanged files from re-ingesting; an edited file gets a new key and re-ingests, its old segments + * remain. + */ + private void directoryRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String registerRef = repositoryRef(pipeline, true); + Expression documentId = documentIdExpression(pipeline.documentId(), Exchange.FILE_NAME); + + ProcessorDefinition<?> route = from(fileEndpointUri(pipeline, registerRef)) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); + route = parseSteps(route, pipeline); + // the file consumer discards the reply, so no repository is passed to the producer: the + // endpoint register already keeps the same file version from being ingested twice + ProcessorDefinition<?> tail = route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef, null, null)); + if (pipeline.parser() != null) { + // the discarded reply would otherwise hide it: a parse to nothing typically means a + // missing Tika parser module or an image-only document, and the file's register key + // is committed, so it is not retried until the file changes + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.warn("Ingestion pipeline '{}': document '{}' parsed to no text and was skipped; its key is" + + " committed, so it is not retried until the file changes (missing parser module?" + + " image-only document?)", + pipeline.name(), result.documentId()); + } + }); + } else { + // the discarded reply would otherwise hide even the trace of a blank file + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.debug("Ingestion pipeline '{}': document '{}' contained no text, nothing was written", + pipeline.name(), result.documentId()); + } + }); + } + } + + /** The route: consume → resolve the id → (parse) → split, embed, store; the reply is the result. */ + private void consumerRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String scheme = StringHelper.before(pipeline.uri(), ":"); + requireComponent(scheme, pipeline, "its source"); + String registerRef = repositoryRef(pipeline, false); + Expression documentId = documentIdExpression(pipeline.documentId(), LangChain4jIngestHeaders.DOCUMENT_ID); + + ProcessorDefinition<?> route = from(pipeline.uri()) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); Review Comment: (non-blocking) If `documentId` evaluates to null here, `setProperty` removes the property instead of setting it. The producer's `resolveDocumentId` then falls back to the id header, but by then Tika has copied document metadata into headers. Example: a consumer pipeline with `parser=tika` gets a delivery with no id header (a keyless Kafka record with `documentId=CamelKafkaKey`, or a `direct:` call without `CamelIngestDocumentId`). If the payload contains `<meta name="CamelIngestDocumentId" content="victim">`, it is ingested under `victim`. The previous engine failed that delivery. The spoofing test only covers the case where the header is present. Directory pipelines whose `document-id` resolves to null, and simple-expression pipelines that fall back to `CamelLangChain4jIngestDocumentId`, work the same way. Could the route fail the exchange before the parse when the captured id is null or blank? ########## extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java: ########## @@ -0,0 +1,390 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.langchain4j.ingest; + +import java.io.InputStream; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Component; +import org.apache.camel.Exchange; +import org.apache.camel.Expression; +import org.apache.camel.Processor; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.langchain4j.ingest.IngestResult; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders; +import org.apache.camel.component.langchain4j.ingest.TikaTextDecode; +import org.apache.camel.model.ProcessorDefinition; +import org.apache.camel.spi.IdempotentRepository; +import org.apache.camel.support.builder.ExpressionBuilder; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.apache.camel.util.StringHelper; +import org.apache.camel.util.URISupport; + +/** + * Generates one Camel route per {@link IngestPipelineDefinition}: consume, resolve the document id, optionally parse, + * then split, embed and store through the {@code langchain4j-ingest} producer. + * + * <p> + * A directory pipeline watches its folder with the safe file-consumer defaults — documents are left in place, unchanged + * files are remembered in a duplicate register keyed on path, modification time and size, and a file still being copied + * in is waited for. A consumer pipeline reads any component and deduplicates by document id when a repository is + * configured. The id is always captured into an exchange property <em>before</em> the parse stage: a parser copies + * document metadata over the headers, so a crafted document could otherwise forge its own identity. + * + * <p> + * An internal copy of the pipeline assembly proposed alongside the upstream component: the delegation to Apache Camel + * is engine-only, so the topology lives here, package-private — replaceable by an upstream artifact or kamelets if the + * community adopts one of them. + * + * <p> + * The class is abstract on purpose: camel-quarkus routes discovery instantiates every concrete public + * {@code RouteBuilder} it finds reflectively, and an abstract base is skipped by construction. + */ +abstract class IngestPipelineRouteBuilder extends RouteBuilder { + + /** + * The built-in register capacity, sized above Camel's 1000-entry default so eviction does not re-ingest large + * directories during normal operation; in-memory, so lost on restart. + */ + static final int DEFAULT_REGISTER_CAPACITY = 100_000; + + /** + * The pipelines to build routes for; supplied by the subclass assembling its definitions from another source — a + * configuration model, say. + */ + protected abstract List<IngestPipelineDefinition> pipelines(); + + @Override + public void configure() { + Set<String> names = new HashSet<>(); + for (IngestPipelineDefinition pipeline : pipelines()) { + if (!names.add(pipeline.name())) { + throw new IllegalArgumentException( + "Ingestion pipeline '" + pipeline.name() + "' is defined twice. Use one name per pipeline."); + } + configurePipeline(pipeline); + } + } + + private void configurePipeline(IngestPipelineDefinition pipeline) { + requireParserComponent(pipeline); + String storeRef = bindInstance(pipeline.embeddingStore(), pipeline.embeddingStoreRef(), pipeline, "store"); + String modelRef = bindInstance(pipeline.embeddingModel(), pipeline.embeddingModelRef(), pipeline, "model"); + String splitterRef = pipeline.documentSplitterRef(); + + if (pipeline.directory() != null) { + directoryRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source=file:{}", pipeline.name(), pipeline.directory()); + } else { + consumerRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source={}", pipeline.name(), URISupport.sanitizeUri(pipeline.uri())); + } + } + + // ******************************************************************************** + // The two route topologies + // ******************************************************************************** + + /** + * The route: watch the directory → resolve the id → (parse) → split, embed, store. The register in the file + * endpoint keeps unchanged files from re-ingesting; an edited file gets a new key and re-ingests, its old segments + * remain. + */ + private void directoryRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String registerRef = repositoryRef(pipeline, true); + Expression documentId = documentIdExpression(pipeline.documentId(), Exchange.FILE_NAME); + + ProcessorDefinition<?> route = from(fileEndpointUri(pipeline, registerRef)) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); + route = parseSteps(route, pipeline); + // the file consumer discards the reply, so no repository is passed to the producer: the + // endpoint register already keeps the same file version from being ingested twice + ProcessorDefinition<?> tail = route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef, null, null)); + if (pipeline.parser() != null) { + // the discarded reply would otherwise hide it: a parse to nothing typically means a + // missing Tika parser module or an image-only document, and the file's register key + // is committed, so it is not retried until the file changes + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.warn("Ingestion pipeline '{}': document '{}' parsed to no text and was skipped; its key is" + + " committed, so it is not retried until the file changes (missing parser module?" + + " image-only document?)", + pipeline.name(), result.documentId()); + } + }); + } else { + // the discarded reply would otherwise hide even the trace of a blank file + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.debug("Ingestion pipeline '{}': document '{}' contained no text, nothing was written", + pipeline.name(), result.documentId()); + } + }); + } + } + + /** The route: consume → resolve the id → (parse) → split, embed, store; the reply is the result. */ + private void consumerRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String scheme = StringHelper.before(pipeline.uri(), ":"); + requireComponent(scheme, pipeline, "its source"); + String registerRef = repositoryRef(pipeline, false); + Expression documentId = documentIdExpression(pipeline.documentId(), LangChain4jIngestHeaders.DOCUMENT_ID); + + ProcessorDefinition<?> route = from(pipeline.uri()) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); + route = parseSteps(route, pipeline); Review Comment: (non-blocking) On consumer pipelines with `idempotent-repository`, the Tika or Docling parse runs before the `langchain4j-ingest` producer claims the id. So every duplicate is parsed in full, including a remote Docling Serve conversion with OCR, only to be answered `SKIPPED`. That goes against the producer's note that a skipped duplicate never pays for materialising a large payload. Example: an `aws2-s3` source with `deleteAfterRead(false)`, a register and `parser=docling` re-lists every object on each poll, so every object is downloaded and sent to docling-serve on every poll. Could the id be checked or claimed before `parseSteps`? ########## extensions/langchain4j-ingest/runtime/src/main/java/org/apache/camel/quarkus/component/langchain4j/ingest/IngestPipelineRouteBuilder.java: ########## @@ -0,0 +1,390 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.langchain4j.ingest; + +import java.io.InputStream; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Component; +import org.apache.camel.Exchange; +import org.apache.camel.Expression; +import org.apache.camel.Processor; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.langchain4j.ingest.IngestResult; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngest; +import org.apache.camel.component.langchain4j.ingest.LangChain4jIngestHeaders; +import org.apache.camel.component.langchain4j.ingest.TikaTextDecode; +import org.apache.camel.model.ProcessorDefinition; +import org.apache.camel.spi.IdempotentRepository; +import org.apache.camel.support.builder.ExpressionBuilder; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.apache.camel.util.StringHelper; +import org.apache.camel.util.URISupport; + +/** + * Generates one Camel route per {@link IngestPipelineDefinition}: consume, resolve the document id, optionally parse, + * then split, embed and store through the {@code langchain4j-ingest} producer. + * + * <p> + * A directory pipeline watches its folder with the safe file-consumer defaults — documents are left in place, unchanged + * files are remembered in a duplicate register keyed on path, modification time and size, and a file still being copied + * in is waited for. A consumer pipeline reads any component and deduplicates by document id when a repository is + * configured. The id is always captured into an exchange property <em>before</em> the parse stage: a parser copies + * document metadata over the headers, so a crafted document could otherwise forge its own identity. + * + * <p> + * An internal copy of the pipeline assembly proposed alongside the upstream component: the delegation to Apache Camel + * is engine-only, so the topology lives here, package-private — replaceable by an upstream artifact or kamelets if the + * community adopts one of them. + * + * <p> + * The class is abstract on purpose: camel-quarkus routes discovery instantiates every concrete public + * {@code RouteBuilder} it finds reflectively, and an abstract base is skipped by construction. + */ +abstract class IngestPipelineRouteBuilder extends RouteBuilder { + + /** + * The built-in register capacity, sized above Camel's 1000-entry default so eviction does not re-ingest large + * directories during normal operation; in-memory, so lost on restart. + */ + static final int DEFAULT_REGISTER_CAPACITY = 100_000; + + /** + * The pipelines to build routes for; supplied by the subclass assembling its definitions from another source — a + * configuration model, say. + */ + protected abstract List<IngestPipelineDefinition> pipelines(); + + @Override + public void configure() { + Set<String> names = new HashSet<>(); + for (IngestPipelineDefinition pipeline : pipelines()) { + if (!names.add(pipeline.name())) { + throw new IllegalArgumentException( + "Ingestion pipeline '" + pipeline.name() + "' is defined twice. Use one name per pipeline."); + } + configurePipeline(pipeline); + } + } + + private void configurePipeline(IngestPipelineDefinition pipeline) { + requireParserComponent(pipeline); + String storeRef = bindInstance(pipeline.embeddingStore(), pipeline.embeddingStoreRef(), pipeline, "store"); + String modelRef = bindInstance(pipeline.embeddingModel(), pipeline.embeddingModelRef(), pipeline, "model"); + String splitterRef = pipeline.documentSplitterRef(); + + if (pipeline.directory() != null) { + directoryRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source=file:{}", pipeline.name(), pipeline.directory()); + } else { + consumerRoute(pipeline, storeRef, modelRef, splitterRef); + log.info("Ingestion pipeline '{}': source={}", pipeline.name(), URISupport.sanitizeUri(pipeline.uri())); + } + } + + // ******************************************************************************** + // The two route topologies + // ******************************************************************************** + + /** + * The route: watch the directory → resolve the id → (parse) → split, embed, store. The register in the file + * endpoint keeps unchanged files from re-ingesting; an edited file gets a new key and re-ingests, its old segments + * remain. + */ + private void directoryRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String registerRef = repositoryRef(pipeline, true); + Expression documentId = documentIdExpression(pipeline.documentId(), Exchange.FILE_NAME); + + ProcessorDefinition<?> route = from(fileEndpointUri(pipeline, registerRef)) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); + route = parseSteps(route, pipeline); + // the file consumer discards the reply, so no repository is passed to the producer: the + // endpoint register already keeps the same file version from being ingested twice + ProcessorDefinition<?> tail = route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef, null, null)); + if (pipeline.parser() != null) { + // the discarded reply would otherwise hide it: a parse to nothing typically means a + // missing Tika parser module or an image-only document, and the file's register key + // is committed, so it is not retried until the file changes + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.warn("Ingestion pipeline '{}': document '{}' parsed to no text and was skipped; its key is" + + " committed, so it is not retried until the file changes (missing parser module?" + + " image-only document?)", + pipeline.name(), result.documentId()); + } + }); + } else { + // the discarded reply would otherwise hide even the trace of a blank file + tail.process(exchange -> { + IngestResult result = exchange.getMessage().getBody(IngestResult.class); + if (result != null && result.outcome() == IngestResult.Outcome.EMPTY) { + log.debug("Ingestion pipeline '{}': document '{}' contained no text, nothing was written", + pipeline.name(), result.documentId()); + } + }); + } + } + + /** The route: consume → resolve the id → (parse) → split, embed, store; the reply is the result. */ + private void consumerRoute(IngestPipelineDefinition pipeline, String storeRef, String modelRef, String splitterRef) { + String scheme = StringHelper.before(pipeline.uri(), ":"); + requireComponent(scheme, pipeline, "its source"); + String registerRef = repositoryRef(pipeline, false); + Expression documentId = documentIdExpression(pipeline.documentId(), LangChain4jIngestHeaders.DOCUMENT_ID); + + ProcessorDefinition<?> route = from(pipeline.uri()) + .routeId(routeId(pipeline)) + .setProperty(LangChain4jIngest.DOCUMENT_ID_PROPERTY, documentId); + route = parseSteps(route, pipeline); + // deduplication by document id happens inside the producer, so a duplicate is answered + // SKIPPED, a blank delivery releases its claim and a failed write can be retried + // when the id comes from a plain header, the producer is told its name: the property set + // above always wins, but the producer's missing-id error then names the header the route + // actually reads instead of the component default + String configured = pipeline.documentId(); + String documentIdHeader = configured != null && !configured.contains("${") && !configured.contains("$simple{") + ? configured + : null; + route.to(ingestEndpointUri(pipeline, storeRef, modelRef, splitterRef, registerRef, documentIdHeader)); + } + + /** + * The optional parse stage; the route is returned unchanged when the pipeline has no parser. Tika parses in-process + * to its plain-text output — body subtree only, block boundaries newline-separated by Tika's own handlers, the + * encoding pinned — and {@link TikaTextDecode} then decodes those bytes as the pinned charset, dodging the exchange + * charset heuristic that the parsed document's own Content-Type header could steer. Docling hands the payload to a + * Docling Serve instance and answers finished markdown; the body is pinned to bytes and the {@code CamelDocling*} + * control headers are swept, because none of them may be decided by a consumer-delivered payload. + */ + private ProcessorDefinition<?> parseSteps(ProcessorDefinition<?> route, IngestPipelineDefinition pipeline) { + if (pipeline.parser() == null) { + return route; + } + if (pipeline.maxDocumentSize() > 0) { + // the endpoint's own cap counts extracted characters, which protects the splitter and + // the model but not the parse: this guard rejects the raw payload first, before tika + // or docling materialize it + route = route.process(rawSizeGuard(pipeline)); + } + return switch (pipeline.parser()) { + case TIKA -> route.to("tika:parse?tikaParseOutputFormat=text&tikaParseOutputEncoding=UTF-8") + .process(new TikaTextDecode()); + case DOCLING -> route.convertBodyTo(byte[].class) + .removeHeaders("CamelDocling*") + .to("docling:convert?operation=CONVERT_TO_MARKDOWN&contentInBody=true"); + }; + } + + private static Processor rawSizeGuard(IngestPipelineDefinition pipeline) { + // the declared-length header spares even the read, but only the directory pipeline's own + // file consumer is trusted to have set it: on a consumer pipeline every header may be + // attacker-supplied along with the payload, so its body is always measured - a forged + // CamelFileLength must not talk an oversized payload past the guard and into the parser + boolean trustDeclaredLength = pipeline.directory() != null; + return exchange -> { + Long declared = trustDeclaredLength + ? exchange.getMessage().getHeader(Exchange.FILE_LENGTH, Long.class) + : null; + long size; + byte[] bounded = null; + if (declared != null) { + size = declared; + } else if (exchange.getMessage().getBody() instanceof InputStream stream) { + // bounded read: an attacker-sized stream is rejected after maxDocumentSize + 1 + // bytes instead of being materialized whole in the heap just to be measured; + // an accepted stream is consumed here, so the bytes replace it as the body + int limit = pipeline.maxDocumentSize() == Integer.MAX_VALUE + ? Integer.MAX_VALUE + : pipeline.maxDocumentSize() + 1; + bounded = stream.readNBytes(limit); + size = bounded.length; + } else { + // a null body carries no bytes to guard; it flows on and becomes the EMPTY outcome + byte[] body = exchange.getMessage().getBody(byte[].class); Review Comment: (non-blocking) Only `InputStream` bodies get the bounded read. Other body types go through `getBody(byte[].class)`, which loads the whole payload into memory just to measure it, and the parser then reads it again. Example: the documented `source.uri=file:/inbox?delete=true` pipeline with `parser=tika` and `max-document-size=10000000`. It is a consumer pipeline, so `CamelFileLength` isn't trusted and the `GenericFile` body is converted in full. A multi-GB file can exhaust the heap (or go past the 2 GB array limit) before the guard rejects it, which is what the guard is meant to prevent. Converting to `InputStream` first and reusing the bounded path would avoid that. -- 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]
