This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new f3ef837e36b9 CAMEL-25463: camel-langchain4j-ingest - read a body only
up to maxDocumentSize (#27569)
f3ef837e36b9 is described below
commit f3ef837e36b9fb952f19e913424a8bf277d7be88
Author: Jiří Ondrušek <[email protected]>
AuthorDate: Thu Oct 8 18:53:42 2026 +0200
CAMEL-25463: camel-langchain4j-ingest - read a body only up to
maxDocumentSize (#27569)
A body not yet in memory (a stream, a file) was read whole before
maxDocumentSize was checked. It is now read only one byte, or character,
past the cap, which is enough to refuse it. With a cap, a File or Path
body decodes strictly, like a file-consumer body.
The text cap is now checked before the blank check, since a capped read
sees only the start of the body. An oversized body that is all blank
now fails; it answered EMPTY before.
Stream caching, on by default, still reads a stream whole before the
endpoint, into memory unless it spools to disk; the docs say so.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../catalog/docs/langchain4j-ingest-component.adoc | 10 ++-
.../main/docs/langchain4j-ingest-component.adoc | 10 ++-
.../langchain4j/ingest/IngestService.java | 20 +++---
.../ingest/LangChain4jIngestProducer.java | 73 +++++++++++++++++++---
.../ingest/LangChain4jIngestMediaTest.java | 44 +++++++++++--
.../ingest/LangChain4jIngestProducerTest.java | 66 +++++++++++++++++++
6 files changed, 195 insertions(+), 28 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-ingest-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-ingest-component.adoc
index 114985860897..b9e2977f79bb 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-ingest-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-ingest-component.adoc
@@ -201,6 +201,11 @@ memory- and CPU-expensive and a small file can still parse
into a lot of text. A
document fails the exchange like any other error (see below) and releases its
dedup claim, so
a trimmed re-delivery under the same id still ingests.
+A body that is not yet a String, such as a file or a stream, is read only one
character past
+the cap. Stream caching, on by default, reads a stream into memory before the
route reaches
+the endpoint, unless it spools to disk: on a consumer-fed pipeline, limit the
size at the
+consumer as well, or enable spooling.
+
=== Ingesting media: images, audio, video and PDF
With `modality=media` the body is read as bytes and embedded whole, as one
vector, by a
@@ -239,8 +244,9 @@ apply, a configured `documentSplitter` fails the start, and
`maxDocumentSize` an
`minDocumentSize` count bytes. Base64 encoding adds a third, so set
`maxDocumentSize`
against the model's request limit. A file consumer announces the length in
`CamelFileLength`, which is checked against the cap before the file is read;
any other
-body is checked once it is in memory. A `documentFilter` runs before the body
is read and
-sees it as delivered, a file or stream rather than text, so filter on headers
or the id.
+body not yet in memory is read only one byte past the cap. A `documentFilter`
runs before
+the body is read and sees it as delivered, a file or stream rather than text,
so filter on
+headers or the id.
Several LangChain4j providers embed images (Jina, Voyage AI, Cohere, Google
Gemini and
Bedrock Titan among them); none embeds audio yet, so an audio model is
typically a custom
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/main/docs/langchain4j-ingest-component.adoc
b/components/camel-ai/camel-langchain4j-ingest/src/main/docs/langchain4j-ingest-component.adoc
index 114985860897..b9e2977f79bb 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/main/docs/langchain4j-ingest-component.adoc
+++
b/components/camel-ai/camel-langchain4j-ingest/src/main/docs/langchain4j-ingest-component.adoc
@@ -201,6 +201,11 @@ memory- and CPU-expensive and a small file can still parse
into a lot of text. A
document fails the exchange like any other error (see below) and releases its
dedup claim, so
a trimmed re-delivery under the same id still ingests.
+A body that is not yet a String, such as a file or a stream, is read only one
character past
+the cap. Stream caching, on by default, reads a stream into memory before the
route reaches
+the endpoint, unless it spools to disk: on a consumer-fed pipeline, limit the
size at the
+consumer as well, or enable spooling.
+
=== Ingesting media: images, audio, video and PDF
With `modality=media` the body is read as bytes and embedded whole, as one
vector, by a
@@ -239,8 +244,9 @@ apply, a configured `documentSplitter` fails the start, and
`maxDocumentSize` an
`minDocumentSize` count bytes. Base64 encoding adds a third, so set
`maxDocumentSize`
against the model's request limit. A file consumer announces the length in
`CamelFileLength`, which is checked against the cap before the file is read;
any other
-body is checked once it is in memory. A `documentFilter` runs before the body
is read and
-sees it as delivered, a file or stream rather than text, so filter on headers
or the id.
+body not yet in memory is read only one byte past the cap. A `documentFilter`
runs before
+the body is read and sees it as delivered, a file or stream rather than text,
so filter on
+headers or the id.
Several LangChain4j providers embed images (Jina, Voyage AI, Cohere, Google
Gemini and
Bedrock Titan among them); none embeds audio yet, so an audio model is
typically a custom
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/IngestService.java
b/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/IngestService.java
index 4c629fbc7cf1..29837e000ed0 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/IngestService.java
+++
b/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/IngestService.java
@@ -85,23 +85,25 @@ class IngestService {
if (documentId == null || documentId.isBlank()) {
throw new IllegalArgumentException("Ingestion pipeline '" +
pipeline + "': documentId is required");
}
+ // the pipeline is whole-document-in-memory by design, so the cap is
the protection
+ // against oversized - and, on a consumer-fed pipeline, attacker-sized
- payloads. The
+ // failure releases a dedup claim like any other, so a trimmed
re-delivery still ingests.
+ // It comes first: the producer may hand over only the start of an
oversized stream, and
+ // a blank start says nothing about the rest
+ if (text != null && maxDocumentSize > 0 && text.length() >
maxDocumentSize) {
+ throw new IllegalArgumentException(
+ "Ingestion pipeline '" + pipeline + "': document '" +
documentId + "' exceeds maxDocumentSize ("
+ + text.length() + " > " +
maxDocumentSize + " characters)");
+ }
if (text == null || text.isBlank()) {
return new IngestResult(pipeline, documentId, 0,
IngestResult.Outcome.EMPTY);
}
// a benign filter, not a guard: a too-short document answers
FILTERED, which releases
- // a dedup claim like EMPTY does. Deliberately soft where the max
check below throws -
+ // a dedup claim like EMPTY does. Deliberately soft where the max
check above throws -
// undersized is a data decision, oversized a resource risk - so do
not align the two
if (minDocumentSize > 0 && text.length() < minDocumentSize) {
return new IngestResult(pipeline, documentId, 0,
IngestResult.Outcome.FILTERED);
}
- // the pipeline is whole-document-in-memory by design, so the cap is
the protection
- // against oversized - and, on a consumer-fed pipeline, attacker-sized
- payloads. The
- // failure releases a dedup claim like any other, so a trimmed
re-delivery still ingests
- if (maxDocumentSize > 0 && text.length() > maxDocumentSize) {
- throw new IllegalArgumentException(
- "Ingestion pipeline '" + pipeline + "': document '" +
documentId + "' exceeds maxDocumentSize ("
- + text.length() + " > " +
maxDocumentSize + " characters)");
- }
// the document id travels with every segment: retrieval can cite it,
and the engine that
// replaces this one needs it to find a document's vectors again
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java
b/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java
index efc3cb0566a0..50bea6afe60e 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java
+++
b/components/camel-ai/camel-langchain4j-ingest/src/main/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducer.java
@@ -16,6 +16,10 @@
*/
package org.apache.camel.component.langchain4j.ingest;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
import java.util.Arrays;
import java.util.Locale;
import java.util.Set;
@@ -279,7 +283,7 @@ public class LangChain4jIngestProducer extends
DefaultProducer {
* {@link IngestService}) needs the text - both run after the dedup claim,
unlike the id patterns.
*/
private IngestResult ingestUnlessFiltered(Exchange exchange, String
documentId, String mimeType, ContentType medium)
- throws InvalidPayloadException {
+ throws InvalidPayloadException, IOException {
Predicate filter = configuration.getDocumentFilter();
if (filter != null && !filter.matches(exchange)) {
return filtered(documentId);
@@ -290,15 +294,64 @@ public class LangChain4jIngestProducer extends
DefaultProducer {
if (announced != null) {
mediaService.checkSize(documentId, announced);
}
- // bytes, not a String - a charset conversion would corrupt them.
A null body stays
- // EMPTY; anything else must convert, so a wrong body type (a
POJO, say) fails as an
- // InvalidPayloadException instead of being answered EMPTY by a
silent null conversion
- byte[] bytes = exchange.getMessage().getBody() == null
- ? null
- : exchange.getMessage().getMandatoryBody(byte[].class);
- return mediaService.ingest(documentId, bytes, mimeType, medium);
- }
- return textService.ingest(documentId,
exchange.getMessage().getBody(String.class));
+ return mediaService.ingest(documentId, mediaBody(exchange),
mimeType, medium);
+ }
+ return textService.ingest(documentId, textBody(exchange));
+ }
+
+ /**
+ * The media body as bytes, not a String - a charset conversion would
corrupt them. A null body stays EMPTY;
+ * anything else must convert, so a wrong body type (a POJO, say) fails as
an InvalidPayloadException instead of
+ * being answered EMPTY by a silent null conversion. With maxDocumentSize
set, a body not yet in memory is read only
+ * one byte past the cap, enough for the size check to refuse it.
+ */
+ private byte[] mediaBody(Exchange exchange) throws
InvalidPayloadException, IOException {
+ Object body = exchange.getMessage().getBody();
+ int limit = readLimit();
+ InputStream stream = body == null || body instanceof byte[] || limit < 0
+ ? null : exchange.getMessage().getBody(InputStream.class);
+ if (stream == null) {
+ return body == null ? null :
exchange.getMessage().getMandatoryBody(byte[].class);
+ }
+ try (stream) {
+ return stream.readNBytes(limit);
+ }
+ }
+
+ /**
+ * The text body. With maxDocumentSize set, a body not yet a String is
read only one character past the cap, enough
+ * for the size check to refuse it.
+ */
+ private String textBody(Exchange exchange) throws IOException {
+ Object body = exchange.getMessage().getBody();
+ int limit = readLimit();
+ Reader reader = body == null || body instanceof String || limit < 0
+ ? null : readerOf(exchange);
+ if (reader == null) {
+ return exchange.getMessage().getBody(String.class);
+ }
+ try (reader) {
+ StringBuilder text = new StringBuilder();
+ char[] chunk = new char[8192];
+ int read;
+ while (text.length() < limit
+ && (read = reader.read(chunk, 0, Math.min(chunk.length,
limit - text.length()))) != -1) {
+ text.append(chunk, 0, read);
+ }
+ return text.toString();
+ }
+ }
+
+ /** The body as a Reader, or null; a File or a Path converts to a
BufferedReader, not to a Reader. */
+ private static Reader readerOf(Exchange exchange) {
+ Reader reader = exchange.getMessage().getBody(Reader.class);
+ return reader != null ? reader :
exchange.getMessage().getBody(BufferedReader.class);
+ }
+
+ /** How far a body is read: one unit past maxDocumentSize, so an oversized
one shows; -1 without a cap. */
+ private int readLimit() {
+ int max = configuration.getMaxDocumentSize();
+ return max <= 0 ? -1 : max == Integer.MAX_VALUE ? max : max + 1;
}
private IngestResult filtered(String documentId) {
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestMediaTest.java
b/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestMediaTest.java
index 322d7dc22f65..ca4343a1d138 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestMediaTest.java
+++
b/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestMediaTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.langchain4j.ingest;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
@@ -27,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import dev.langchain4j.data.message.ContentType;
import dev.langchain4j.data.segment.TextSegment;
@@ -83,6 +85,10 @@ class LangChain4jIngestMediaTest extends CamelTestSupport {
from("direct:capped")
.to("langchain4j-ingest:capped?modality=media&documentIdHeader=CamelFileName"
+ "&maxDocumentSize=100");
+ // without stream caching, which would read the whole stream
before the endpoint does
+ from("direct:capped-stream").noStreamCaching()
+
.to("langchain4j-ingest:capped-stream?modality=media&documentIdHeader=CamelFileName"
+ + "&maxDocumentSize=100");
from("direct:min")
.to("langchain4j-ingest:min?modality=media&documentIdHeader=CamelFileName"
+ "&minDocumentSize=1000000");
@@ -159,11 +165,13 @@ class LangChain4jIngestMediaTest extends CamelTestSupport
{
assertThat(result.outcome()).isEqualTo(IngestResult.Outcome.EMPTY);
// a POJO has no converter to bytes: a silent null conversion would
answer EMPTY and
- // release the claim, hiding a wiring mistake - it must fail instead
- assertThatThrownBy(() -> template.requestBodyAndHeader("direct:media",
new Object(),
- Exchange.FILE_NAME, "pojo.wav"))
- .isInstanceOf(CamelExecutionException.class)
- .hasStackTraceContaining("InvalidPayloadException");
+ // release the claim, hiding a wiring mistake - it must fail instead,
with a cap or without
+ for (String route : new String[] { "direct:media", "direct:capped" }) {
+ assertThatThrownBy(() -> template.requestBodyAndHeader(route, new
Object(),
+ Exchange.FILE_NAME, "pojo.wav"))
+ .isInstanceOf(CamelExecutionException.class)
+ .hasStackTraceContaining("InvalidPayloadException");
+ }
}
@Test
@@ -194,6 +202,32 @@ class LangChain4jIngestMediaTest extends CamelTestSupport {
assertThat(stackTraceOf(thrown)).as("the body must not have been
read").doesNotContain("body must not be read");
}
+ @Test
+ void oversizedStreamIsReadOnlyPastTheCap() {
+ // a stream is read one byte past the cap, not whole, even when its
announced length is a lie
+ AtomicLong read = new AtomicLong();
+ InputStream tenMegabytes = new InputStream() {
+ @Override
+ public int read() {
+ if (read.get() >= 10_000_000) {
+ return -1;
+ }
+ read.incrementAndGet();
+ return 0;
+ }
+ };
+ Throwable thrown = catchThrowable(() ->
template.requestBodyAndHeaders("direct:capped-stream", tenMegabytes,
+ Map.<String, Object> of(Exchange.FILE_NAME, "big.wav",
Exchange.FILE_LENGTH, 10L)));
+
+ assertThat(thrown).isInstanceOf(CamelExecutionException.class)
+ .hasStackTraceContaining("exceeds maxDocumentSize (101 > 100
bytes)");
+ assertThat(read.get()).isEqualTo(101);
+
+ IngestResult result =
template.requestBodyAndHeader("direct:capped-stream", new
ByteArrayInputStream(wav(1)),
+ Exchange.FILE_NAME, "small.wav", IngestResult.class);
+ assertThat(result.outcome()).isEqualTo(IngestResult.Outcome.INGESTED);
+ }
+
@Test
void mimeTypeComesFromTheExtensionOrTheNormalisedOption() {
// the same WAV bytes travel under every name below: only the name, or
the option, decides
diff --git
a/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducerTest.java
b/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducerTest.java
index 4d3316e19006..df44131aa81f 100644
---
a/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducerTest.java
+++
b/components/camel-ai/camel-langchain4j-ingest/src/test/java/org/apache/camel/component/langchain4j/ingest/LangChain4jIngestProducerTest.java
@@ -16,8 +16,14 @@
*/
package org.apache.camel.component.langchain4j.ingest;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -31,6 +37,7 @@ import org.apache.camel.CamelExecutionException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.test.junit6.CamelTestSupport;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -73,6 +80,9 @@ class LangChain4jIngestProducerTest extends CamelTestSupport {
+ "&documentSplitter=#bean:pipeSplitter");
from("direct:capped")
.to("langchain4j-ingest:capped?documentIdHeader=MyDocId&maxDocumentSize=20");
+ // without stream caching, which would read the whole stream
before the endpoint does
+ from("direct:capped-stream").noStreamCaching()
+
.to("langchain4j-ingest:capped-stream?documentIdHeader=MyDocId&maxDocumentSize=20");
from("direct:empty-splitter")
.to("langchain4j-ingest:empty-splitter?documentIdHeader=MyDocId"
+ "&documentSplitter=#bean:emptySplitter");
@@ -194,12 +204,68 @@ class LangChain4jIngestProducerTest extends
CamelTestSupport {
.isInstanceOf(CamelExecutionException.class)
.hasStackTraceContaining("exceeds maxDocumentSize")
.hasStackTraceContaining("characters");
+ // the cap is checked before the blank check: an oversized blank
document fails, not EMPTY
+ assertThatThrownBy(() -> template.requestBodyAndHeader("direct:capped",
+ " ".repeat(30), "MyDocId", "doc-oversized-blank"))
+ .isInstanceOf(CamelExecutionException.class)
+ .hasStackTraceContaining("exceeds maxDocumentSize (30 > 20
characters)");
IngestResult result = template.requestBodyAndHeader("direct:capped",
"twenty characters ok", "MyDocId", "doc-at-cap",
IngestResult.class);
assertThat(result.outcome()).isEqualTo(IngestResult.Outcome.INGESTED);
}
+ @Test
+ void oversizedStreamIsReadOnlyPastTheCap() {
+ // 10 MB of text is refused after one character past the cap; a blank
start does not make
+ // it EMPTY, the rest being unread
+ for (char content : new char[] { 'a', ' ' }) {
+ AtomicLong read = new AtomicLong();
+ assertThatThrownBy(() ->
template.requestBodyAndHeader("direct:capped-stream",
+ stream(10_000_000, content, read), "MyDocId",
"doc-stream"))
+ .isInstanceOf(CamelExecutionException.class)
+ .hasStackTraceContaining("exceeds maxDocumentSize (21 > 20
characters)");
+ // the reader buffers ahead, but nowhere near the whole stream
+ assertThat(read.get()).isLessThan(100_000);
+ }
+
+ IngestResult result =
template.requestBodyAndHeader("direct:capped-stream",
+ new ByteArrayInputStream("twenty characters
ok".getBytes(StandardCharsets.UTF_8)), "MyDocId",
+ "doc-stream-at-cap", IngestResult.class);
+ assertThat(result.outcome()).isEqualTo(IngestResult.Outcome.INGESTED);
+ }
+
+ /** A stream of the given number of bytes of one character, counting how
many are read. */
+ private static InputStream stream(long size, char content, AtomicLong
read) {
+ return new InputStream() {
+ @Override
+ public int read() {
+ if (read.get() >= size) {
+ return -1;
+ }
+ read.incrementAndGet();
+ return content;
+ }
+ };
+ }
+
+ @Test
+ void oversizedFileIsReadOnlyPastTheCap(@TempDir Path directory) throws
Exception {
+ // a File or a Path, as the file-watch consumer delivers, is read one
character past the
+ // cap as well: the reported size is what was read, not the whole file
+ Path big = Files.writeString(directory.resolve("big.txt"),
"a".repeat(10_000));
+ for (Object body : new Object[] { big.toFile(), big }) {
+ assertThatThrownBy(() ->
template.requestBodyAndHeader("direct:capped", body, "MyDocId", "doc-file"))
+ .isInstanceOf(CamelExecutionException.class)
+ .hasStackTraceContaining("exceeds maxDocumentSize (21 > 20
characters)");
+ }
+
+ Path atCap = Files.writeString(directory.resolve("ok.txt"), "twenty
characters ok");
+ IngestResult result = template.requestBodyAndHeader("direct:capped",
atCap.toFile(), "MyDocId",
+ "doc-file-at-cap", IngestResult.class);
+ assertThat(result.outcome()).isEqualTo(IngestResult.Outcome.INGESTED);
+ }
+
@Test
void customBatchSizeControlsTheEmbeddingRequests() {
// 10 paragraphs of ~19 characters against maxSegmentSize=30: one
segment each, and a
