diff --git a/kamelets/docling-convert-action.kamelet.yaml b/kamelets/docling-convert-action.kamelet.yaml index 706985f92..6f6a3e0b6 100644 --- a/kamelets/docling-convert-action.kamelet.yaml +++ b/kamelets/docling-convert-action.kamelet.yaml @@ -41,7 +41,8 @@ spec: the body is pinned to bytes and the `CamelDocling*` control headers are swept, because none of them may be decided by a consumer-delivered payload. By default the `docling` component executes a local docling CLI; set `doclingServeUrl` to use a Docling Serve - instance instead - the natural fit for self-contained Pipes. + instance instead - the natural fit for self-contained Pipes. With `maxDocumentSize` + set, a payload larger than that many bytes fails the exchange before the conversion. type: object properties: documentIdHeader: @@ -50,6 +51,13 @@ spec: CamelLangChain4jIngestDocumentId exchange property before the conversion. type: string default: CamelLangChain4jIngestDocumentId + maxDocumentSize: + title: Max Document Size + description: Maximum size of the payload in bytes, checked before the conversion; a + larger document fails the exchange without being converted. 0, the default, means no + limit. + type: integer + default: 0 doclingServeUrl: title: Docling Serve URL description: Address of a Docling Serve instance to convert through, such as @@ -74,6 +82,38 @@ spec: expression: simple: expression: "${header[{{documentIdHeader}}]}" + # the size cap, present only when maxDocumentSize is set: the conversion materializes + # the document, so an oversized one is rejected first. The precondition is evaluated + # once, when the Kamelet is instantiated - only then does ${properties:maxDocumentSize:0} + # see the parameter, at runtime it resolves to 0 - so the check must stay a + # precondition, and the comparison below uses {{maxDocumentSize}} + - choice: + precondition: true + when: + - expression: + simple: + expression: "${properties:maxDocumentSize:0} > 0" + steps: + # length() reads a stream to its end and closes it, so a stream body that + # stream caching (on by default) left uncached is read into bytes first; files, + # byte arrays and cached streams are measured in place, without a copy + - choice: + when: + - expression: + simple: + expression: "${body} is 'java.io.InputStream' && ${body} !is 'org.apache.camel.StreamCache'" + steps: + - convertBodyTo: + type: byte[] + - choice: + when: + - expression: + simple: + expression: "${length()} > {{maxDocumentSize}}" + steps: + - throwException: + exceptionType: java.lang.IllegalArgumentException + message: "Document '${exchangeProperty.CamelLangChain4jIngestDocumentId}' exceeds maxDocumentSize (${length()} > {{maxDocumentSize}} bytes)" - convertBodyTo: type: byte[] - removeHeaders: diff --git a/kamelets/tika-extract-text-action.kamelet.yaml b/kamelets/tika-extract-text-action.kamelet.yaml index b959b8875..1cc141923 100644 --- a/kamelets/tika-extract-text-action.kamelet.yaml +++ b/kamelets/tika-extract-text-action.kamelet.yaml @@ -42,7 +42,8 @@ spec: after the parse all non-Camel headers are removed: past this action, parsed document metadata is indistinguishable from caller headers, so it must not travel further. Each format needs its Tika parser module on the classpath; camel-tika ships the HTML and - text modules, PDF needs org.apache.tika:tika-parser-pdf-module. + text modules, PDF needs org.apache.tika:tika-parser-pdf-module. With `maxDocumentSize` + set, a payload larger than that many bytes fails the exchange before the parse. type: object properties: documentIdHeader: @@ -51,6 +52,12 @@ spec: CamelLangChain4jIngestDocumentId exchange property before the parse. type: string default: CamelLangChain4jIngestDocumentId + maxDocumentSize: + title: Max Document Size + description: Maximum size of the payload in bytes, checked before the parse; a larger + document fails the exchange without being parsed. 0, the default, means no limit. + type: integer + default: 0 dependencies: - "camel:tika" - "camel:kamelet" @@ -65,6 +72,38 @@ spec: expression: simple: expression: "${header[{{documentIdHeader}}]}" + # the size cap, present only when maxDocumentSize is set: the parse materializes the + # document, so an oversized one is rejected first. The precondition is evaluated once, + # when the Kamelet is instantiated - only then does ${properties:maxDocumentSize:0} see + # the parameter, at runtime it resolves to 0 - so the check must stay a precondition, + # and the comparison below uses {{maxDocumentSize}} + - choice: + precondition: true + when: + - expression: + simple: + expression: "${properties:maxDocumentSize:0} > 0" + steps: + # length() reads a stream to its end and closes it, so a stream body that + # stream caching (on by default) left uncached is read into bytes first; files, + # byte arrays and cached streams are measured in place, without a copy + - choice: + when: + - expression: + simple: + expression: "${body} is 'java.io.InputStream' && ${body} !is 'org.apache.camel.StreamCache'" + steps: + - convertBodyTo: + type: byte[] + - choice: + when: + - expression: + simple: + expression: "${length()} > {{maxDocumentSize}}" + steps: + - throwException: + exceptionType: java.lang.IllegalArgumentException + message: "Document '${exchangeProperty.CamelLangChain4jIngestDocumentId}' exceeds maxDocumentSize (${length()} > {{maxDocumentSize}} bytes)" # tika answers plain text (body only, blocks newline-separated) at the pinned encoding - to: uri: tika diff --git a/library/camel-kamelets/src/main/resources/kamelets/docling-convert-action.kamelet.yaml b/library/camel-kamelets/src/main/resources/kamelets/docling-convert-action.kamelet.yaml index 706985f92..6f6a3e0b6 100644 --- a/library/camel-kamelets/src/main/resources/kamelets/docling-convert-action.kamelet.yaml +++ b/library/camel-kamelets/src/main/resources/kamelets/docling-convert-action.kamelet.yaml @@ -41,7 +41,8 @@ spec: the body is pinned to bytes and the `CamelDocling*` control headers are swept, because none of them may be decided by a consumer-delivered payload. By default the `docling` component executes a local docling CLI; set `doclingServeUrl` to use a Docling Serve - instance instead - the natural fit for self-contained Pipes. + instance instead - the natural fit for self-contained Pipes. With `maxDocumentSize` + set, a payload larger than that many bytes fails the exchange before the conversion. type: object properties: documentIdHeader: @@ -50,6 +51,13 @@ spec: CamelLangChain4jIngestDocumentId exchange property before the conversion. type: string default: CamelLangChain4jIngestDocumentId + maxDocumentSize: + title: Max Document Size + description: Maximum size of the payload in bytes, checked before the conversion; a + larger document fails the exchange without being converted. 0, the default, means no + limit. + type: integer + default: 0 doclingServeUrl: title: Docling Serve URL description: Address of a Docling Serve instance to convert through, such as @@ -74,6 +82,38 @@ spec: expression: simple: expression: "${header[{{documentIdHeader}}]}" + # the size cap, present only when maxDocumentSize is set: the conversion materializes + # the document, so an oversized one is rejected first. The precondition is evaluated + # once, when the Kamelet is instantiated - only then does ${properties:maxDocumentSize:0} + # see the parameter, at runtime it resolves to 0 - so the check must stay a + # precondition, and the comparison below uses {{maxDocumentSize}} + - choice: + precondition: true + when: + - expression: + simple: + expression: "${properties:maxDocumentSize:0} > 0" + steps: + # length() reads a stream to its end and closes it, so a stream body that + # stream caching (on by default) left uncached is read into bytes first; files, + # byte arrays and cached streams are measured in place, without a copy + - choice: + when: + - expression: + simple: + expression: "${body} is 'java.io.InputStream' && ${body} !is 'org.apache.camel.StreamCache'" + steps: + - convertBodyTo: + type: byte[] + - choice: + when: + - expression: + simple: + expression: "${length()} > {{maxDocumentSize}}" + steps: + - throwException: + exceptionType: java.lang.IllegalArgumentException + message: "Document '${exchangeProperty.CamelLangChain4jIngestDocumentId}' exceeds maxDocumentSize (${length()} > {{maxDocumentSize}} bytes)" - convertBodyTo: type: byte[] - removeHeaders: diff --git a/library/camel-kamelets/src/main/resources/kamelets/tika-extract-text-action.kamelet.yaml b/library/camel-kamelets/src/main/resources/kamelets/tika-extract-text-action.kamelet.yaml index b959b8875..1cc141923 100644 --- a/library/camel-kamelets/src/main/resources/kamelets/tika-extract-text-action.kamelet.yaml +++ b/library/camel-kamelets/src/main/resources/kamelets/tika-extract-text-action.kamelet.yaml @@ -42,7 +42,8 @@ spec: after the parse all non-Camel headers are removed: past this action, parsed document metadata is indistinguishable from caller headers, so it must not travel further. Each format needs its Tika parser module on the classpath; camel-tika ships the HTML and - text modules, PDF needs org.apache.tika:tika-parser-pdf-module. + text modules, PDF needs org.apache.tika:tika-parser-pdf-module. With `maxDocumentSize` + set, a payload larger than that many bytes fails the exchange before the parse. type: object properties: documentIdHeader: @@ -51,6 +52,12 @@ spec: CamelLangChain4jIngestDocumentId exchange property before the parse. type: string default: CamelLangChain4jIngestDocumentId + maxDocumentSize: + title: Max Document Size + description: Maximum size of the payload in bytes, checked before the parse; a larger + document fails the exchange without being parsed. 0, the default, means no limit. + type: integer + default: 0 dependencies: - "camel:tika" - "camel:kamelet" @@ -65,6 +72,38 @@ spec: expression: simple: expression: "${header[{{documentIdHeader}}]}" + # the size cap, present only when maxDocumentSize is set: the parse materializes the + # document, so an oversized one is rejected first. The precondition is evaluated once, + # when the Kamelet is instantiated - only then does ${properties:maxDocumentSize:0} see + # the parameter, at runtime it resolves to 0 - so the check must stay a precondition, + # and the comparison below uses {{maxDocumentSize}} + - choice: + precondition: true + when: + - expression: + simple: + expression: "${properties:maxDocumentSize:0} > 0" + steps: + # length() reads a stream to its end and closes it, so a stream body that + # stream caching (on by default) left uncached is read into bytes first; files, + # byte arrays and cached streams are measured in place, without a copy + - choice: + when: + - expression: + simple: + expression: "${body} is 'java.io.InputStream' && ${body} !is 'org.apache.camel.StreamCache'" + steps: + - convertBodyTo: + type: byte[] + - choice: + when: + - expression: + simple: + expression: "${length()} > {{maxDocumentSize}}" + steps: + - throwException: + exceptionType: java.lang.IllegalArgumentException + message: "Document '${exchangeProperty.CamelLangChain4jIngestDocumentId}' exceeds maxDocumentSize (${length()} > {{maxDocumentSize}} bytes)" # tika answers plain text (body only, blocks newline-separated) at the pinned encoding - to: uri: tika diff --git a/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-route.citrus.it.yaml b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-route.citrus.it.yaml new file mode 100644 index 000000000..722ae4800 --- /dev/null +++ b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-route.citrus.it.yaml @@ -0,0 +1,101 @@ +# --------------------------------------------------------------------------- +# 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. +# --------------------------------------------------------------------------- + +name: parser-max-size-route-test +actions: + # drain any request a previous test may have left on the shared server + - purge: + endpoints: + - name: "httpServer" + timeout: 100 + - createVariables: + variables: + - name: "docling.serve.url" + value: "http://localhost:${http.server.port}" + - name: "report.url" + value: "http://localhost:${http.server.port}/result/parser-size" + # a fresh directory per test run, see the file-source test + - name: "ingest.dir" + value: "ingest-size-citrus:randomNumber(5)" + + - camel: + cli: + run: + waitForRunningState: false + args: + - "--max-seconds=60" + integration: + file: "langchain4j-ingest/parser-max-size-route.yaml" + systemProperties: + properties: + - name: "docling.serve.url" + value: "${docling.serve.url}" + - name: "report.url" + value: "${report.url}" + - name: "ingest.dir" + value: "${ingest.dir}" + + # under the cap: parsed as usual + - http: + server: "httpServer" + receiveRequest: + POST: + path: "/result/parser-size" + type: "plaintext" + body: + data: "@contains('parsed: Short text.')@" + - http: + server: "httpServer" + sendResponse: + response: + status: 200 + reasonPhrase: "OK" + version: "HTTP/1.1" + + # over the cap: rejected before the conversion; a conversion request instead would fail here + - http: + server: "httpServer" + receiveRequest: + POST: + path: "/result/parser-size" + type: "plaintext" + body: + data: "@matches('rejected: Document .long.pdf. exceeds maxDocumentSize .100 > 40 bytes.')@" + - http: + server: "httpServer" + sendResponse: + response: + status: 200 + reasonPhrase: "OK" + version: "HTTP/1.1" + + # a file over the cap: rejected before the parse + - http: + server: "httpServer" + receiveRequest: + POST: + path: "/result/parser-size" + type: "plaintext" + body: + data: "@matches('rejected: Document .big.txt. exceeds maxDocumentSize .100 > 40 bytes.')@" + - http: + server: "httpServer" + sendResponse: + response: + status: 200 + reasonPhrase: "OK" + version: "HTTP/1.1" diff --git a/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-route.yaml b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-route.yaml new file mode 100644 index 000000000..da97a95b7 --- /dev/null +++ b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-route.yaml @@ -0,0 +1,84 @@ +# --------------------------------------------------------------------------- +# 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. +# --------------------------------------------------------------------------- + +# maxDocumentSize on the parser actions: the cap is checked before the parse or conversion. +# One route feeds three documents in turn, each report waiting for the test's answer, so +# the order is fixed: a short text through Tika under the cap, an oversized payload through +# Docling (pointed at the test's HTTP server, so a payload leaking past the cap would show up +# there as a conversion request), and an oversized file through the file source and Tika. +- route: + id: parser-max-size-route + from: + uri: "timer:size" + parameters: + repeatCount: 1 + steps: + - setBody: + constant: "Short text." + - setHeader: + name: "CamelLangChain4jIngestDocumentId" + constant: "short.txt" + - doTry: + steps: + - to: "kamelet:tika-extract-text-action?maxDocumentSize=40" + - setBody: + simple: "parsed: ${body}" + doCatch: + - exception: + - "java.lang.IllegalArgumentException" + steps: + - setBody: + simple: "rejected: ${exception.message}" + - to: "{{report.url}}" + # 100 bytes, over the cap of 40 + - setBody: + constant: "0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789" + - setHeader: + name: "CamelLangChain4jIngestDocumentId" + constant: "long.pdf" + - doTry: + steps: + - to: "kamelet:docling-convert-action?maxDocumentSize=40&useDoclingServe=true&doclingServeUrl={{docling.serve.url}}" + - setBody: + simple: "converted: ${body}" + doCatch: + - exception: + - "java.lang.IllegalArgumentException" + steps: + - setBody: + simple: "rejected: ${exception.message}" + - to: "{{report.url}}" + - setBody: + constant: "0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789" + - to: "file:{{ingest.dir}}?fileName=big.txt" +- route: + id: parser-max-size-file-route + from: + uri: "kamelet:langchain4j-ingest-file-source?directory={{ingest.dir}}" + steps: + - doTry: + steps: + - to: "kamelet:tika-extract-text-action?maxDocumentSize=40" + - setBody: + simple: "parsed: ${body}" + doCatch: + - exception: + - "java.lang.IllegalArgumentException" + steps: + - setBody: + simple: "rejected: ${exception.message}" + - to: "{{report.url}}" diff --git a/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-stream-route.citrus.it.yaml b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-stream-route.citrus.it.yaml new file mode 100644 index 000000000..91fb30293 --- /dev/null +++ b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-stream-route.citrus.it.yaml @@ -0,0 +1,78 @@ +# --------------------------------------------------------------------------- +# 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. +# --------------------------------------------------------------------------- + +name: parser-max-size-stream-route-test +actions: + # drain any request a previous test may have left on the shared server + - purge: + endpoints: + - name: "httpServer" + timeout: 100 + - createVariables: + variables: + - name: "report.url" + value: "http://localhost:${http.server.port}/result/parser-size-stream" + + # stream caching off: the actions see raw streams, not cached ones + - camel: + cli: + run: + waitForRunningState: false + args: + - "--max-seconds=60" + integration: + file: "langchain4j-ingest/parser-max-size-stream-route.yaml" + systemProperties: + properties: + - name: "report.url" + value: "${report.url}" + - name: "camel.main.streamCachingEnabled" + value: "false" + + # under the cap: the whole stream reaches the parser + - http: + server: "httpServer" + receiveRequest: + POST: + path: "/result/parser-size-stream" + type: "plaintext" + body: + data: "@contains('parsed: Short text.')@" + - http: + server: "httpServer" + sendResponse: + response: + status: 200 + reasonPhrase: "OK" + version: "HTTP/1.1" + + # over the cap: rejected, the size measured once and reported right + - http: + server: "httpServer" + receiveRequest: + POST: + path: "/result/parser-size-stream" + type: "plaintext" + body: + data: "@matches('rejected: Document .long.txt. exceeds maxDocumentSize .100 > 40 bytes.')@" + - http: + server: "httpServer" + sendResponse: + response: + status: 200 + reasonPhrase: "OK" + version: "HTTP/1.1" diff --git a/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-stream-route.yaml b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-stream-route.yaml new file mode 100644 index 000000000..38a464f50 --- /dev/null +++ b/tests/camel-kamelets-itest/src/test/resources/langchain4j-ingest/parser-max-size-stream-route.yaml @@ -0,0 +1,66 @@ +# --------------------------------------------------------------------------- +# 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. +# --------------------------------------------------------------------------- + +# maxDocumentSize with stream caching switched off, so a stream body reaches the action raw +# instead of cached. A raw stream under the cap must still be parsed whole - the measurement +# must not drain it - and one over the cap is rejected with its exact size. +- route: + id: parser-max-size-stream-route + from: + uri: "timer:stream" + parameters: + repeatCount: 1 + steps: + - setBody: + constant: "Short text." + - setBody: + simple: "${bodyAs(java.io.InputStream)}" + - setHeader: + name: "CamelLangChain4jIngestDocumentId" + constant: "short.txt" + - doTry: + steps: + - to: "kamelet:tika-extract-text-action?maxDocumentSize=40" + - setBody: + simple: "parsed: ${body}" + doCatch: + - exception: + - "java.lang.IllegalArgumentException" + steps: + - setBody: + simple: "rejected: ${exception.message}" + - to: "{{report.url}}" + # 100 bytes, over the cap of 40 + - setBody: + constant: "0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789" + - setBody: + simple: "${bodyAs(java.io.InputStream)}" + - setHeader: + name: "CamelLangChain4jIngestDocumentId" + constant: "long.txt" + - doTry: + steps: + - to: "kamelet:tika-extract-text-action?maxDocumentSize=40" + - setBody: + simple: "parsed: ${body}" + doCatch: + - exception: + - "java.lang.IllegalArgumentException" + steps: + - setBody: + simple: "rejected: ${exception.message}" + - to: "{{report.url}}"