This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4810-timeout-improvements in repository https://gitbox.apache.org/repos/asf/tika.git
commit 846f87f8036b4c7e3acfe7d40f0d1b8f691e72a4 Author: tallison <[email protected]> AuthorDate: Mon Aug 10 21:36:43 2026 -0400 TIKA-4813: wire pipes watchdog to ParseTimeout checkpoints --- docs/modules/ROOT/pages/pipes/timeouts.adoc | 335 +++++++++++++++------ .../apache/tika/async/cli/AsyncProcessorTest.java | 54 ++++ .../org/apache/tika/pipes/api/PipesResult.java | 13 +- .../org/apache/tika/pipes/core/PipesClient.java | 43 +-- .../tika/pipes/core/async/AsyncProcessor.java | 3 +- .../tika/pipes/core/server/ConnectionHandler.java | 20 +- .../apache/tika/pipes/core/server/EmitHandler.java | 20 ++ .../tika/pipes/core/server/ParseHandler.java | 8 +- .../apache/tika/pipes/core/server/PipesServer.java | 57 ++-- .../apache/tika/pipes/core/EmbeddedLimitsTest.java | 6 +- .../apache/tika/pipes/core/PipesClientTest.java | 35 +++ .../pipes/pipesiterator/PipesIteratorBase.java | 14 +- .../tika/pipes/fetcher/http/HttpFetcher.java | 5 +- .../server/core/resource/PipesParsingHelper.java | 2 +- 14 files changed, 449 insertions(+), 166 deletions(-) diff --git a/docs/modules/ROOT/pages/pipes/timeouts.adoc b/docs/modules/ROOT/pages/pipes/timeouts.adoc index 1cdddd7746..c33a234c4c 100644 --- a/docs/modules/ROOT/pages/pipes/timeouts.adoc +++ b/docs/modules/ROOT/pages/pipes/timeouts.adoc @@ -19,144 +19,293 @@ == Overview -Tika Pipes uses a two-tier timeout system to handle both long-running tasks and hung parsers: +Three timeout settings bound a parse, and each answers a different question: -* **`progressTimeoutMillis`** -- Maximum time between progress updates. - If no progress is reported within this interval, the task is considered stalled and killed. - Default: `60000` (1 minute). +* **`totalTaskTimeoutMillis`** -- How long may this *document* take, end to end, + including all of its embedded documents. Default: `3600000` (1 hour). -* **`totalTaskTimeoutMillis`** -- Maximum wall-clock time for an entire task. - Even if the parser is making progress, the task is killed after this time. - Default: `3600000` (1 hour). +* **`progressTimeoutMillis`** -- How long may the parse go *silent* before it is + considered hung and killed. Default: `60000` (1 minute). -Parsers that never report progress effectively get `progressTimeoutMillis` as their total timeout. -Parsers that do report progress (e.g., OCR processing multiple pages) can run up to `totalTaskTimeoutMillis`. +* **Per-parser timeouts** (e.g., `tesseract-ocr-parser.timeoutMillis`) -- How long may + *one call* to that parser's external process or service take. + +The three do not interfere with each other: + +* A per-parser timeout is always honored, except that no single operation may be granted + more time than the document has left: the effective budget is + `min(configured, time remaining in the document)`. +* An external call in progress counts as progress -- a 10-minute `readpst` run does not + need `progressTimeoutMillis` raised to 10 minutes. While Tika waits on a bounded + external call, the wait itself reports progress; the stall detector only fires on + genuine silence (in-JVM hangs, wedged kills). +* Budgets compose recursively. A PDF inside a zip inside an email draws all of its + operations from the same document budget, at any nesting depth. == Configuration -Timeouts are configured via `TimeoutLimits` in the `parse-context` section of your JSON configuration: +Timeouts are configured via `TimeoutLimits` in the `parse-context` section of your JSON +configuration, alongside any per-parser timeouts: [source,json] ---- { + "parsers": [ + "default-parser", + { "tesseract-ocr-parser": { "timeoutMillis": 120000 } } + ], "parse-context": { "timeout-limits": { - "totalTaskTimeoutMillis": 3600000, + "totalTaskTimeoutMillis": 600000, "progressTimeoutMillis": 60000 } - } -} ----- - -This can be combined with other parse-context settings: - -[source,json] ----- -{ + }, "pipes": { "numClients": 4, "forkedJvmArgs": ["-Xmx1g"] - }, - "parse-context": { - "timeout-limits": { - "totalTaskTimeoutMillis": 7200000, - "progressTimeoutMillis": 120000 - } } } ---- -== Per-Request Overrides +== How Timeouts Are Reported -When using Tika Server with `allowPerRequestConfig: true`, timeouts can be overridden per-request -by including `TimeoutLimits` in the `ParseContext` of a `FetchEmitTuple`: +There are two families of outcome: -[source,java] ----- -ParseContext parseContext = new ParseContext(); -parseContext.setJsonConfig("timeout-limits", - "{\"progressTimeoutMillis\": 300000}"); +* **The document survives.** The worker JVM is healthy and everything extracted so far is + emitted. Interior timeouts are reported in metadata: + ** An embedded document whose external call timed out carries + `ExternalProcess.IS_TIMEOUT = true` and a recorded exception; the result status is + `PARSE_SUCCESS_WITH_EXCEPTION` / `EMIT_SUCCESS_PARSE_EXCEPTION`. + ** If the *document budget* ran out partway through, remaining embedded documents are + skipped cleanly, the document metadata carries `taskDeadlineReached = true`, and the + result status is `PARTIAL_TIMEOUT`. The content is complete up to the point the + budget was exhausted. +* **The worker is lost.** A genuine hang (or an unkillable child process) forces the + forked JVM to exit; the client restarts it automatically and the result status is + `TIMEOUT`. Content for that document is lost (except any intermediate result). + +Every timeout message reports the *requested* budget, the *granted* budget, and which +limit did the clipping, e.g.: -FetchEmitTuple t = new FetchEmitTuple("id", - new FetchKey("my-fetcher", "large-document.pdf"), - new EmitKey("my-emitter", "output-key"), - parseContext); +---- +requested=120000ms, granted=120000ms -- budget exhausted <- parser limit was binding +requested=120000ms, granted=30000ms (task remaining) <- document budget was binding ---- -== How It Works +`granted == requested` means the parser's own timeout is the one to change; +`granted < requested` means the document ran out of time and `totalTaskTimeoutMillis` +is the one to change. -=== Progress Tracking +== Scenario Walkthroughs -When a task starts, the server creates a `TikaProgressTracker` and places it in the `ParseContext`. -Parsers that perform long-running external operations (OCR, VLM inference, etc.) call -`TikaProgressTracker.update(context)` after completing each unit of work: +All scenarios use the same job -- `archive.zip` containing `report.pdf` with three +embedded images that need OCR -- and the same settings: +`totalTaskTimeoutMillis=600000` (10 min), `progressTimeoutMillis=60000` (60 s), +`tesseract-ocr-parser.timeoutMillis=120000` (120 s). -[source,java] ----- -// In a parser after completing an external process: -TikaProgressTracker.update(parseContext); ----- +=== 1. Slow but legal OCR -The server's monitoring loop checks both timeouts on every heartbeat: +Each image's OCR takes 90 seconds. -1. Has `totalTaskTimeoutMillis` elapsed since the task started? -> TIMEOUT -2. Has `progressTimeoutMillis` elapsed since the last progress update? -> TIMEOUT +* *What happens:* each OCR call is granted its full 120 s and finishes within it. The + bounded wait reports progress throughout, so the 60 s stall detector never fires even + though single calls run longer than 60 s. +* *What you see:* `EMIT_SUCCESS`, full content. +* *What to change:* nothing. -=== Which Parsers Report Progress? +=== 2. One image exceeds the parser timeout -The following parsers call `TikaProgressTracker.update()` after each external operation: +The second image is a huge noisy scan; OCR would need more than 120 s. -* `TesseractOCRParser` -- after each OCR invocation -* `ExternalParser` -- after each external process completes -* `GDALParser` -- after GDAL processing -* `Tess4JParser` -- after each in-process OCR operation -* VLM parsers (`OllamaParser`, `ClaudeParser`, etc.) -- after each API call -* `OpenAIImageEmbeddingParser` -- after each embedding call -* `StringsParser` -- after the strings command completes +* *What happens:* granted the full 120 s, the tesseract process is killed at 120 s. The + timeout is recorded against that embedded image; the third image parses normally. +* *What you see:* status `PARSE_SUCCESS_WITH_EXCEPTION`; on that image's metadata, + `ExternalProcess.IS_TIMEOUT=true` and + `requested=120000, granted=120000 -- budget exhausted`. All other content present. +* *What to change:* raise `tesseract-ocr-parser.timeoutMillis` -- or accept the loss of + one unreadable page. -Parsers that don't report progress (most built-in parsers) are bounded by `progressTimeoutMillis` alone. +=== 3. The document budget runs out -== Example: OCR Pipeline +Same job but `totalTaskTimeoutMillis=300000` (5 min), and non-OCR work takes 30 s. -For a pipeline processing scanned PDFs with hundreds of pages: +* *What happens:* image 1 is granted min(120, 270 remaining) = 120 s; image 2 gets + 120 s; image 3 gets min(120, 30 remaining) = **30 s** and is killed when that expires. + Any further embedded documents are skipped without being started. +* *What you see:* status `PARTIAL_TIMEOUT`; document metadata + `taskDeadlineReached=true`; on image 3, + `requested=120000, granted=30000 (task remaining)`. All earlier content is intact. +* *What to change:* raise `totalTaskTimeoutMillis` (this job needs roughly + 30 s + 3 x 120 s plus slack). Do *not* raise the tesseract timeout -- + `granted < requested` tells you the parser limit was not the problem. Alternatively, + accept truncation and route flagged documents to a slower lane. -[source,json] ----- -{ - "parse-context": { - "timeout-limits": { - "totalTaskTimeoutMillis": 7200000, - "progressTimeoutMillis": 300000 - } - } -} ----- +=== 4. A genuine hang -This allows up to 2 hours total per document, but kills the task if any single OCR page takes longer -than 5 minutes. A 200-page document where each page takes 30 seconds of OCR will complete successfully -(~100 minutes total), while a document stuck on a single page will be killed after 5 minutes. +A corrupt PDF sends the parser into an infinite loop before any OCR starts. -== Example: Quick Batch Processing +* *What happens:* the loop is in-JVM: no external call, so no progress. After 60 s of + silence the watchdog kills the forked JVM; the client restarts it automatically. +* *What you see:* status `TIMEOUT`; message `no progress for 60000ms`. Content for that + document is lost. Subsequent documents are unaffected. +* *What to change:* nothing -- this is the stall detector doing its job. If a retry hangs + at the same place, exclude the file and report the parser bug. -For processing many small documents where you want fast failure: +=== 5. Stall detector set too tight -[source,json] +An operator lowers `progressTimeoutMillis` to 30000; the corpus contains large, +perfectly healthy text-only PDFs. + +* *What happens:* a 500-page PDF legitimately spends 45 s inside text extraction with no + external calls, therefore no progress reports. The watchdog kills it at 30 s. +* *What you see:* status `TIMEOUT`, `no progress for 30000ms` -- but on *healthy* + documents, correlated with document size rather than reproducing at one fixed spot the + way a true hang (scenario 4) does. +* *What to change:* raise `progressTimeoutMillis` back to at least the longest honest + in-process stretch in your corpus. The 60 s default exists for this reason. + +=== 6. An unkillable child process + +Tesseract hangs in a way that ignores forced termination (rare: NFS stalls, broken +builds). + +* *What happens:* the process is killed at its 120 s budget but does not exit. The wait + cannot complete, progress stops, and 60 s later the watchdog exits the forked JVM -- + correctly, since the JVM has a stuck thread it cannot reclaim. +* *What you see:* status `TIMEOUT`; the message states the child was killed at its budget + but did not exit. +* *What to change:* no timeout setting helps. This is an environment problem (the OCR + binary, the filesystem). The message says so to prevent futile knob-turning. + +=== 7. The worker JVM dies + +The OS OOM-killer takes the forked JVM mid-parse. + +* *What happens:* all messages from the worker stop. The client's socket read times out + (`pipes.socketTimeoutMs`, default 60 s) and the client restarts the worker. +* *What you see:* status `TIMEOUT` with a client-side socket timeout stack trace and no + server diagnostic (the server is gone). +* *What to change:* infrastructure -- typically raise `-Xmx` in `pipes.forkedJvmArgs` or + check host memory. Unlike scenario 3, a plain retry may well succeed. + +== Diagnostic Quick Reference + +[cols="2,1"] +|=== +|Observation |Action + +|`requested=N, granted=N -- budget exhausted` on a child +|Raise that parser's timeout, or accept + +|`PARTIAL_TIMEOUT` / `taskDeadlineReached` (children clipped by `task remaining` or skipped) +|Raise `totalTaskTimeoutMillis`, or accept truncation + +|`TIMEOUT`, "no progress" -- reproduces at the same spot +|Parser hang: exclude the file, report the bug + +|`TIMEOUT`, "no progress" -- healthy docs, correlates with size +|`progressTimeoutMillis` too low; raise it + +|`TIMEOUT`, child killed at budget but did not exit +|Environment problem; no timeout setting helps + +|`TIMEOUT`, client socket timeout, no server messages +|Worker JVM died: memory/infrastructure +|=== + +== Misconfiguration Handling + +Some invalid combinations are rejected outright; suspicious ones produce a warning; +the rest are safe by construction. + +*Rejected at config load or startup:* + +* Negative values for `totalTaskTimeoutMillis` or `progressTimeoutMillis` (zero is + accepted -- see below). +* `pipes.socketTimeoutMs` less than or equal to `pipes.heartbeatIntervalMs` -- the client + would kill a healthy server between heartbeats. +* Unknown or renamed configuration fields (including pre-4.0 `timeoutSeconds` names) -- + configuration parsing fails on unknown properties rather than silently ignoring them. + +*Warned, but allowed (clamped rather than rejected):* + +* Zero or negative per-parser timeout (e.g. `tesseract-ocr-parser.timeoutMillis: 0`) -- + treated as unset and clamped to whatever remains of the document budget. Warned once + per parser at runtime rather than rejected at config load: there is no single + chokepoint across the several per-parser config classes to validate this centrally. +* A per-parser timeout larger than `totalTaskTimeoutMillis` (e.g., tesseract at 900 s + with a 600 s total). This is allowed because a per-request override may raise the total + for individual documents; but with these values as-is, the parser can never receive its + full budget. Warned at config load and again (once per parser) at runtime. +* `progressTimeoutMillis` greater than or equal to `totalTaskTimeoutMillis` -- stall + detection is effectively disabled, since the total deadline always arrives first. +* External-call timeouts under one second -- almost always a seconds-vs-milliseconds + mistake. + +*Accepted without warning (a coherent edge case, not a misconfiguration):* + +* Zero for `totalTaskTimeoutMillis` or `progressTimeoutMillis` -- means the task is + resuming with none of its budget left, not "disabled." A zero `progressTimeoutMillis` + is in fact the *strictest* possible stall detector (any silence at all trips it) -- + the opposite of disabling it. There is no dedicated sentinel value to disable stall + detection; to make it effectively inert, set `progressTimeoutMillis` at or above + `totalTaskTimeoutMillis` (see above). +* A per-parser timeout larger than the time *remaining* in a document -- the grant is + clipped to what remains, and the result reports + `requested=... granted=... (task remaining)`. + +== Upgrading from Tika 3.x + +Two behavioral changes in 4.0 are easy to miss -- neither shows up as a compile error or +a config validation failure. They change what already-written code and already-tuned +configs actually do. + +`TikaTimeoutException` is now checked, not a `RuntimeException`. It extends +`TikaException` so a single embedded document's timeout is recorded and its siblings +continue, using the same recovery path as every other recoverable per-embedded failure. +The cost: any code with `catch (RuntimeException e)` around a `Parser.parse()` call -- +including a custom one you wrote -- silently stops catching it. This compiles without +warning; the only symptom is an exception that used to be caught now propagating +instead. If you have such a catch block, add `TikaTimeoutException` (or its supertype +`TikaException`) explicitly. + +Deployments that raised `progressTimeoutMillis` to tolerate slow-but-legitimate OCR will +see those calls start timing out again. Before 4.0, the stall detector doubled as an +informal ceiling on how long any single external call could run, so operators with +occasional very slow scans often worked around it by raising `progressTimeoutMillis` +alone. In 4.0 the stall detector and the per-call budget are fully decoupled (see +Overview): a bounded external call that reports progress no longer needs a larger +`progressTimeoutMillis` to survive it, but it does still need an adequate per-parser +timeout (e.g. `tesseract-ocr-parser.timeoutMillis`) or `totalTaskTimeoutMillis`. If your +only lever was `progressTimeoutMillis`, raise the actual per-parser timeout instead -- +see Scenario 2. + +== Per-Request Overrides + +When using Tika Server with `allowPerRequestConfig: true`, timeouts can be overridden +per-request by including `TimeoutLimits` in the `ParseContext` of a `FetchEmitTuple`: + +[source,java] ---- -{ - "parse-context": { - "timeout-limits": { - "totalTaskTimeoutMillis": 30000, - "progressTimeoutMillis": 10000 - } - } -} +ParseContext parseContext = new ParseContext(); +parseContext.setJsonConfig("timeout-limits", + "{\"progressTimeoutMillis\": 300000}"); + +FetchEmitTuple t = new FetchEmitTuple("id", + new FetchKey("my-fetcher", "large-document.pdf"), + new EmitKey("my-emitter", "output-key"), + parseContext); ---- +Enforcement happens in the forked server against the merged configuration (server config +with per-request values applied on top). + == CLI Usage === Standard mode (single file) -For single-document parsing, `--fork` runs the parser in a forked JVM and `--fork-timeout` (milliseconds) caps how long it may run: +For single-document parsing, `--fork` runs the parser in a forked JVM and +`--fork-timeout` (milliseconds) caps how long it may run: [source,bash] ---- @@ -165,7 +314,10 @@ java -jar tika-app.jar --fork --fork-timeout=120000 document.pdf === Pipes mode (`-i` / `-o`) -In Pipes mode the parser ALREADY runs in forked JVMs — that's what `numClients` controls — so `--fork` does not apply. Setting it on the command line is silently ignored because tika-app routes `-i`/`-o` straight into the async dispatcher before its standard-mode flags are processed. +In Pipes mode the parser ALREADY runs in forked JVMs — that's what `numClients` controls +— so `--fork` does not apply. Setting it on the command line is silently ignored because +tika-app routes `-i`/`-o` straight into the async dispatcher before its standard-mode +flags are processed. Set per-parse timeouts in your `tika-config.json` instead: @@ -195,6 +347,5 @@ java -jar tika-app.jar --config=tika-config.json -i /input -o /output * link:https://github.com/apache/tika/blob/main/tika-core/src/main/java/org/apache/tika/config/TimeoutLimits.java[`TimeoutLimits.java`] -- Configuration class with defaults and helper methods * link:https://github.com/apache/tika/blob/main/tika-core/src/main/java/org/apache/tika/config/TikaProgressTracker.java[`TikaProgressTracker.java`] -- Progress tracking for parsers -* link:https://github.com/apache/tika/blob/main/tika-core/src/test/java/org/apache/tika/config/TikaProgressTrackerTest.java[`TikaProgressTrackerTest.java`] -- Unit tests for the progress tracker -* link:https://github.com/apache/tika/blob/main/tika-serialization/src/test/java/org/apache/tika/config/TimeoutLimitsTest.java[`TimeoutLimitsTest.java`] -- Unit tests for TimeoutLimits serialization +* link:https://github.com/apache/tika/blob/main/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java[`ProcessUtils.java`] -- Bounded external-process execution * link:https://github.com/apache/tika/blob/main/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java[`PipesClientTest.java`] -- Integration tests including timeout behavior diff --git a/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/AsyncProcessorTest.java b/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/AsyncProcessorTest.java index 89756956f3..d7c0ea33af 100644 --- a/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/AsyncProcessorTest.java +++ b/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/AsyncProcessorTest.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.BufferedReader; import java.io.OutputStream; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; @@ -39,6 +40,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.tika.TikaTest; +import org.apache.tika.config.TimeoutLimits; import org.apache.tika.metadata.Metadata; import org.apache.tika.metadata.TikaCoreProperties; import org.apache.tika.parser.ParseContext; @@ -279,4 +281,56 @@ public class AsyncProcessorTest extends TikaTest { processor.close(); } + + @Test + public void testPartialTimeoutContentIsEmitted() throws Exception { + // totalTaskTimeoutMillis=0 makes ParseTimeout deterministically "already exhausted" + // (remainingMillis() always 0), so the embedded doc is skipped immediately and the + // result becomes PARTIAL_TIMEOUT -- reliably ahead of the server's separate hard + // totalTaskTimeoutMillis watchdog (same threshold, polled every ~100-200ms). A + // non-zero "tight" value raced unreliably against both that watchdog and machine speed. + String mockContent = "<?xml version=\"1.0\" encoding=\"UTF-8\" ?>" + "<mock>" + + "<metadata action=\"add\" name=\"dc:creator\">Test</metadata>" + + "<write element=\"p\">main_content</write>" + + "<embedded filename=\"embed1.xml\" content-type=\"application/mock+xml\">" + + "<mock><metadata action=\"add\" name=\"dc:creator\">embeddedAuthor</metadata>" + + "<write element=\"p\">some_embedded_content</write></mock>" + + "</embedded>" + + "</mock>"; + Path mockFile = inputDir.resolve("mock-partial-timeout.xml"); + Files.write(mockFile, mockContent.getBytes(StandardCharsets.UTF_8)); + + AsyncProcessor processor = AsyncProcessor.load(configDir.resolve("tika-config.json")); + + ParseContext parseContext = new ParseContext(); + parseContext.set(ParseMode.class, ParseMode.RMETA); + parseContext.set(TimeoutLimits.class, new TimeoutLimits(0, 10000)); + FetchEmitTuple t = new FetchEmitTuple("partial-timeout-1", + new FetchKey("fsf", "mock-partial-timeout.xml"), + new EmitKey("fse-json", "emit-partial-timeout"), new Metadata(), parseContext, + FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT); + + processor.offer(t, 1000); + for (int i = 0; i < 10; i++) { + processor.offer(PipesIterator.COMPLETED_SEMAPHORE, 1000); + } + while (processor.checkActive()) { + Thread.sleep(100); + } + processor.close(); + + Path emitted = jsonOutputDir.resolve("emit-partial-timeout"); + assertTrue(Files.exists(emitted), + "PARTIAL_TIMEOUT result must still be emitted, not silently dropped: " + emitted); + List<Metadata> metadataList; + try (BufferedReader reader = Files.newBufferedReader(emitted)) { + metadataList = JsonMetadataList.fromJson(reader); + } + assertFalse(metadataList.isEmpty(), "container metadata must be present"); + // markdown output escapes underscores + assertContains("main", metadataList.get(0).get(TikaCoreProperties.TIKA_CONTENT)); + assertContains("content", metadataList.get(0).get(TikaCoreProperties.TIKA_CONTENT)); + assertEquals("true", metadataList.get(0).get(TikaCoreProperties.TASK_DEADLINE_REACHED), + "container metadata should record that the task deadline was reached"); + } } diff --git a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java index 894125fe37..fabbcd25ca 100644 --- a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java +++ b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/PipesResult.java @@ -80,7 +80,18 @@ public record PipesResult(RESULT_STATUS status, EmitData emitData, String messag PARSE_EXCEPTION_NO_EMIT(CATEGORY.SUCCESS), EMIT_SUCCESS(CATEGORY.SUCCESS), EMIT_SUCCESS_PARSE_EXCEPTION(CATEGORY.SUCCESS), - EMIT_SUCCESS_PASSBACK(CATEGORY.SUCCESS); + EMIT_SUCCESS_PASSBACK(CATEGORY.SUCCESS), + + /** + * The task's total timeout was exhausted mid-parse: the worker is healthy (no + * restart), content extracted up to that point was emitted, but one or more + * embedded documents were skipped rather than attempted. Distinct from + * {@link #TIMEOUT} (worker lost, content lost, restart) -- here the parse + * completed and a retry would truncate at the same place, so the fix is a + * bigger {@code totalTaskTimeoutMillis} or accepting the truncation, not a + * retry. See {@code TikaCoreProperties#TASK_DEADLINE_REACHED}. + */ + PARTIAL_TIMEOUT(CATEGORY.SUCCESS); private final CATEGORY category; diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java index e5702962e8..88e314eace 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java @@ -29,8 +29,6 @@ import java.io.IOException; import java.net.Socket; import java.net.SocketTimeoutException; import java.nio.charset.StandardCharsets; -import java.time.Duration; -import java.time.Instant; import java.util.ArrayList; import java.util.List; import java.util.Optional; @@ -40,7 +38,6 @@ import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.tika.config.TimeoutLimits; import org.apache.tika.metadata.Metadata; import org.apache.tika.metadata.TikaCoreProperties; import org.apache.tika.pipes.api.FetchEmitTuple; @@ -334,6 +331,17 @@ public class PipesClient implements Closeable { PipesMessage.newRequest(bytes).write(tuple.output); } + /** + * Waits for the server to finish processing {@code t}. + * <p> + * The client has no visibility into per-parser timeouts (enforced entirely inside + * the forked server, whose plugins may not even be on the client's classpath), so it + * doesn't duplicate deadline tracking here. Instead it relies on the socket's own + * {@code SO_TIMEOUT} ({@link PipesConfig#getSocketTimeoutMs()}): the server sends a + * {@code WORKING} heartbeat while alive and making progress, so a healthy-but-slow + * parse never starves this blocking read -- only a dead or wedged server lets it + * time out. + */ private PipesResult waitForServer(FetchEmitTuple t, IntermediateResult intermediateResult) throws InterruptedException { // Snapshot the volatile once; a concurrent close() may null the field, but the // local stays valid and its blocking read unblocks via socket close (IOException). @@ -342,36 +350,11 @@ public class PipesClient implements Closeable { return buildFatalResult(t.getId(), t.getEmitKey(), UNSPECIFIED_CRASH, intermediateResult.get()); } - TimeoutLimits limits = TimeoutLimits.get(t.getParseContext()); - long progressTimeoutMillis = limits.getProgressTimeoutMillis(); - long totalTaskTimeoutMillis = limits.getTotalTaskTimeoutMillis(); - Instant start = Instant.now(); - Instant lastUpdate = start; while (true) { if (Thread.currentThread().isInterrupted()) { throw new InterruptedException("thread interrupt"); } - Instant now = Instant.now(); - long totalElapsed = Duration.between(start, now).toMillis(); - if (totalElapsed > totalTaskTimeoutMillis) { - LOG.warn("clientId={}: total task timeout: id={} elapsed={}ms limit={}ms", - pipesClientId, t.getId(), totalElapsed, totalTaskTimeoutMillis); - // Mark for restart - server is stuck on current request and needs to be restarted - serverManager.markServerForRestart(); - closeConnection(); - return buildFatalResult(t.getId(), t.getEmitKey(), PipesResult.RESULT_STATUS.TIMEOUT, - intermediateResult.get()); - } - long timeSinceUpdate = Duration.between(lastUpdate, now).toMillis(); - if (timeSinceUpdate > progressTimeoutMillis) { - LOG.warn("clientId={}: progress timeout: id={} timeSinceUpdate={}ms limit={}ms", - pipesClientId, t.getId(), timeSinceUpdate, progressTimeoutMillis); - serverManager.markServerForRestart(); - closeConnection(); - return buildFatalResult(t.getId(), t.getEmitKey(), PipesResult.RESULT_STATUS.TIMEOUT, - intermediateResult.get()); - } try { PipesMessage msg = PipesMessage.read(tuple.input, maxIpcPayloadBytes); LOG.trace("clientId={}: received message type={} id={}", pipesClientId, msg.type(), t.getId()); @@ -402,10 +385,10 @@ public class PipesClient implements Closeable { intermediateResult.get(), crashMsg); case INTERMEDIATE_RESULT: intermediateResult.set(JsonPipesIpc.fromBytes(msg.payload(), Metadata.class)); - lastUpdate = Instant.now(); break; case WORKING: - lastUpdate = Instant.ofEpochMilli(msg.lastProgressMillis()); + // No-op: receiving anything at all -- including this heartbeat -- + // is what keeps the blocking read below from hitting SO_TIMEOUT. break; case FINISHED: PipesResult result = JsonPipesIpc.fromBytes(msg.payload(), PipesResult.class); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java index 69c781eb64..6afd747e8a 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/async/AsyncProcessor.java @@ -464,7 +464,8 @@ public class AsyncProcessor implements Closeable { private boolean shouldEmit(PipesResult result) { if (result.status() == PipesResult.RESULT_STATUS.PARSE_SUCCESS || - result.status() == PipesResult.RESULT_STATUS.PARSE_SUCCESS_WITH_EXCEPTION) { + result.status() == PipesResult.RESULT_STATUS.PARSE_SUCCESS_WITH_EXCEPTION || + result.status() == PipesResult.RESULT_STATUS.PARTIAL_TIMEOUT) { return true; } // Emit intermediate results on any non-success if configured diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java index ce91bbc923..75aab04648 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java @@ -41,7 +41,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.tika.config.TikaProgressTracker; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.config.TimeoutLimits; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.metadata.Metadata; @@ -166,14 +166,16 @@ public class ConnectionHandler implements Runnable, Closeable { mergedContext = resources.createMergedParseContext(fetchEmitTuple.getParseContext()); ParseContextUtils.resolveAll(mergedContext, getClass().getClassLoader()); ServerProtocolIO.validateParseContext(mergedContext); - TikaProgressTracker tracker = new TikaProgressTracker(); - mergedContext.set(TikaProgressTracker.class, tracker); + // Installed here, before submit, so the worker thread's own + // ParseTimeout.getOrCreate(mergedContext) call (inside CompositeParser) + // sees this instance rather than racing to install its own. + ParseTimeout parseTimeout = ParseTimeout.getOrCreate(mergedContext); PipesWorker pipesWorker = createPipesWorker(intermediateResult, fetchEmitTuple, mergedContext, countDownLatch); executorCompletionService.submit(pipesWorker); - loopUntilDone(fetchEmitTuple, mergedContext, intermediateResult, countDownLatch, tracker); + loopUntilDone(fetchEmitTuple, mergedContext, intermediateResult, countDownLatch, parseTimeout); } catch (TikaConfigException e) { LOG.error("handlerId={}: config error processing request", handlerId, e); handleCrash(PipesMessageType.UNSPECIFIED_CRASH, fetchEmitTuple.getId(), e); @@ -236,7 +238,7 @@ public class ConnectionHandler implements Runnable, Closeable { private void loopUntilDone(FetchEmitTuple fetchEmitTuple, ParseContext mergedContext, ArrayBlockingQueue<Metadata> intermediateResult, CountDownLatch countDownLatch, - TikaProgressTracker tracker) throws InterruptedException, IOException { + ParseTimeout parseTimeout) throws InterruptedException, IOException { Instant start = Instant.now(); TimeoutLimits limits = TimeoutLimits.get(mergedContext); long progressTimeoutMillis = limits.getProgressTimeoutMillis(); @@ -286,7 +288,7 @@ public class ConnectionHandler implements Runnable, Closeable { long elapsed = System.currentTimeMillis() - start.toEpochMilli(); if (elapsed > heartbeatCounter * heartbeatIntervalMs) { LOG.trace("handlerId={}: still processing, counter={}", handlerId, heartbeatCounter); - PipesMessage.working(tracker.getLastProgressMillis()).write(output); + PipesMessage.working(parseTimeout.getLastProgressMillis()).write(output); heartbeatCounter++; } @@ -294,7 +296,7 @@ public class ConnectionHandler implements Runnable, Closeable { if (checkTotalTimeout(start, totalTaskTimeoutMillis, fetchEmitTuple.getId())) { return; } - if (checkProgressTimeout(tracker, progressTimeoutMillis, fetchEmitTuple.getId())) { + if (checkProgressTimeout(parseTimeout, progressTimeoutMillis, fetchEmitTuple.getId())) { return; } } @@ -313,8 +315,8 @@ public class ConnectionHandler implements Runnable, Closeable { return false; } - private boolean checkProgressTimeout(TikaProgressTracker tracker, long progressTimeoutMillis, String id) { - long timeSinceProgress = System.currentTimeMillis() - tracker.getLastProgressMillis(); + private boolean checkProgressTimeout(ParseTimeout parseTimeout, long progressTimeoutMillis, String id) { + long timeSinceProgress = System.currentTimeMillis() - parseTimeout.getLastProgressMillis(); if (timeSinceProgress > progressTimeoutMillis) { handleCrash(PipesMessageType.TIMEOUT, id, new RuntimeException("Server-side progress timeout: no progress for " + timeSinceProgress + "ms (limit: " + progressTimeoutMillis + "ms)")); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java index 90ef9a6aeb..fd9f8b6289 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java @@ -68,6 +68,26 @@ class EmitHandler { } public PipesResult emitParseData(FetchEmitTuple t, MetadataListAndEmbeddedBytes parseData, ParseContext parseContext) { + PipesResult result = emitParseDataInternal(t, parseData, parseContext); + // Deadline exhaustion is a document-level fact, independent of which SUCCESS + // variant (parsed-only, emitted, passed back...) the rest of this class chose -- + // relabel any of them uniformly rather than special-casing every return site + // above. Never relabels a non-SUCCESS category: a genuine failure stays a failure. + if (result.getCategory() == PipesResult.CATEGORY.SUCCESS + && isTaskDeadlineReached(parseData.getMetadataList())) { + return new PipesResult(PipesResult.RESULT_STATUS.PARTIAL_TIMEOUT, result.emitData(), result.message()); + } + return result; + } + + private static boolean isTaskDeadlineReached(List<Metadata> metadataList) { + if (metadataIsEmpty(metadataList)) { + return false; + } + return "true".equals(metadataList.get(0).get(TikaCoreProperties.TASK_DEADLINE_REACHED)); + } + + private PipesResult emitParseDataInternal(FetchEmitTuple t, MetadataListAndEmbeddedBytes parseData, ParseContext parseContext) { long start = System.currentTimeMillis(); String stack = getContainerStacktrace(t, parseData.getMetadataList()); //we need to apply the metadata filter after we pull out the stacktrace diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java index 69a3893346..efea4cf390 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java @@ -50,7 +50,6 @@ import org.apache.tika.parser.RecursiveParserWrapper; import org.apache.tika.pipes.api.FetchEmitTuple; import org.apache.tika.pipes.api.ParseMode; import org.apache.tika.pipes.core.extractor.UnpackConfig; -import org.apache.tika.sax.AbstractRecursiveParserWrapperHandler; import org.apache.tika.sax.ContentHandlerFactory; import org.apache.tika.sax.RecursiveParserWrapperHandler; import org.apache.tika.utils.ExceptionUtils; @@ -277,10 +276,13 @@ class ParseHandler { } // Set limit reached flags from ParseRecord if (parseRecord.isEmbeddedCountLimitReached()) { - metadata.set(AbstractRecursiveParserWrapperHandler.EMBEDDED_RESOURCE_LIMIT_REACHED, true); + metadata.set(TikaCoreProperties.EMBEDDED_RESOURCE_LIMIT_REACHED, true); } if (parseRecord.isEmbeddedDepthLimitReached()) { - metadata.set(AbstractRecursiveParserWrapperHandler.EMBEDDED_DEPTH_LIMIT_REACHED, true); + metadata.set(TikaCoreProperties.EMBEDDED_DEPTH_LIMIT_REACHED, true); + } + if (parseRecord.isTaskDeadlineReached()) { + metadata.set(TikaCoreProperties.TASK_DEADLINE_REACHED, true); } if (LOG.isTraceEnabled()) { LOG.trace("timer -- parse only time: {} ms", System.currentTimeMillis() - start); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java index 5bd291cbba..c37df499de 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java @@ -47,7 +47,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.xml.sax.SAXException; -import org.apache.tika.config.TikaProgressTracker; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.config.TimeoutLimits; import org.apache.tika.config.loader.TikaJsonConfig; import org.apache.tika.config.loader.TikaLoader; @@ -188,19 +188,7 @@ public class PipesServer implements AutoCloseable { this.input = new DataInputStream(in); this.output = new DataOutputStream(out); this.heartbeatIntervalMs = pipesConfig.getHeartbeatIntervalMs(); - - // Validate heartbeat interval is less than socket timeout - if (heartbeatIntervalMs >= pipesConfig.getSocketTimeoutMs()) { - String msg = String.format(Locale.ROOT, "Heartbeat interval (%dms) must be less than socket timeout (%dms). " + - "This configuration will cause socket timeouts during normal processing.", - heartbeatIntervalMs, pipesConfig.getSocketTimeoutMs()); - - // Allow override for testing only - if (!"true".equals(System.getProperty("tika.pipes.allowInvalidHeartbeat"))) { - throw new TikaConfigException(msg); - } - LOG.error(msg + " Proceeding because tika.pipes.allowInvalidHeartbeat=true"); - } + validateHeartbeatInterval(pipesConfig); emitStrategy = pipesConfig.getEmitStrategy().getType(); this.protocolIO = new ServerProtocolIO(input, output); @@ -245,6 +233,28 @@ public class PipesServer implements AutoCloseable { } } + /** + * Fails fast if heartbeatIntervalMs >= socketTimeoutMs: the client's liveness check + * ({@code PipesClient#waitForServer}) relies solely on the socket's own + * {@code SO_TIMEOUT}, so a too-slow heartbeat makes a healthy server look dead. + * Checked once at startup rather than left as a violable javadoc warning. + */ + private static void validateHeartbeatInterval(PipesConfig pipesConfig) throws TikaConfigException { + long heartbeatIntervalMs = pipesConfig.getHeartbeatIntervalMs(); + long socketTimeoutMs = pipesConfig.getSocketTimeoutMs(); + if (heartbeatIntervalMs >= socketTimeoutMs) { + String msg = String.format(Locale.ROOT, "Heartbeat interval (%dms) must be less than socket timeout (%dms). " + + "This configuration will cause socket timeouts during normal processing.", + heartbeatIntervalMs, socketTimeoutMs); + + // Allow override for testing only + if (!"true".equals(System.getProperty("tika.pipes.allowInvalidHeartbeat"))) { + throw new TikaConfigException(msg); + } + LOG.error(msg + " Proceeding because tika.pipes.allowInvalidHeartbeat=true"); + } + } + /** * Runs the server in shared mode, accepting multiple client connections. * <p> @@ -259,6 +269,7 @@ public class PipesServer implements AutoCloseable { byte[] expectedToken) throws Exception { TikaLoader tikaLoader = TikaLoader.load(tikaConfigPath); PipesConfig pipesConfig = PipesConfig.load(tikaLoader.getConfig()); + validateHeartbeatInterval(pipesConfig); // Load shared resources SharedServerResources resources = SharedServerResources.load(tikaLoader, pipesConfig); @@ -375,13 +386,15 @@ public class PipesServer implements AutoCloseable { ParseContextUtils.resolveAll(mergedContext, getClass().getClassLoader()); // Validate the effective (merged + resolved) context ServerProtocolIO.validateParseContext(mergedContext); - TikaProgressTracker tracker = new TikaProgressTracker(); - mergedContext.set(TikaProgressTracker.class, tracker); + // Installed here, before submit, so the worker thread's own + // ParseTimeout.getOrCreate(mergedContext) call (inside CompositeParser) + // sees this instance rather than racing to install its own. + ParseTimeout parseTimeout = ParseTimeout.getOrCreate(mergedContext); PipesWorker pipesWorker = getPipesWorker(intermediateResult, fetchEmitTuple, mergedContext, countDownLatch); executorCompletionService.submit(pipesWorker); try { - loopUntilDone(fetchEmitTuple, mergedContext, executorCompletionService, intermediateResult, countDownLatch, tracker); + loopUntilDone(fetchEmitTuple, mergedContext, executorCompletionService, intermediateResult, countDownLatch, parseTimeout); } catch (Throwable t) { LOG.error("Serious problem processing request", t); } @@ -425,7 +438,7 @@ public class PipesServer implements AutoCloseable { private void loopUntilDone(FetchEmitTuple fetchEmitTuple, ParseContext mergedContext, ExecutorCompletionService<PipesResult> executorCompletionService, ArrayBlockingQueue<Metadata> intermediateResult, CountDownLatch countDownLatch, - TikaProgressTracker tracker) throws InterruptedException, IOException { + ParseTimeout parseTimeout) throws InterruptedException, IOException { Instant start = Instant.now(); TimeoutLimits limits = TimeoutLimits.get(mergedContext); long progressTimeoutMillis = limits.getProgressTimeoutMillis(); @@ -471,14 +484,14 @@ public class PipesServer implements AutoCloseable { // Send fire-and-forget heartbeat if we've waited long enough long elapsed = System.currentTimeMillis() - start.toEpochMilli(); if (elapsed > heartbeatCounter * heartbeatIntervalMs) { - PipesMessage.working(tracker.getLastProgressMillis()).write(output); + PipesMessage.working(parseTimeout.getLastProgressMillis()).write(output); heartbeatCounter++; } if (checkTotalTimeout(start, totalTaskTimeoutMillis, fetchEmitTuple.getId())) { return; // handleCrash calls exit(), but guard against unexpected return } - if (checkProgressTimeout(tracker, progressTimeoutMillis, fetchEmitTuple.getId())) { + if (checkProgressTimeout(parseTimeout, progressTimeoutMillis, fetchEmitTuple.getId())) { return; } } @@ -495,8 +508,8 @@ public class PipesServer implements AutoCloseable { return false; } - private boolean checkProgressTimeout(TikaProgressTracker tracker, long progressTimeoutMillis, String id) { - long timeSinceProgress = System.currentTimeMillis() - tracker.getLastProgressMillis(); + private boolean checkProgressTimeout(ParseTimeout parseTimeout, long progressTimeoutMillis, String id) { + long timeSinceProgress = System.currentTimeMillis() - parseTimeout.getLastProgressMillis(); if (timeSinceProgress > progressTimeoutMillis) { handleCrash(PipesMessageType.TIMEOUT, id, new RuntimeException("Server-side progress timeout: no progress for " + timeSinceProgress + "ms (limit: " + progressTimeoutMillis + "ms)")); diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/EmbeddedLimitsTest.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/EmbeddedLimitsTest.java index 58a6f6e28d..54c92dbffd 100644 --- a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/EmbeddedLimitsTest.java +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/EmbeddedLimitsTest.java @@ -27,13 +27,13 @@ import org.junit.jupiter.api.io.TempDir; import org.apache.tika.config.EmbeddedLimits; import org.apache.tika.config.loader.TikaJsonConfig; import org.apache.tika.metadata.Metadata; +import org.apache.tika.metadata.TikaCoreProperties; import org.apache.tika.parser.ParseContext; import org.apache.tika.pipes.api.FetchEmitTuple; import org.apache.tika.pipes.api.ParseMode; import org.apache.tika.pipes.api.PipesResult; import org.apache.tika.pipes.api.emitter.EmitKey; import org.apache.tika.pipes.api.fetcher.FetchKey; -import org.apache.tika.sax.AbstractRecursiveParserWrapperHandler; /** * Tests for EmbeddedLimits functionality in pipes-based parsing. @@ -265,7 +265,7 @@ public class EmbeddedLimitsTest { // Check that the limit reached flag is set Metadata containerMetadata = pipesResult.emitData().getMetadataList().get(0); - String limitReached = containerMetadata.get(AbstractRecursiveParserWrapperHandler.EMBEDDED_RESOURCE_LIMIT_REACHED); + String limitReached = containerMetadata.get(TikaCoreProperties.EMBEDDED_RESOURCE_LIMIT_REACHED); assertEquals("true", limitReached, "Container metadata should have limit reached flag set"); } @@ -295,7 +295,7 @@ public class EmbeddedLimitsTest { // Check that the depth limit reached flag is set Metadata containerMetadata = pipesResult.emitData().getMetadataList().get(0); - String limitReached = containerMetadata.get(AbstractRecursiveParserWrapperHandler.EMBEDDED_DEPTH_LIMIT_REACHED); + String limitReached = containerMetadata.get(TikaCoreProperties.EMBEDDED_DEPTH_LIMIT_REACHED); assertEquals("true", limitReached, "Container metadata should have depth limit reached flag set"); } diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java index 09bfbfaccc..12eb96aa39 100644 --- a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java +++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java @@ -234,6 +234,41 @@ public class PipesClientTest { } } + @Test + public void testWatchdogHonorsCheckpointsDuringLongExternalCall(@TempDir Path tmp) throws Exception { + // A long "external call" (checkpointedSleep, standing in for e.g. Tesseract's + // ProcessUtils.execute wait) reports progress every 300ms over its 4s run, under + // a progressTimeoutMillis (1000ms) shorter than that but a totalTaskTimeoutMillis + // with plenty of room. If the watchdog reads stale progress instead of live + // checkpoints, this times out well before the sleep completes. + Path inputDir = tmp.resolve("input"); + Files.createDirectories(inputDir); + String mockContent = "<?xml version=\"1.0\" encoding=\"UTF-8\" ?>" + "<mock>" + + "<metadata action=\"add\" name=\"dc:creator\">Test</metadata>" + + "<write element=\"p\">main_content</write>" + + "<checkpointedSleep millis=\"4000\" intervalMillis=\"300\"/>" + + "</mock>"; + String testFile = "mock-checkpointed-sleep-4s.xml"; + Files.write(inputDir.resolve(testFile), mockContent.getBytes(StandardCharsets.UTF_8)); + + Path tikaConfigPath = PluginsTestHelper.getFileSystemFetcherConfig(tmp, inputDir, tmp.resolve("output")); + TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(tikaConfigPath); + PipesConfig pipesConfig = PipesConfig.load(tikaJsonConfig); + + ParseContext parseContext = new ParseContext(); + parseContext.set(TimeoutLimits.class, new TimeoutLimits(15000, 1000)); + + try (PipesClient pipesClient = new PipesClient(pipesConfig, tikaConfigPath)) { + PipesResult result = pipesClient.process( + new FetchEmitTuple(testFile, new FetchKey(fetcherName, testFile), + new EmitKey(), new Metadata(), parseContext, + FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP)); + + assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS, result.status(), + "A 4s checkpointing wait must not trip a 1000ms progress timeout"); + } + } + @Test public void testStartupFailure(@TempDir Path tmp) throws Exception { // Create a config that references a non-existent fetcher plugin diff --git a/tika-pipes/tika-pipes-iterator-commons/src/main/java/org/apache/tika/pipes/pipesiterator/PipesIteratorBase.java b/tika-pipes/tika-pipes-iterator-commons/src/main/java/org/apache/tika/pipes/pipesiterator/PipesIteratorBase.java index 8a4622dcb8..90d9c9f51a 100644 --- a/tika-pipes/tika-pipes-iterator-commons/src/main/java/org/apache/tika/pipes/pipesiterator/PipesIteratorBase.java +++ b/tika-pipes/tika-pipes-iterator-commons/src/main/java/org/apache/tika/pipes/pipesiterator/PipesIteratorBase.java @@ -104,7 +104,7 @@ public abstract class PipesIteratorBase extends AbstractTikaExtension implements @Override public boolean hasNext() { if (next == null) { - next = pollNext(); + next = pollNextUnchecked(); } return next != COMPLETED_SEMAPHORE; } @@ -116,10 +116,20 @@ public abstract class PipesIteratorBase extends AbstractTikaExtension implements "don't call next() after hasNext() has returned false!"); } FetchEmitTuple ret = next; - next = pollNext(); + next = pollNextUnchecked(); return ret; } + // Iterator<T>'s methods cannot declare a checked exception; wrap to match + // this class's documented contract ("this will throw a RuntimeException"). + private FetchEmitTuple pollNextUnchecked() { + try { + return pollNext(); + } catch (TikaTimeoutException e) { + throw new RuntimeException(e); + } + } + private FetchEmitTuple pollNext() throws TikaTimeoutException { FetchEmitTuple t = null; diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java b/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java index dad55c9e55..170d772068 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java @@ -261,7 +261,8 @@ public class HttpFetcher extends AbstractTikaExtension implements Fetcher, Range get.setHeader(headerKey, headerValue); } - private TikaInputStream execute(HttpGet get, Metadata metadata, HttpClient client, boolean retryOnBadLength) throws IOException { + private TikaInputStream execute(HttpGet get, Metadata metadata, HttpClient client, boolean retryOnBadLength) + throws IOException, TikaTimeoutException { HttpClientContext context = HttpClientContext.create(); HttpResponse response = null; final AtomicBoolean timeout = new AtomicBoolean(false); @@ -466,7 +467,7 @@ public class HttpFetcher extends AbstractTikaExtension implements Fetcher, Range httpClientFactory.setRequestTimeoutMillis(httpFetcherConfig.getRequestTimeoutMillis()); } if (httpFetcherConfig.getConnectTimeoutMillis() != null) { - httpClientFactory.setSocketTimeoutMillis(httpFetcherConfig.getConnectTimeoutMillis()); + httpClientFactory.setConnectTimeoutMillis(httpFetcherConfig.getConnectTimeoutMillis()); } if (httpFetcherConfig.getMaxConnections() != null) { httpClientFactory.setMaxConnections(httpFetcherConfig.getMaxConnections()); diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java index 2d87de0605..14dae8c77f 100644 --- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java @@ -325,7 +325,7 @@ public class PipesParsingHelper { return switch (status) { case PARSE_SUCCESS, PARSE_SUCCESS_WITH_EXCEPTION, EMPTY_OUTPUT, EMIT_SUCCESS, EMIT_SUCCESS_PARSE_EXCEPTION, EMIT_SUCCESS_PASSBACK, - PARSE_EXCEPTION_NO_EMIT -> + PARSE_EXCEPTION_NO_EMIT, PARTIAL_TIMEOUT -> Response.Status.OK; case TIMEOUT, OOM, UNSPECIFIED_CRASH -> Response.Status.SERVICE_UNAVAILABLE;
