This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4813-pre-400 in repository https://gitbox.apache.org/repos/asf/tika.git
commit d1ca6c0c5f5b22b2822aabda0c8c85b689ec50ab Author: tallison <[email protected]> AuthorDate: Wed Aug 12 11:54:48 2026 -0400 TIKA-4813 pre-4.0 cleanup: clock-agnostic ParseTimeout, nanoTime internals, drop TikaProgressTracker, timeoutMillis naming sweep, Tess4J pool-leak fixes --- .../integration-testing/tika-eval-regression.adoc | 2 +- .../configuration/parsers/external-parser.adoc | 2 +- .../pages/configuration/parsers/tess4j-parser.adoc | 6 +- .../pages/migration-to-4x/migrating-to-4x.adoc | 11 ++ docs/modules/ROOT/pages/pipes/configuration.adoc | 8 +- docs/modules/ROOT/pages/pipes/getting-started.adoc | 2 +- docs/modules/ROOT/pages/pipes/timeouts.adoc | 44 +++++-- docs/modules/ROOT/pages/using-tika/cli/index.adoc | 2 +- .../src/main/java/org/apache/tika/cli/TikaCLI.java | 2 +- .../java/org/apache/tika/config/ParseTimeout.java | 65 +++++----- .../apache/tika/config/TikaProgressTracker.java | 90 -------------- .../java/org/apache/tika/config/TimeoutLimits.java | 8 +- .../exception/EmbeddedLimitReachedException.java | 14 +-- .../ParsingEmbeddedDocumentExtractor.java | 4 +- .../tika/parser/external/ExternalParser.java | 4 +- .../java/org/apache/tika/utils/ProcessUtils.java | 6 +- .../org/apache/tika/config/ParseTimeoutTest.java | 33 ++--- .../tika/config/TikaProgressTrackerTest.java | 115 ----------------- .../org/apache/tika/utils/ProcessUtilsTest.java | 6 +- .../java/org/apache/tika/http/TikaHttpClient.java | 17 ++- .../org/apache/tika/http/TikaHttpClientTest.java | 25 +++- .../org/apache/tika/parser/gdal/GDALParser.java | 4 +- .../tika/inference/OpenAIEmbeddingFilter.java | 4 +- .../tika/inference/OpenAIImageEmbeddingParser.java | 4 +- .../tika/parser/ocr/tess4j/Tess4JParser.java | 69 +++++++---- .../parser/ocr/tess4j/Tess4JOCRTimeoutTest.java | 138 ++++++++++++++++++++- .../apache/tika/parser/vlm/AbstractVLMParser.java | 4 +- .../apache/tika/parser/ocr/TesseractOCRParser.java | 3 +- .../apache/tika/parser/strings/StringsParser.java | 3 +- .../org/apache/tika/async/cli/PluginsWriter.java | 4 +- .../apache/tika/async/cli/SimpleAsyncConfig.java | 16 +-- .../org/apache/tika/async/cli/TikaAsyncCLI.java | 8 +- .../apache/tika/async/cli/AsyncCliParserTest.java | 12 +- .../tika/pipes/core/PerClientServerManager.java | 4 +- .../org/apache/tika/pipes/core/PipesClient.java | 8 +- .../org/apache/tika/pipes/core/PipesConfig.java | 56 ++++----- .../org/apache/tika/pipes/core/ServerManager.java | 4 +- .../tika/pipes/core/SharedServerManager.java | 4 +- .../tika/pipes/core/protocol/PipesMessage.java | 30 +---- .../tika/pipes/core/server/ConnectionHandler.java | 10 +- .../apache/tika/pipes/core/server/PipesServer.java | 26 ++-- .../tika/pipes/core/protocol/PipesMessageTest.java | 4 +- .../apache/tika/pipes/core/PipesClientTest.java | 22 ++-- .../configs/tika-config-timeout-lt-heartbeat.json | 4 +- 44 files changed, 450 insertions(+), 457 deletions(-) diff --git a/docs/modules/ROOT/pages/advanced/integration-testing/tika-eval-regression.adoc b/docs/modules/ROOT/pages/advanced/integration-testing/tika-eval-regression.adoc index 7dce903c31..6253eb9fec 100644 --- a/docs/modules/ROOT/pages/advanced/integration-testing/tika-eval-regression.adoc +++ b/docs/modules/ROOT/pages/advanced/integration-testing/tika-eval-regression.adoc @@ -169,7 +169,7 @@ Flags: * `-n` / `--numClients` — forked worker processes (default 4) * `-p` / `--pluginsDir` — explicit plugins directory (omit when running from the unzipped distribution — auto-discovery handles it) -* `-T` / `--timeoutMs` — per-parse timeout +* `-T` / `--timeoutMillis` — per-parse timeout Each run forks worker processes and produces one `.json` per input file under the output directory. Run them sequentially or in diff --git a/docs/modules/ROOT/pages/configuration/parsers/external-parser.adoc b/docs/modules/ROOT/pages/configuration/parsers/external-parser.adoc index 1bda565f56..5e4002545a 100644 --- a/docs/modules/ROOT/pages/configuration/parsers/external-parser.adoc +++ b/docs/modules/ROOT/pages/configuration/parsers/external-parser.adoc @@ -97,7 +97,7 @@ configured, the raw bytes are written as text. |`boolean` |Store raw stderr in metadata. Default: `true`. -|`timeoutMs` +|`timeoutMillis` |`long` |Process timeout in milliseconds. Default: `60000`. diff --git a/docs/modules/ROOT/pages/configuration/parsers/tess4j-parser.adoc b/docs/modules/ROOT/pages/configuration/parsers/tess4j-parser.adoc index c81a173f98..ab47e364c1 100644 --- a/docs/modules/ROOT/pages/configuration/parsers/tess4j-parser.adoc +++ b/docs/modules/ROOT/pages/configuration/parsers/tess4j-parser.adoc @@ -181,9 +181,9 @@ libraries. Prepended to `jna.library.path` at initialization time. threads that will call the parser concurrently. Each instance consumes native memory. -|`timeoutSeconds` -|`120` -|Maximum time (seconds) to wait for a pooled `Tesseract` instance before +|`timeoutMillis` +|`120000` +|Maximum time (milliseconds) to wait for a pooled `Tesseract` instance before throwing an exception. |`dpi` diff --git a/docs/modules/ROOT/pages/migration-to-4x/migrating-to-4x.adoc b/docs/modules/ROOT/pages/migration-to-4x/migrating-to-4x.adoc index a4d178eb59..66b986bedd 100644 --- a/docs/modules/ROOT/pages/migration-to-4x/migrating-to-4x.adoc +++ b/docs/modules/ROOT/pages/migration-to-4x/migrating-to-4x.adoc @@ -241,6 +241,17 @@ External parsers must now be explicitly configured via JSON. See xref:configuration/parsers/external-parser.adoc[External Parser Configuration] for details. +== Timeout Model Changes + +4.x replaces the previous ad hoc, per-parser timeout handling with a single unified model +(`TimeoutLimits` / `ParseTimeout`) shared across library use, `tika-app --fork`, and Tika +Pipes. This affects error handling (`TikaTimeoutException` is now a checked exception), +CLI flags (`tika-app --fork-timeout` was removed), and several parser/pipes config field +names (`*TimeoutSeconds`/`*TimeoutMs` -> `*TimeoutMillis`, including a unit change for +Tess4J specifically). See +xref:pipes/timeouts.adoc#_upgrading_from_tika_3_x[Timeouts: Upgrading from Tika 3.x] for +the full list of behavioral changes and required config edits. + == Deprecations and Removals * `TikaConfig` -- replaced by `TikaLoader` diff --git a/docs/modules/ROOT/pages/pipes/configuration.adoc b/docs/modules/ROOT/pages/pipes/configuration.adoc index 7049456f24..5412ad3a52 100644 --- a/docs/modules/ROOT/pages/pipes/configuration.adoc +++ b/docs/modules/ROOT/pages/pipes/configuration.adoc @@ -25,7 +25,7 @@ how many forked JVMs to run, timeouts, memory management, and parse behavior. { "pipes": { "numClients": 4, - "socketTimeoutMs": 60000, + "socketTimeoutMillis": 60000, "maxFilesProcessedPerProcess": 10000, "parseMode": "RMETA", "onParseException": "EMIT", @@ -69,13 +69,13 @@ See also xref:pipes/timeouts.adoc[Timeouts] for the full timeout model. |=== |Field |Default |Description -|`socketTimeoutMs` +|`socketTimeoutMillis` |`60000` |Maximum time (ms) to wait for data from a forked process. If no heartbeat or result is received within this window, the parse is considered hung. -|`heartbeatIntervalMs` +|`heartbeatIntervalMillis` |`1000` -|Interval (ms) between heartbeats sent from the forked process. Must be significantly less than `socketTimeoutMs`. +|Interval (ms) between heartbeats sent from the forked process. Must be significantly less than `socketTimeoutMillis`. |`shutdownClientAfterMillis` |`300000` diff --git a/docs/modules/ROOT/pages/pipes/getting-started.adoc b/docs/modules/ROOT/pages/pipes/getting-started.adoc index 4baa06c88b..ba9bfa8108 100644 --- a/docs/modules/ROOT/pages/pipes/getting-started.adoc +++ b/docs/modules/ROOT/pages/pipes/getting-started.adoc @@ -113,7 +113,7 @@ The `pipes` section controls the pipeline behavior: |`RMETA` |Output mode: `RMETA` (full recursive metadata), `CONCATENATE`, `CONTENT_ONLY`, `UNPACK` -|`socketTimeoutMs` +|`socketTimeoutMillis` |`60000` |Maximum time (ms) for a single parse operation diff --git a/docs/modules/ROOT/pages/pipes/timeouts.adoc b/docs/modules/ROOT/pages/pipes/timeouts.adoc index 6285e22114..8828f44913 100644 --- a/docs/modules/ROOT/pages/pipes/timeouts.adoc +++ b/docs/modules/ROOT/pages/pipes/timeouts.adoc @@ -41,6 +41,10 @@ The three do not interfere with each other: 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. +* All three are measured against the JVM's own monotonic clock, not calendar time -- a + system clock adjustment (NTP step, DST change, manual change) during a parse never + affects them, and time the JVM spends suspended (e.g. a laptop sleeping mid-parse) does + not count against a task's budget. == Configuration @@ -221,7 +225,7 @@ builds). 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. + (`pipes.socketTimeoutMillis`, 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 @@ -261,7 +265,7 @@ the rest are safe by construction. * Negative values for `totalTaskTimeoutMillis` or `progressTimeoutMillis` (zero is accepted -- see below). -* `pipes.socketTimeoutMs` less than or equal to `pipes.heartbeatIntervalMs` -- the client +* `pipes.socketTimeoutMillis` less than or equal to `pipes.heartbeatIntervalMillis` -- 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. @@ -295,9 +299,9 @@ the rest are safe by construction. == Upgrading from Tika 3.x -Three behavioral changes in 4.0 are easy to miss -- none of them shows up as a compile -error or a config validation failure. They change what already-written code and -already-tuned configs actually do. +Several behavioral and configuration changes in 4.0 are easy to miss -- none of them shows +up as a compile error, and most don't show up as a config validation failure either. 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 @@ -319,6 +323,27 @@ timeout (e.g. `tesseract-ocr-parser.timeoutMillis`) or `totalTaskTimeoutMillis`. only lever was `progressTimeoutMillis`, raise the actual per-parser timeout instead -- see Scenario 2. +WARNING: `tess4j-parser.timeoutSeconds` is now `tess4j-parser.timeoutMillis` -- and unlike +every other rename in this section, this one is not just a key rename, it also changes the +unit. A mechanical find-and-replace of the key name alone (`"timeoutSeconds": 120` -> +`"timeoutMillis": 120`) leaves the *value* unconverted, silently producing a **120 +millisecond** timeout instead of the intended 120 seconds -- Tess4J will then fail to +borrow a pooled `Tesseract` instance almost immediately. The value must be multiplied by +1000 along with the key rename: `"timeoutMillis": 120000`. See +xref:configuration/parsers/tess4j-parser.adoc[Tess4J OCR Configuration]. + +Several other pre-4.0 `*TimeoutSeconds`/`*TimeoutMs` fields were renamed to the +`*TimeoutMillis` convention as part of this same cleanup -- `PipesConfig`'s +`socketTimeoutMs`/`startupTimeoutMs`/`heartbeatIntervalMs` (now `socketTimeoutMillis` / +`startupTimeoutMillis` / `heartbeatIntervalMillis`), the async/pipes CLI's `--timeoutMs` +flag (now `--timeoutMillis`), and `ExternalParser`'s `timeoutMs` (now `timeoutMillis`, a +rename that landed in 4.0 itself so this doc previously lagged the code). Unlike Tess4J's +field above, none of these changed units -- only the name -- so a plain key rename with the +existing value is correct for all of them. Fields that are genuinely seconds-based and +stayed that way (e.g. `staleFetcherTimeoutSeconds`, JDBC's `queryTimeoutSeconds`, which +feeds `java.sql.Statement.setQueryTimeout(int seconds)` directly) were deliberately left +alone. + `ProcessUtils.checkCommand()` -- used by several detectors and parsers (`FileCommandDetector`, `GDALParser`, etc.) to probe whether an external binary is installed and responds, e.g. `file --version` -- now waits at most 5 seconds by default, @@ -367,7 +392,12 @@ java -jar tika-app.jar --fork --task-timeout=600000 --progress-timeout=60000 doc NOTE: The pre-4.0 `--fork-timeout` flag was removed because its meaning was ambiguous -- it conflated the total per-file budget with the stall detector. Passing it now is a hard error naming `--task-timeout` and `--progress-timeout` as the replacements, rather than -silently mapping to one or the other. +silently mapping to one or the other. `--fork-timeout` defaulted to `60000` (60 s) and, as +the sole knob, acted as an effective hard kill for the whole parse. That single default is +now split: `--task-timeout` (total per-file budget) defaults to `3600000` (1 hour) and +`--progress-timeout` (stall detector) defaults to `120000` (2 minutes) -- so a fork-mode +file that previously died around 60 s of inactivity now runs up to 2 minutes before the +stall detector fires, and up to an hour overall. === Pipes mode (`-i` / `-o`) @@ -403,6 +433,6 @@ java -jar tika-app.jar --config=tika-config.json -i /input -o /output == Living Code Reference * 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/main/java/org/apache/tika/config/ParseTimeout.java[`ParseTimeout.java`] -- Per-task timeout/progress state (budgeting, checkpointing) * 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/docs/modules/ROOT/pages/using-tika/cli/index.adoc b/docs/modules/ROOT/pages/using-tika/cli/index.adoc index 9b177e9972..e8e6a04ec1 100644 --- a/docs/modules/ROOT/pages/using-tika/cli/index.adoc +++ b/docs/modules/ROOT/pages/using-tika/cli/index.adoc @@ -405,7 +405,7 @@ This processes all files in the input directory and writes JSON metadata |`-X<size>` |`-Xmx` size for the forked processes (e.g. `-X512m`). -|`-T` or `--timeoutMs=<ms>` +|`-T` or `--timeoutMillis=<ms>` |Timeout for each parse in milliseconds. |=== diff --git a/tika-app/src/main/java/org/apache/tika/cli/TikaCLI.java b/tika-app/src/main/java/org/apache/tika/cli/TikaCLI.java index 779f54ee1b..9d137172e8 100644 --- a/tika-app/src/main/java/org/apache/tika/cli/TikaCLI.java +++ b/tika-app/src/main/java/org/apache/tika/cli/TikaCLI.java @@ -962,7 +962,7 @@ public class TikaCLI { out.println(" -o, --output=<dir> Output directory"); out.println(" -n, --numClients Number of forked processes"); out.println(" -X -Xmx in the forked processes"); - out.println(" -T, --timeoutMs Timeout for each parse in milliseconds"); + out.println(" -T, --timeoutMillis Timeout for each parse in milliseconds"); out.println(" -c, --config=<file> Tika config file (--config=<file> also accepted)"); out.println(" -p, --pluginsDir Plugins directory"); out.println(" --fileList File list (one path per line, relative to -i or absolute)"); diff --git a/tika-core/src/main/java/org/apache/tika/config/ParseTimeout.java b/tika-core/src/main/java/org/apache/tika/config/ParseTimeout.java index bcb77679d0..62f5318212 100644 --- a/tika-core/src/main/java/org/apache/tika/config/ParseTimeout.java +++ b/tika-core/src/main/java/org/apache/tika/config/ParseTimeout.java @@ -32,6 +32,10 @@ import org.apache.tika.parser.ParseContext; * (e.g. OCR on an image inside a zip inside a PDF) draws from the same remaining time as the * top-level task -- no per-depth bookkeeping needed. * <p> + * All public accessors are relative to the task (elapsed/remaining/since-last-progress), not + * anchored to wall-clock time -- internally this is backed by {@link System#nanoTime()}, so + * a system clock adjustment mid-task doesn't affect it. + * <p> * Runtime-only state (not Serializable); never sent over the wire. * <p> * Two responsibilities: @@ -49,21 +53,24 @@ public class ParseTimeout { private static final Logger LOG = LoggerFactory.getLogger(ParseTimeout.class); - private final long startMillis; - private final long hardDeadlineMillis; + private final long startNanos; + // Long.MAX_VALUE means unbounded. Kept in millis (not converted to a nanos deadline) so + // remainingMillis() only ever subtracts a small elapsed value from it -- never adds to it -- + // which sidesteps overflow without needing a special case. + private final long totalTimeoutMillis; private final long progressTimeoutMillis; - private final AtomicLong lastProgressMillis; + private final AtomicLong lastProgressNanos; // Each fires at most once per task, not once per embedded document/operation. private final AtomicBoolean warnedNonPositiveRequest = new AtomicBoolean(false); private final AtomicBoolean warnedSubSecondRequest = new AtomicBoolean(false); private final AtomicBoolean warnedRequestExceedsTotal = new AtomicBoolean(false); - private ParseTimeout(long startMillis, long hardDeadlineMillis, long progressTimeoutMillis) { - this.startMillis = startMillis; - this.hardDeadlineMillis = hardDeadlineMillis; + private ParseTimeout(long startNanos, long totalTimeoutMillis, long progressTimeoutMillis) { + this.startNanos = startNanos; + this.totalTimeoutMillis = totalTimeoutMillis; this.progressTimeoutMillis = progressTimeoutMillis; - this.lastProgressMillis = new AtomicLong(startMillis); + this.lastProgressNanos = new AtomicLong(startNanos); } /** @@ -78,7 +85,6 @@ public class ParseTimeout { * @throws IllegalArgumentException if either limit is negative */ public static ParseTimeout start(TimeoutLimits limits) { - long now = System.currentTimeMillis(); long total = limits.getTotalTaskTimeoutMillis(); long progress = limits.getProgressTimeoutMillis(); if (total < 0) { @@ -91,9 +97,7 @@ public class ParseTimeout { LOG.warn("progressTimeoutMillis ({}) >= totalTaskTimeoutMillis ({}) -- the stall " + "detector can never fire before the total deadline does", progress, total); } - // Avoid overflow: a total near MAX_VALUE would wrap negative and expire the task immediately. - long deadline = (total >= Long.MAX_VALUE - now) ? Long.MAX_VALUE : now + total; - return new ParseTimeout(now, deadline, progress); + return new ParseTimeout(System.nanoTime(), total, progress); } /** @@ -167,7 +171,7 @@ public class ParseTimeout { LOG.warn("a requested timeout of {}ms is under one second -- this is often a " + "seconds-vs-milliseconds mistake in the caller's configuration", requestedMillis); } - long total = totalMillis(); + long total = getTotalTimeoutMillis(); if (total != Long.MAX_VALUE && requestedMillis > total && !warnedRequestExceedsTotal.getAndSet(true)) { LOG.warn("a requested timeout of {}ms exceeds totalTaskTimeoutMillis ({}ms) -- it can " + "never be granted in full; raise totalTaskTimeoutMillis or lower this timeout", @@ -180,18 +184,25 @@ public class ParseTimeout { * @return the task's original total timeout in milliseconds, or {@code Long.MAX_VALUE} * if unbounded -- unlike {@link #remainingMillis()}, this does not shrink over time */ - private long totalMillis() { - return hardDeadlineMillis == Long.MAX_VALUE ? Long.MAX_VALUE : hardDeadlineMillis - startMillis; + public long getTotalTimeoutMillis() { + return totalTimeoutMillis; + } + + /** + * @return milliseconds elapsed since the task started + */ + public long elapsedMillis() { + return (System.nanoTime() - startNanos) / 1_000_000L; } /** * @return milliseconds remaining before the task's total timeout, never negative */ public long remainingMillis() { - if (hardDeadlineMillis == Long.MAX_VALUE) { + if (totalTimeoutMillis == Long.MAX_VALUE) { return Long.MAX_VALUE; } - return Math.max(0, hardDeadlineMillis - System.currentTimeMillis()); + return Math.max(0, totalTimeoutMillis - elapsedMillis()); } /** @@ -200,14 +211,14 @@ public class ParseTimeout { * {@code ParseRecord}), not here. */ public void checkpoint() { - lastProgressMillis.set(System.currentTimeMillis()); + lastProgressNanos.set(System.nanoTime()); } /** - * @return epoch millis of the last checkpoint + * @return milliseconds elapsed since the last checkpoint */ - public long getLastProgressMillis() { - return lastProgressMillis.get(); + public long millisSinceLastProgress() { + return (System.nanoTime() - lastProgressNanos.get()) / 1_000_000L; } /** @@ -216,18 +227,4 @@ public class ParseTimeout { public long getProgressTimeoutMillis() { return progressTimeoutMillis; } - - /** - * @return epoch millis when this task's total timeout started counting - */ - public long getStartMillis() { - return startMillis; - } - - /** - * @return epoch millis of the task's hard deadline, or {@code Long.MAX_VALUE} if unbounded - */ - public long getHardDeadlineMillis() { - return hardDeadlineMillis; - } } diff --git a/tika-core/src/main/java/org/apache/tika/config/TikaProgressTracker.java b/tika-core/src/main/java/org/apache/tika/config/TikaProgressTracker.java deleted file mode 100644 index abc76c382c..0000000000 --- a/tika-core/src/main/java/org/apache/tika/config/TikaProgressTracker.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.tika.config; - -import java.util.concurrent.atomic.AtomicLong; - -import org.apache.tika.parser.ParseContext; - -/** - * Tracks parse progress for the two-tier timeout system. - * <p> - * Parsers call {@link #update(ParseContext)} after completing a unit of work - * (e.g., finishing an external process for one page of OCR). The monitoring loop - * reads {@link #getLastProgressMillis()} to detect stalled parsers. - * <p> - * This is runtime-only state (NOT Serializable) — it is placed into - * {@link ParseContext} on the server side before submitting a parse task - * and is never sent over the wire. - * - * @deprecated superseded by {@link ParseTimeout#checkpoint()}, installed automatically per - * task (see {@link ParseTimeout#getOrCreate(ParseContext)}) and also bounding the task's - * total timeout. {@link #update(ParseContext)} still works and now also checkpoints the - * {@link ParseTimeout} in the same context, so existing callers are unaffected; new code - * should call {@link ParseTimeout#checkpoint(ParseContext)} directly. - * @since Apache Tika 4.0 - */ -@Deprecated -public class TikaProgressTracker { - - private final AtomicLong lastProgressMillis; - - /** - * Creates a tracker initialized to the current time. - */ - public TikaProgressTracker() { - this.lastProgressMillis = new AtomicLong(System.currentTimeMillis()); - } - - /** - * Signals progress from a ParseContext. If no tracker is present, this is a no-op. - * <p> - * This is the preferred way for parsers to signal progress: - * <pre> - * TikaProgressTracker.update(context); - * </pre> - * - * @param context the ParseContext (may be null) - */ - public static void update(ParseContext context) { - if (context == null) { - return; - } - TikaProgressTracker tracker = context.get(TikaProgressTracker.class); - if (tracker != null) { - tracker.update(); - } - // Delegate to the successor so old-API callers still feed stall detection/total-timeout accounting. - ParseTimeout.checkpoint(context); - } - - /** - * Signals progress directly on this tracker instance. - */ - public void update() { - lastProgressMillis.set(System.currentTimeMillis()); - } - - /** - * Returns the epoch millis of the last progress update. - * - * @return epoch millis of last progress - */ - public long getLastProgressMillis() { - return lastProgressMillis.get(); - } -} diff --git a/tika-core/src/main/java/org/apache/tika/config/TimeoutLimits.java b/tika-core/src/main/java/org/apache/tika/config/TimeoutLimits.java index ab839a5a34..47a18d9582 100644 --- a/tika-core/src/main/java/org/apache/tika/config/TimeoutLimits.java +++ b/tika-core/src/main/java/org/apache/tika/config/TimeoutLimits.java @@ -22,7 +22,7 @@ import org.apache.tika.annotation.TikaComponent; import org.apache.tika.parser.ParseContext; /** - * Configuration for the two-tier task timeout system. + * Configuration for the task timeout system. * <p> * <ul> * <li>{@code totalTaskTimeoutMillis} — bounds entire task wall-clock time, including @@ -30,9 +30,13 @@ import org.apache.tika.parser.ParseContext; * (default: 3,600,000 ms = 1 hour)</li> * <li>{@code progressTimeoutMillis} — bounds time since the last progress update; * catches infinite loops and hung processes (default: 120,000 ms = 2 minutes)</li> + * <li>{@code throwOnDeadline} — whether reaching the total timeout mid-parse throws + * (via {@link org.apache.tika.exception.EmbeddedLimitReachedException}) instead of + * skipping remaining embedded documents and returning content extracted so far + * (default: {@code false})</li> * </ul> * <p> - * These compose with any per-parser timeout via {@link ParseTimeout#budgetFor(long)}: a + * The first two compose with any per-parser timeout via {@link ParseTimeout#budgetFor(long)}: a * parser's own timeout is honored, but no operation gets more than what remains of * {@code totalTaskTimeoutMillis}. A bounded external call reports its own progress, so a * legitimately long call (e.g. a multi-minute external process) doesn't need diff --git a/tika-core/src/main/java/org/apache/tika/exception/EmbeddedLimitReachedException.java b/tika-core/src/main/java/org/apache/tika/exception/EmbeddedLimitReachedException.java index d68dac0e58..62a080c56a 100644 --- a/tika-core/src/main/java/org/apache/tika/exception/EmbeddedLimitReachedException.java +++ b/tika-core/src/main/java/org/apache/tika/exception/EmbeddedLimitReachedException.java @@ -37,10 +37,6 @@ public class EmbeddedLimitReachedException extends RuntimeException { private final LimitType limitType; private final long limit; - public EmbeddedLimitReachedException(LimitType limitType, int limit) { - this(limitType, (long) limit); - } - /** * @param limit the configured limit -- for {@link LimitType#DEADLINE} this is * {@code totalTaskTimeoutMillis}, not a count or depth @@ -68,19 +64,11 @@ public class EmbeddedLimitReachedException extends RuntimeException { return limitType; } - /** - * @return the configured limit as an int -- for {@link LimitType#DEADLINE}, prefer - * {@link #getLimitMillis()} since a millisecond timeout may exceed int range - */ - public int getLimit() { - return (int) limit; - } - /** * @return the configured limit, e.g. {@code totalTaskTimeoutMillis} for * {@link LimitType#DEADLINE} */ - public long getLimitMillis() { + public long getLimit() { return limit; } } diff --git a/tika-core/src/main/java/org/apache/tika/extractor/ParsingEmbeddedDocumentExtractor.java b/tika-core/src/main/java/org/apache/tika/extractor/ParsingEmbeddedDocumentExtractor.java index 5494df5eb2..65252905b6 100644 --- a/tika-core/src/main/java/org/apache/tika/extractor/ParsingEmbeddedDocumentExtractor.java +++ b/tika-core/src/main/java/org/apache/tika/extractor/ParsingEmbeddedDocumentExtractor.java @@ -118,7 +118,7 @@ public class ParsingEmbeddedDocumentExtractor implements EmbeddedDocumentExtract // threw; every later sibling silently took the skip path instead. if (parseRecord.isThrowOnDeadline()) { throw new EmbeddedLimitReachedException(EmbeddedLimitReachedException.LimitType.DEADLINE, - timeout == null ? 0 : timeout.getHardDeadlineMillis() - timeout.getStartMillis()); + timeout == null ? 0 : timeout.getTotalTimeoutMillis()); } return false; } @@ -126,7 +126,7 @@ public class ParsingEmbeddedDocumentExtractor implements EmbeddedDocumentExtract parseRecord.setTaskDeadlineReached(true); if (parseRecord.isThrowOnDeadline()) { throw new EmbeddedLimitReachedException( - EmbeddedLimitReachedException.LimitType.DEADLINE, timeout.getHardDeadlineMillis() - timeout.getStartMillis()); + EmbeddedLimitReachedException.LimitType.DEADLINE, timeout.getTotalTimeoutMillis()); } return false; } diff --git a/tika-core/src/main/java/org/apache/tika/parser/external/ExternalParser.java b/tika-core/src/main/java/org/apache/tika/parser/external/ExternalParser.java index 5e0cfb8689..7c426006f7 100644 --- a/tika-core/src/main/java/org/apache/tika/parser/external/ExternalParser.java +++ b/tika-core/src/main/java/org/apache/tika/parser/external/ExternalParser.java @@ -37,7 +37,7 @@ import org.xml.sax.SAXException; import org.apache.tika.annotation.TikaComponent; import org.apache.tika.config.ConfigDeserializer; import org.apache.tika.config.JsonConfig; -import org.apache.tika.config.TikaProgressTracker; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.TikaException; import org.apache.tika.io.TemporaryResources; import org.apache.tika.io.TikaInputStream; @@ -193,7 +193,7 @@ public class ExternalParser implements Parser { // Set process metadata metadata.set(ExternalProcess.IS_TIMEOUT, result.isTimeout()); metadata.set(ExternalProcess.EXIT_VALUE, result.getExitValue()); - TikaProgressTracker.update(context); + ParseTimeout.checkpoint(context); metadata.set(ExternalProcess.STD_OUT_LENGTH, result.getStdoutLength()); metadata.set(ExternalProcess.STD_OUT_IS_TRUNCATED, result.isStdoutTruncated()); diff --git a/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java b/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java index 7124bd6078..b99c3688e8 100644 --- a/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java +++ b/tika-core/src/main/java/org/apache/tika/utils/ProcessUtils.java @@ -420,10 +420,10 @@ public class ProcessUtils { */ public static boolean waitForWithHeartbeat(Process p, ParseContext context, long timeoutMillis) throws InterruptedException { - long now = System.currentTimeMillis(); - long deadline = (timeoutMillis >= Long.MAX_VALUE - now) ? Long.MAX_VALUE : now + timeoutMillis; + long startNanos = System.nanoTime(); while (true) { - long remaining = deadline - System.currentTimeMillis(); + long elapsedMillis = (System.nanoTime() - startNanos) / 1_000_000L; + long remaining = timeoutMillis - elapsedMillis; long pollMillis = remaining <= 0 ? 0 : Math.min(remaining, HEARTBEAT_INTERVAL_MILLIS); if (p.waitFor(pollMillis, TimeUnit.MILLISECONDS)) { return true; diff --git a/tika-core/src/test/java/org/apache/tika/config/ParseTimeoutTest.java b/tika-core/src/test/java/org/apache/tika/config/ParseTimeoutTest.java index 74619474f2..6385099c1f 100644 --- a/tika-core/src/test/java/org/apache/tika/config/ParseTimeoutTest.java +++ b/tika-core/src/test/java/org/apache/tika/config/ParseTimeoutTest.java @@ -33,26 +33,27 @@ import org.apache.tika.parser.ParseContext; public class ParseTimeoutTest { @Test - public void testInitialTimestamp() { - long before = System.currentTimeMillis(); + public void testInitialElapsedIsNearZero() { ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits()); - long after = System.currentTimeMillis(); - assertTrue(timeout.getLastProgressMillis() >= before); - assertTrue(timeout.getLastProgressMillis() <= after); - assertTrue(timeout.getStartMillis() >= before); - assertTrue(timeout.getStartMillis() <= after); + // Generous upper bound for slack against jitter under a loaded test run -- the + // point is "close to zero right after start", not an exact value. + assertTrue(timeout.elapsedMillis() >= 0); + assertTrue(timeout.elapsedMillis() < 5000); + assertTrue(timeout.millisSinceLastProgress() >= 0); + assertTrue(timeout.millisSinceLastProgress() < 5000); } @Test - public void testCheckpointAdvancesTimestamp() throws Exception { + public void testCheckpointResetsMillisSinceLastProgress() throws Exception { ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits()); - long initial = timeout.getLastProgressMillis(); Thread.sleep(20); - timeout.checkpoint(); + long beforeCheckpoint = timeout.millisSinceLastProgress(); + assertTrue(beforeCheckpoint >= 20); - assertTrue(timeout.getLastProgressMillis() > initial); + timeout.checkpoint(); + assertTrue(timeout.millisSinceLastProgress() < beforeCheckpoint); } @Test @@ -79,7 +80,7 @@ public class ParseTimeoutTest { startLatch.countDown(); assertTrue(doneLatch.await(5, TimeUnit.SECONDS)); - assertTrue(timeout.getLastProgressMillis() > 0); + assertTrue(timeout.millisSinceLastProgress() >= 0); } @Test @@ -119,7 +120,7 @@ public class ParseTimeoutTest { // A huge or MAX_VALUE total must not wrap the deadline negative and expire // the task immediately. ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits(Long.MAX_VALUE, 60_000L)); - assertEquals(Long.MAX_VALUE, timeout.getHardDeadlineMillis()); + assertEquals(Long.MAX_VALUE, timeout.getTotalTimeoutMillis()); assertEquals(Long.MAX_VALUE, timeout.remainingMillis()); assertTrue(timeout.budgetFor(60_000L) == 60_000L); } @@ -163,12 +164,14 @@ public class ParseTimeoutTest { public void testCheckpointStaticUpdatesInstalledTimeout() throws Exception { ParseContext context = new ParseContext(); ParseTimeout timeout = ParseTimeout.getOrCreate(context); - long initial = timeout.getLastProgressMillis(); Thread.sleep(20); + long beforeCheckpoint = timeout.millisSinceLastProgress(); + assertTrue(beforeCheckpoint >= 20); + ParseTimeout.checkpoint(context); - assertTrue(timeout.getLastProgressMillis() > initial); + assertTrue(timeout.millisSinceLastProgress() < beforeCheckpoint); } // ---- misconfiguration validation (design doc §9) ---------------------------------- diff --git a/tika-core/src/test/java/org/apache/tika/config/TikaProgressTrackerTest.java b/tika-core/src/test/java/org/apache/tika/config/TikaProgressTrackerTest.java deleted file mode 100644 index 007f227a97..0000000000 --- a/tika-core/src/test/java/org/apache/tika/config/TikaProgressTrackerTest.java +++ /dev/null @@ -1,115 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.tika.config; - -import static org.junit.jupiter.api.Assertions.assertTrue; - -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import org.junit.jupiter.api.Test; - -import org.apache.tika.parser.ParseContext; - -public class TikaProgressTrackerTest { - - @Test - public void testInitialTimestamp() { - long before = System.currentTimeMillis(); - TikaProgressTracker tracker = new TikaProgressTracker(); - long after = System.currentTimeMillis(); - - assertTrue(tracker.getLastProgressMillis() >= before); - assertTrue(tracker.getLastProgressMillis() <= after); - } - - @Test - public void testUpdateAdvancesTimestamp() throws Exception { - TikaProgressTracker tracker = new TikaProgressTracker(); - long initial = tracker.getLastProgressMillis(); - - Thread.sleep(50); - tracker.update(); - - assertTrue(tracker.getLastProgressMillis() > initial); - } - - @Test - public void testConcurrentUpdates() throws Exception { - TikaProgressTracker tracker = new TikaProgressTracker(); - int numThreads = 4; - CountDownLatch startLatch = new CountDownLatch(1); - CountDownLatch doneLatch = new CountDownLatch(numThreads); - - for (int i = 0; i < numThreads; i++) { - new Thread(() -> { - try { - startLatch.await(); - for (int j = 0; j < 100; j++) { - tracker.update(); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } finally { - doneLatch.countDown(); - } - }).start(); - } - - startLatch.countDown(); - assertTrue(doneLatch.await(5, TimeUnit.SECONDS)); - assertTrue(tracker.getLastProgressMillis() > 0); - } - - @Test - public void testStaticUpdateWithParseContext() throws Exception { - TikaProgressTracker tracker = new TikaProgressTracker(); - long initial = tracker.getLastProgressMillis(); - - ParseContext context = new ParseContext(); - context.set(TikaProgressTracker.class, tracker); - - Thread.sleep(50); - TikaProgressTracker.update(context); - - assertTrue(tracker.getLastProgressMillis() > initial); - } - - @Test - public void testStaticUpdateWithNullContext() { - // Should not throw - TikaProgressTracker.update(null); - } - - @Test - public void testStaticUpdateWithNoTracker() { - // Should not throw - TikaProgressTracker.update(new ParseContext()); - } - - @Test - public void testStaticUpdateDelegatesToParseTimeout() throws Exception { - ParseContext context = new ParseContext(); - ParseTimeout parseTimeout = ParseTimeout.getOrCreate(context); - long initial = parseTimeout.getLastProgressMillis(); - - Thread.sleep(20); - TikaProgressTracker.update(context); - - assertTrue(parseTimeout.getLastProgressMillis() > initial); - } -} diff --git a/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java b/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java index 0ee0303067..cb5ba0cb9b 100644 --- a/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java +++ b/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java @@ -41,7 +41,6 @@ public class ProcessUtilsTest { ParseContext context = new ParseContext(); context.set(TimeoutLimits.class, new TimeoutLimits(60_000, 60_000)); ParseTimeout parseTimeout = ParseTimeout.getOrCreate(context); - long initialProgress = parseTimeout.getLastProgressMillis(); ProcessBuilder pb = new ProcessBuilder("sleep", "6"); Thread runner = new Thread(() -> { @@ -57,9 +56,10 @@ public class ProcessUtilsTest { // Wait a generous multiple of HEARTBEAT_INTERVAL_MILLIS (~1000ms), well short of // the process's 6s completion, for slack against jitter under a loaded test run. Thread.sleep(3000); - long midProgress = parseTimeout.getLastProgressMillis(); - assertTrue(midProgress > initialProgress, + // If no checkpoint fired during the sleep, millisSinceLastProgress() would be + // >= the full 3000ms; a mid-wait checkpoint resets it back down. + assertTrue(parseTimeout.millisSinceLastProgress() < 3000, "expected a checkpoint to have fired while the process was still running"); } finally { runner.join(10_000); diff --git a/tika-parsers/tika-http-jdk/src/main/java/org/apache/tika/http/TikaHttpClient.java b/tika-parsers/tika-http-jdk/src/main/java/org/apache/tika/http/TikaHttpClient.java index aa5507cfcb..a6b4161052 100644 --- a/tika-parsers/tika-http-jdk/src/main/java/org/apache/tika/http/TikaHttpClient.java +++ b/tika-parsers/tika-http-jdk/src/main/java/org/apache/tika/http/TikaHttpClient.java @@ -202,9 +202,16 @@ public class TikaHttpClient implements Closeable { * Resolves the requested timeout (millis) against the task's remaining budget. Unlike * the old seconds-floored-at-1 version, this can legitimately return 0 -- see * {@link #failFastIfExhausted}, which is always called right after this. + * <p> + * A null context has no task to clip against -- {@code ParseTimeout.getOrCreate(null)} + * would otherwise hand back a detached ParseTimeout built from *default* TimeoutLimits + * (1 hour), silently capping any request above that and re-firing budgetFor's + * once-per-task warnings on every call, since a fresh detached instance is created each + * time (see {@code ProcessUtils.execute}'s null-context handling for the same + * rationale). Package-private (rather than private) so it can be unit tested directly. */ - private long grantedMillis(long requestedMillis, ParseContext context) { - return ParseTimeout.getOrCreate(context).budgetFor(requestedMillis); + long grantedMillis(long requestedMillis, ParseContext context) { + return context == null ? requestedMillis : ParseTimeout.getOrCreate(context).budgetFor(requestedMillis); } /** @@ -271,10 +278,10 @@ public class TikaHttpClient implements Closeable { ParseContext context, long grantedMillis, URI uri, long requestedMillis) throws InterruptedException, ExecutionException, TikaTimeoutException { - long now = System.currentTimeMillis(); - long deadline = (grantedMillis >= Long.MAX_VALUE - now) ? Long.MAX_VALUE : now + grantedMillis; + long startNanos = System.nanoTime(); while (true) { - long remaining = deadline - System.currentTimeMillis(); + long elapsedMillis = (System.nanoTime() - startNanos) / 1_000_000L; + long remaining = grantedMillis - elapsedMillis; long pollMillis = remaining <= 0 ? 0 : Math.min(remaining, HEARTBEAT_INTERVAL_MILLIS); try { return future.get(pollMillis, TimeUnit.MILLISECONDS); diff --git a/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java index 2e122a8be5..8714f31a5c 100644 --- a/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java +++ b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java @@ -41,7 +41,6 @@ public class TikaHttpClientTest { ParseContext context = new ParseContext(); context.set(TimeoutLimits.class, new TimeoutLimits(60_000, 60_000)); ParseTimeout parseTimeout = ParseTimeout.getOrCreate(context); - long initialProgress = parseTimeout.getLastProgressMillis(); Thread requester = new Thread(() -> { try { @@ -56,9 +55,10 @@ public class TikaHttpClientTest { // Wait a generous multiple of HEARTBEAT_INTERVAL_MILLIS (~1000ms), well short of // the server's 6s delay, for slack against jitter under a loaded test run. Thread.sleep(3000); - long midProgress = parseTimeout.getLastProgressMillis(); - assertTrue(midProgress > initialProgress, + // If no checkpoint fired during the sleep, millisSinceLastProgress() would be + // >= the full 3000ms; a mid-wait checkpoint resets it back down. + assertTrue(parseTimeout.millisSinceLastProgress() < 3000, "expected a checkpoint to have fired while the request was still in flight"); } finally { requester.join(10_000); @@ -78,6 +78,25 @@ public class TikaHttpClientTest { } } + /** + * TIKA-4813 follow-up: a null context has no task to clip against, but + * {@code ParseTimeout.getOrCreate(null)} used to hand back a detached ParseTimeout built + * from *default* TimeoutLimits (1 hour) -- silently capping any request above that and + * contradicting the "granted unclipped" contract the javadoc on the no-context overloads + * promises. Exercises the package-private {@code grantedMillis} directly rather than + * waiting out a real request, since proving "not clipped to 1 hour" via actual elapsed + * time isn't practical in a unit test. + */ + @Test + public void testNullContextGrantsRequestUnclippedEvenAboveDefaultOneHour() throws Exception { + try (TikaHttpClient client = TikaHttpClient.build(30)) { + long requestedTimeoutMillis = TimeoutLimits.DEFAULT_TOTAL_TASK_TIMEOUT_MILLIS + 60_000; + + assertEquals(requestedTimeoutMillis, client.grantedMillis(requestedTimeoutMillis, null), + "a null context must not silently clip the request against a default-TimeoutLimits budget"); + } + } + @Test public void testRequestTimeoutStillBoundsTheWait() throws Exception { try (TikaTestHttpServer server = new TikaTestHttpServer(); diff --git a/tika-parsers/tika-parsers-extended/tika-parser-scientific-module/src/main/java/org/apache/tika/parser/gdal/GDALParser.java b/tika-parsers/tika-parsers-extended/tika-parser-scientific-module/src/main/java/org/apache/tika/parser/gdal/GDALParser.java index 75d7d2f89a..85b7695c09 100644 --- a/tika-parsers/tika-parsers-extended/tika-parser-scientific-module/src/main/java/org/apache/tika/parser/gdal/GDALParser.java +++ b/tika-parsers/tika-parsers-extended/tika-parser-scientific-module/src/main/java/org/apache/tika/parser/gdal/GDALParser.java @@ -39,7 +39,7 @@ import org.xml.sax.ContentHandler; import org.xml.sax.SAXException; import org.apache.tika.annotation.TikaComponent; -import org.apache.tika.config.TikaProgressTracker; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.TikaException; import org.apache.tika.io.TemporaryResources; import org.apache.tika.io.TikaInputStream; @@ -198,7 +198,7 @@ public class GDALParser implements Parser { metadata.set(ExternalProcess.IS_TIMEOUT, result.isTimeout()); metadata.set(ExternalProcess.EXIT_VALUE, result.getExitValue()); - TikaProgressTracker.update(context); + ParseTimeout.checkpoint(context); metadata.set(ExternalProcess.STD_OUT_LENGTH, result.getStdoutLength()); metadata.set(ExternalProcess.STD_OUT_IS_TRUNCATED, result.isStdoutTruncated()); metadata.set(ExternalProcess.STD_ERR_LENGTH, result.getStderrLength()); diff --git a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingFilter.java b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingFilter.java index 1f3e37fa94..d0b7e00902 100644 --- a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingFilter.java +++ b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingFilter.java @@ -27,7 +27,7 @@ import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.tika.annotation.TikaComponent; -import org.apache.tika.config.TikaProgressTracker; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.TikaException; import org.apache.tika.http.TikaHttpClient; import org.apache.tika.parser.ParseContext; @@ -96,7 +96,7 @@ public class OpenAIEmbeddingFilter extends AbstractEmbeddingFilter { String responseBody = httpClient.postJson(url, requestJson, headers, config.getTimeoutMillis(), parseContext); - TikaProgressTracker.update(parseContext); + ParseTimeout.checkpoint(parseContext); parseResponse(responseBody, chunks); } diff --git a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIImageEmbeddingParser.java b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIImageEmbeddingParser.java index af8f62a823..f2f7322d22 100644 --- a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIImageEmbeddingParser.java +++ b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIImageEmbeddingParser.java @@ -41,7 +41,7 @@ import org.apache.tika.config.ConfigDeserializer; import org.apache.tika.config.Initializable; import org.apache.tika.config.JsonConfig; import org.apache.tika.config.ParseContextConfig; -import org.apache.tika.config.TikaProgressTracker; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; import org.apache.tika.http.TikaHttpClient; @@ -166,7 +166,7 @@ public class OpenAIImageEmbeddingParser implements Parser, Initializable, Closea String base64Data = Base64.getEncoder().encodeToString(imageBytes); float[] vector = callEmbeddingEndpoint(config, mimeType, base64Data, config.getTimeoutMillis(), parseContext); - TikaProgressTracker.update(parseContext); + ParseTimeout.checkpoint(parseContext); Locators locators = buildLocators(metadata); Chunk chunk = new Chunk(null, locators); diff --git a/tika-parsers/tika-parsers-ml/tika-parser-tess4j-module/src/main/java/org/apache/tika/parser/ocr/tess4j/Tess4JParser.java b/tika-parsers/tika-parsers-ml/tika-parser-tess4j-module/src/main/java/org/apache/tika/parser/ocr/tess4j/Tess4JParser.java index 074cb283f1..da32b168dd 100644 --- a/tika-parsers/tika-parsers-ml/tika-parser-tess4j-module/src/main/java/org/apache/tika/parser/ocr/tess4j/Tess4JParser.java +++ b/tika-parsers/tika-parsers-ml/tika-parser-tess4j-module/src/main/java/org/apache/tika/parser/ocr/tess4j/Tess4JParser.java @@ -29,6 +29,7 @@ import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import javax.imageio.ImageIO; @@ -46,7 +47,6 @@ import org.apache.tika.config.Initializable; import org.apache.tika.config.JsonConfig; import org.apache.tika.config.ParseContextConfig; import org.apache.tika.config.ParseTimeout; -import org.apache.tika.config.TikaProgressTracker; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; import org.apache.tika.exception.TikaTimeoutException; @@ -169,16 +169,17 @@ public class Tess4JParser implements Parser, Initializable { xhtml.startDocument(); Tesseract tesseract = null; - // Pessimistic default: doOCR is a blocking native call with no cancellation hook - // (see doOCRWithTimeout), so until it demonstrably finishes, this borrowed - // instance must not be handed back to another caller. - boolean tesseractStillBusy = true; + // True only while doOCRWithTimeout's background thread genuinely owns the instance. + // Starting false (rather than the old pessimistic true) means any exception during + // setup -- borrow, pixel-count check, image decode -- leaves this false, so the + // instance is returned promptly in the finally below instead of leaking; that setup + // work never touches a second thread, so there's nothing to still be "busy". + boolean tesseractStillBusy = false; long requestedMillis = config.getTimeoutMillis(); long timeoutMillis = ParseTimeout.getOrCreate(parseContext).budgetFor(requestedMillis); try { tesseract = borrowTesseract(parseContext, timeoutMillis); if (tesseract == null) { - tesseractStillBusy = false; throw new TikaTimeoutException("Timed out waiting for a Tesseract instance from the pool", requestedMillis, timeoutMillis); } @@ -195,7 +196,6 @@ public class Tess4JParser implements Parser, Initializable { if (pixels > maxPixels) { LOG.warn("Image has {} pixels, exceeding maxImagePixels={}. " + "Skipping OCR.", pixels, maxPixels); - tesseractStillBusy = false; xhtml.endDocument(); return; } @@ -207,14 +207,17 @@ public class Tess4JParser implements Parser, Initializable { BufferedImage image = readImage(tis); if (image == null) { LOG.warn("Could not read image from stream"); - tesseractStillBusy = false; xhtml.endDocument(); return; } + // From here on, a timeout from doOCRWithTimeout means the native call is still + // running on its own background thread -- see that method for how ownership of + // returning the instance to the pool is handed off in that case. + tesseractStillBusy = true; String ocrResult = doOCRWithTimeout(tesseract, image, parseContext); tesseractStillBusy = false; - TikaProgressTracker.update(parseContext); + ParseTimeout.checkpoint(parseContext); // Emit the text as XHTML AttributesImpl attrs = new AttributesImpl(); @@ -226,11 +229,13 @@ public class Tess4JParser implements Parser, Initializable { xhtml.endElement(XHTML, "div", "div"); } catch (TesseractException e) { + // doOCR itself completed (successfully or not) before we got here -- the + // instance is idle again, regardless of how doOCRWithTimeout got the exception + // to us -- so it's ours to return. tesseractStillBusy = false; throw new TikaException("Tess4J OCR failed", e); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - tesseractStillBusy = false; throw new TikaException("Interrupted while waiting for Tesseract instance", e); } finally { if (tesseract != null && !tesseractStillBusy) { @@ -340,10 +345,10 @@ public class Tess4JParser implements Parser, Initializable { */ private Tesseract borrowTesseract(ParseContext parseContext, long timeoutMillis) throws InterruptedException { - long now = System.currentTimeMillis(); - long deadline = (timeoutMillis >= Long.MAX_VALUE - now) ? Long.MAX_VALUE : now + timeoutMillis; + long startNanos = System.nanoTime(); while (true) { - long remaining = deadline - System.currentTimeMillis(); + long elapsedMillis = (System.nanoTime() - startNanos) / 1_000_000L; + long remaining = timeoutMillis - elapsedMillis; long pollMillis = remaining <= 0 ? 0 : Math.min(remaining, ProcessUtils.HEARTBEAT_INTERVAL_MILLIS); Tesseract tesseract = pool.poll(pollMillis, TimeUnit.MILLISECONDS); @@ -377,22 +382,34 @@ public class Tess4JParser implements Parser, Initializable { * failure mode this timeout model exists to prevent. * <p> * The native call itself cannot be cancelled once started (JNA/native calls don't - * respond to {@link Thread#interrupt()}): on timeout, this method gives up waiting - * and reports failure, but the background thread -- and the {@code tesseract} - * instance it's still using -- keeps running until the native call eventually - * returns on its own. The caller must not return {@code tesseract} to the pool after - * a timeout from this method (see {@code tesseractStillBusy} in {@link #parse}). + * respond to {@link Thread#interrupt()}): on timeout or interrupt, this method gives + * up waiting and throws, but the background thread -- and the {@code tesseract} + * instance it's still using -- keeps running until the native call eventually returns + * on its own. {@code settled} arbitrates who returns {@code tesseract} to the pool in + * that case: both this method (on giving up) and the worker thread (on finishing) + * race to flip it from {@code false} to {@code true}; whichever one loses the race -- + * i.e. finds it already {@code true} -- is the second to arrive and does the + * returning, so the instance goes back exactly once no matter how close the timing is. + * If this method returns normally, or throws {@link TesseractException}, the worker + * finished before either side touched {@code settled} (this method never gave up), so + * the caller retains ownership and returns {@code tesseract} itself, same as before + * this method was ever called. */ private String doOCRWithTimeout(Tesseract tesseract, BufferedImage image, ParseContext parseContext) throws TesseractException, TikaTimeoutException { long budgetMillis = ParseTimeout.getOrCreate(parseContext).remainingMillis(); if (budgetMillis <= 0) { + // Nothing async ever starts here, but the caller already flipped + // tesseractStillBusy to true before calling us, so it won't return the + // instance itself -- do it here or it's stuck in limbo forever. + returnTesseract(tesseract); throw new TikaTimeoutException("Tesseract OCR call not attempted", budgetMillis, budgetMillis); } AtomicReference<String> result = new AtomicReference<>(); AtomicReference<Throwable> failure = new AtomicReference<>(); CountDownLatch done = new CountDownLatch(1); + AtomicBoolean settled = new AtomicBoolean(false); Thread ocrThread = new Thread(() -> { try { result.set(tesseract.doOCR(image)); @@ -400,15 +417,19 @@ public class Tess4JParser implements Parser, Initializable { failure.set(t); } finally { done.countDown(); + if (!settled.compareAndSet(false, true)) { + // The waiter already gave up and is not coming back for this instance. + returnTesseract(tesseract); + } } }, "tess4j-ocr-worker"); ocrThread.setDaemon(true); ocrThread.start(); - long now = System.currentTimeMillis(); - long deadline = (budgetMillis >= Long.MAX_VALUE - now) ? Long.MAX_VALUE : now + budgetMillis; + long startNanos = System.nanoTime(); while (true) { - long remaining = deadline - System.currentTimeMillis(); + long elapsedMillis = (System.nanoTime() - startNanos) / 1_000_000L; + long remaining = budgetMillis - elapsedMillis; long waitMillis = remaining <= 0 ? 0 : Math.min(remaining, ProcessUtils.HEARTBEAT_INTERVAL_MILLIS); boolean finished; @@ -416,6 +437,9 @@ public class Tess4JParser implements Parser, Initializable { finished = done.await(waitMillis, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); + if (!settled.compareAndSet(false, true)) { + returnTesseract(tesseract); + } throw new TikaTimeoutException("interrupted while waiting for Tesseract OCR", budgetMillis, budgetMillis); } @@ -423,6 +447,9 @@ public class Tess4JParser implements Parser, Initializable { break; } if (remaining <= 0) { + if (!settled.compareAndSet(false, true)) { + returnTesseract(tesseract); + } throw new TikaTimeoutException("Tesseract OCR call timed out", budgetMillis, budgetMillis); } ParseTimeout.checkpoint(parseContext); diff --git a/tika-parsers/tika-parsers-ml/tika-parser-tess4j-module/src/test/java/org/apache/tika/parser/ocr/tess4j/Tess4JOCRTimeoutTest.java b/tika-parsers/tika-parsers-ml/tika-parser-tess4j-module/src/test/java/org/apache/tika/parser/ocr/tess4j/Tess4JOCRTimeoutTest.java index 4f9973ca99..9282b996e7 100644 --- a/tika-parsers/tika-parsers-ml/tika-parser-tess4j-module/src/test/java/org/apache/tika/parser/ocr/tess4j/Tess4JOCRTimeoutTest.java +++ b/tika-parsers/tika-parsers-ml/tika-parser-tess4j-module/src/test/java/org/apache/tika/parser/ocr/tess4j/Tess4JOCRTimeoutTest.java @@ -18,12 +18,20 @@ package org.apache.tika.parser.ocr.tess4j; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import java.awt.image.BufferedImage; +import java.io.IOException; +import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import net.sourceforge.tess4j.Tesseract; import net.sourceforge.tess4j.TesseractException; @@ -32,7 +40,10 @@ import org.junit.jupiter.api.Test; import org.apache.tika.config.ParseTimeout; import org.apache.tika.config.TimeoutLimits; import org.apache.tika.exception.TikaTimeoutException; +import org.apache.tika.io.TikaInputStream; +import org.apache.tika.metadata.Metadata; import org.apache.tika.parser.ParseContext; +import org.apache.tika.sax.BodyContentHandler; /** * TIKA-4813 follow-up: {@code Tess4JParser#doOCRWithTimeout} runs the blocking, native @@ -80,7 +91,6 @@ public class Tess4JOCRTimeoutTest { ParseContext context = new ParseContext(); context.set(TimeoutLimits.class, new TimeoutLimits(500, 500)); ParseTimeout parseTimeout = ParseTimeout.getOrCreate(context); - long initialProgress = parseTimeout.getLastProgressMillis(); BufferedImage image = new BufferedImage(1, 1, BufferedImage.TYPE_BYTE_GRAY); @@ -97,7 +107,9 @@ public class Tess4JOCRTimeoutTest { assertTrue(elapsed < 2500, "must time out near the 500ms budget, not wait for the 3s doOCR call to " + "finish naturally; took " + elapsed + "ms"); - assertTrue(parseTimeout.getLastProgressMillis() > initialProgress, + // If no checkpoint fired while waiting, millisSinceLastProgress() would be >= the + // full ~500-2500ms this test just spent; a mid-wait checkpoint resets it back down. + assertTrue(parseTimeout.millisSinceLastProgress() < elapsed, "expected at least one checkpoint to have fired while waiting"); } @@ -120,4 +132,126 @@ public class Tess4JOCRTimeoutTest { Object result = doOCRWithTimeout.invoke(parser, fast, image, context); assertEquals("fast OCR result", result); } + + private static void setPool(Tess4JParser parser, BlockingQueue<Tesseract> pool) throws Exception { + Field poolField = Tess4JParser.class.getDeclaredField("pool"); + poolField.setAccessible(true); + poolField.set(parser, pool); + } + + private static void setInitialized(Tess4JParser parser, boolean initialized) throws Exception { + Field initializedField = Tess4JParser.class.getDeclaredField("initialized"); + initializedField.setAccessible(true); + initializedField.set(parser, initialized); + } + + /** + * TIKA-4813 follow-up: {@code tesseractStillBusy} used to start {@code true} and only + * flip to {@code false} on a handful of enumerated paths, so any *other* exception + * between borrow and doOCR -- e.g. an ImageIO failure decoding the image -- leaked the + * pooled instance forever even though it was never handed to the native OCR call. + * <p> + * Forces {@code initialized=true} and seeds a single fake instance directly into the + * pool (bypassing {@code initialize()}'s native-library probe entirely) so this doesn't + * need a real Tesseract installation -- {@code doOCR} is never reached here, only + * {@code applyConfig}'s plain setters and the image-decode path are exercised. + */ + @Test + public void testImageIOFailureDoesNotShrinkThePool() throws Exception { + Tess4JParser parser = newParser(); + setInitialized(parser, true); + + Tesseract fake = new Tesseract() { + @Override + public String doOCR(BufferedImage bi) { + throw new AssertionError("doOCR must not be reached -- the image never decodes"); + } + }; + BlockingQueue<Tesseract> pool = new ArrayBlockingQueue<>(1); + pool.add(fake); + setPool(parser, pool); + + // A valid PNG signature followed immediately by a truncated IHDR chunk: enough for + // the PNG reader to commit to decoding (unlike unrecognizable bytes, which ImageIO + // just reports as "no reader found" -- a null return, not a thrown exception, and + // not what this test is after), but not enough data to finish, so it throws. + byte[] truncatedPng = { + (byte) 0x89, 'P', 'N', 'G', 0x0D, 0x0A, 0x1A, 0x0A, // PNG signature + 0x00, 0x00, 0x00, 0x0D, 'I', 'H', 'D', 'R' // IHDR chunk header, then EOF + }; + + BodyContentHandler handler = new BodyContentHandler(); + Metadata metadata = new Metadata(); + ParseContext context = new ParseContext(); + + boolean threw = false; + try (TikaInputStream tis = TikaInputStream.get(truncatedPng)) { + parser.parse(tis, handler, metadata, context); + } catch (IOException e) { + threw = true; + } + assertTrue(threw, "expected ImageIO to throw decoding the truncated PNG"); + + Tesseract returned = pool.poll(1, TimeUnit.SECONDS); + assertNotNull(returned, + "borrow must succeed immediately -- the instance was never handed to doOCR, " + + "so it must not have been withheld from the pool"); + assertSame(fake, returned); + } + + /** + * TIKA-4813 follow-up: after a doOCRWithTimeout timeout, the instance used to be leaked + * forever even though the background {@code tess4j-ocr-worker} thread eventually + * finishes the native call and could return it. Verifies the handoff: once the waiter + * gives up, the worker itself returns the instance to the pool when {@code doOCR} + * finally completes. + */ + @Test + public void testInstanceReturnedToPoolAfterSlowOCREventuallyCompletes() throws Exception { + Tess4JParser parser = newParser(); + Method doOCRWithTimeout = doOCRWithTimeoutMethod(); + + CountDownLatch releaseWorker = new CountDownLatch(1); + Tesseract slow = new Tesseract() { + @Override + public String doOCR(BufferedImage bi) throws TesseractException { + try { + assertTrue(releaseWorker.await(10, TimeUnit.SECONDS), + "test bug: releaseWorker was never counted down"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return "finished after the waiter already gave up"; + } + }; + + BlockingQueue<Tesseract> pool = new ArrayBlockingQueue<>(1); + setPool(parser, pool); + + ParseContext context = new ParseContext(); + context.set(TimeoutLimits.class, new TimeoutLimits(300, 300)); + BufferedImage image = new BufferedImage(1, 1, BufferedImage.TYPE_BYTE_GRAY); + + try { + doOCRWithTimeout.invoke(parser, slow, image, context); + fail("expected the 300ms budget to be exceeded by the still-blocked doOCR call"); + } catch (InvocationTargetException e) { + assertInstanceOf(TikaTimeoutException.class, e.getCause(), + "expected TikaTimeoutException, got: " + e.getCause()); + } + + // The waiter gave up, but the worker is still genuinely running (blocked on the + // latch) -- must not have been returned yet. + assertEquals(0, pool.size(), + "must not be returned while the OCR call is genuinely still in flight"); + + // Let the slow OCR call actually finish now. + releaseWorker.countDown(); + + Tesseract returned = pool.poll(5, TimeUnit.SECONDS); + assertNotNull(returned, + "the worker thread must return the instance itself once doOCR completes, " + + "since the waiter already abandoned it"); + assertSame(slow, returned); + } } diff --git a/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/AbstractVLMParser.java b/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/AbstractVLMParser.java index 6c291d67cb..20be659ea8 100644 --- a/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/AbstractVLMParser.java +++ b/tika-parsers/tika-parsers-ml/tika-vlm/src/main/java/org/apache/tika/parser/vlm/AbstractVLMParser.java @@ -33,7 +33,7 @@ import org.xml.sax.helpers.AttributesImpl; import org.apache.tika.config.Initializable; import org.apache.tika.config.ParseContextConfig; -import org.apache.tika.config.TikaProgressTracker; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; import org.apache.tika.extractor.ParentContentHandler; @@ -205,7 +205,7 @@ public abstract class AbstractVLMParser implements Parser, Initializable { String responseBody = httpClient.postJson( call.url(), call.json(), call.headers(), config.getTimeoutMillis(), parseContext); responseText = extractResponseText(responseBody, metadata); - TikaProgressTracker.update(parseContext); + ParseTimeout.checkpoint(parseContext); } catch (TikaException e) { throw e; } catch (IOException e) { diff --git a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-ocr-module/src/main/java/org/apache/tika/parser/ocr/TesseractOCRParser.java b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-ocr-module/src/main/java/org/apache/tika/parser/ocr/TesseractOCRParser.java index a21b2e83dd..e1fc83815a 100644 --- a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-ocr-module/src/main/java/org/apache/tika/parser/ocr/TesseractOCRParser.java +++ b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-ocr-module/src/main/java/org/apache/tika/parser/ocr/TesseractOCRParser.java @@ -63,7 +63,6 @@ import org.apache.tika.config.Initializable; import org.apache.tika.config.JsonConfig; import org.apache.tika.config.ParseContextConfig; import org.apache.tika.config.ParseTimeout; -import org.apache.tika.config.TikaProgressTracker; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; import org.apache.tika.exception.TikaTimeoutException; @@ -516,7 +515,7 @@ public class TesseractOCRParser extends AbstractExternalProcessParser implements process = pb.start(); id = register(process); runOCRProcess(process, parseContext, requestedMillis, timeoutMillis); - TikaProgressTracker.update(parseContext); + ParseTimeout.checkpoint(parseContext); } finally { if (process != null) { process.destroyForcibly(); diff --git a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-text-module/src/main/java/org/apache/tika/parser/strings/StringsParser.java b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-text-module/src/main/java/org/apache/tika/parser/strings/StringsParser.java index 7923fbbe3e..2be4aad602 100644 --- a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-text-module/src/main/java/org/apache/tika/parser/strings/StringsParser.java +++ b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-text-module/src/main/java/org/apache/tika/parser/strings/StringsParser.java @@ -37,7 +37,6 @@ import org.apache.tika.config.ConfigDeserializer; import org.apache.tika.config.Initializable; import org.apache.tika.config.JsonConfig; import org.apache.tika.config.ParseTimeout; -import org.apache.tika.config.TikaProgressTracker; import org.apache.tika.detect.FileCommandDetector; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; @@ -213,7 +212,7 @@ public class StringsParser implements Parser, Initializable { throw new TikaTimeoutException("strings process timed out", requestedMillis, timeoutMillis); } gobbler.join(10000); - TikaProgressTracker.update(context); + ParseTimeout.checkpoint(context); } catch (InterruptedException e) { throw new TikaException("strings process failed", e); } finally { diff --git a/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java b/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java index f3ddce13af..dc1040f348 100644 --- a/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java +++ b/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/PluginsWriter.java @@ -165,12 +165,12 @@ public class PluginsWriter { } // Write timeout limits to parse-context if configured on CLI - if (simpleAsyncConfig.getTimeoutMs() != null) { + if (simpleAsyncConfig.getTimeoutMillis() != null) { ObjectNode parseContext = root.has("parse-context") ? (ObjectNode) root.get("parse-context") : objectMapper.createObjectNode(); ObjectNode timeoutNode = objectMapper.createObjectNode(); - timeoutNode.put("progressTimeoutMillis", simpleAsyncConfig.getTimeoutMs()); + timeoutNode.put("progressTimeoutMillis", simpleAsyncConfig.getTimeoutMillis()); parseContext.set("timeout-limits", timeoutNode); root.set("parse-context", parseContext); } diff --git a/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/SimpleAsyncConfig.java b/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/SimpleAsyncConfig.java index f10788d89a..735fb9a3b8 100644 --- a/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/SimpleAsyncConfig.java +++ b/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/SimpleAsyncConfig.java @@ -29,7 +29,7 @@ class SimpleAsyncConfig { private String inputDir; private String outputDir; private Integer numClients; - private Long timeoutMs; + private Long timeoutMillis; private String xmx; private String fileList; private String tikaConfig;//path to the tikaConfig file to be used in the forked process @@ -51,14 +51,14 @@ class SimpleAsyncConfig { private String onExists; //TODO -- switch to a builder - public SimpleAsyncConfig(String inputDir, String outputDir, Integer numClients, Long timeoutMs, String xmx, String fileList, + public SimpleAsyncConfig(String inputDir, String outputDir, Integer numClients, Long timeoutMillis, String xmx, String fileList, String tikaConfig, BasicContentHandlerFactory.HANDLER_TYPE handlerType, ExtractBytesMode extractBytesMode, String pluginsDir) { - this(inputDir, outputDir, numClients, timeoutMs, xmx, fileList, tikaConfig, handlerType, + this(inputDir, outputDir, numClients, timeoutMillis, xmx, fileList, tikaConfig, handlerType, extractBytesMode, pluginsDir, false, false, null, null, false); } - public SimpleAsyncConfig(String inputDir, String outputDir, Integer numClients, Long timeoutMs, String xmx, String fileList, + public SimpleAsyncConfig(String inputDir, String outputDir, Integer numClients, Long timeoutMillis, String xmx, String fileList, String tikaConfig, BasicContentHandlerFactory.HANDLER_TYPE handlerType, ExtractBytesMode extractBytesMode, String pluginsDir, boolean concatenate, boolean contentOnly, @@ -66,7 +66,7 @@ class SimpleAsyncConfig { this.inputDir = inputDir; this.outputDir = outputDir; this.numClients = numClients; - this.timeoutMs = timeoutMs; + this.timeoutMillis = timeoutMillis; this.xmx = xmx; this.fileList = fileList; this.tikaConfig = tikaConfig; @@ -92,8 +92,8 @@ class SimpleAsyncConfig { return numClients; } - public Long getTimeoutMs() { - return timeoutMs; + public Long getTimeoutMillis() { + return timeoutMillis; } public String getXmx() { @@ -154,7 +154,7 @@ class SimpleAsyncConfig { "inputDir='" + inputDir + '\'' + ", outputDir='" + outputDir + '\'' + ", numClients=" + numClients + - ", timeoutMs=" + timeoutMs + + ", timeoutMillis=" + timeoutMillis + ", xmx='" + xmx + '\'' + ", fileList='" + fileList + '\'' + ", tikaConfig='" + tikaConfig + '\'' + diff --git a/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/TikaAsyncCLI.java b/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/TikaAsyncCLI.java index cd06371b20..7d2bd2dbb4 100644 --- a/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/TikaAsyncCLI.java +++ b/tika-pipes/tika-async-cli/src/main/java/org/apache/tika/async/cli/TikaAsyncCLI.java @@ -70,7 +70,7 @@ public class TikaAsyncCLI { options.addOption("n", "numClients", true, "number of forked clients"); options.addOption(null, "Xmx", true, "heap for the forked clients, e.g. --Xmx 1g"); options.addOption(null, "help", false, "this help message"); - options.addOption("T", "timeoutMs", true, "timeout for each parse in milliseconds"); + options.addOption("T", "timeoutMillis", true, "timeout for each parse in milliseconds"); options.addOption(null, "handler", true, "handler type: t=text, h=html, x=xml, m=markdown, b=body, i=ignore (default: m)"); options.addOption("p", "pluginsDir", true, "plugins directory"); options.addOption("l", "fileList", true, @@ -179,7 +179,7 @@ public class TikaAsyncCLI { String inputDir = null; String outputDir = null; String xmx = null; - Long timeoutMs = null; + Long timeoutMillis = null; Integer numClients = null; String fileList = null; String tikaConfig = null; @@ -197,7 +197,7 @@ public class TikaAsyncCLI { xmx = line.getOptionValue("Xmx"); } if (line.hasOption("T")) { - timeoutMs = Long.parseLong(line.getOptionValue("T")); + timeoutMillis = Long.parseLong(line.getOptionValue("T")); } if (line.hasOption("n")) { numClients = Integer.parseInt(line.getOptionValue("n")); @@ -299,7 +299,7 @@ public class TikaAsyncCLI { } SimpleAsyncConfig config = new SimpleAsyncConfig(inputDir, outputDir, - numClients, timeoutMs, xmx, fileList, tikaConfig, handlerType, + numClients, timeoutMillis, xmx, fileList, tikaConfig, handlerType, extractBytesMode, pluginsDir, concatenate, contentOnly, unpackFormat, unpackMode, unpackIncludeMetadata); config.setOnExists(onExists); diff --git a/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/AsyncCliParserTest.java b/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/AsyncCliParserTest.java index 975506989e..7c151e25e2 100644 --- a/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/AsyncCliParserTest.java +++ b/tika-pipes/tika-async-cli/src/test/java/org/apache/tika/async/cli/AsyncCliParserTest.java @@ -44,7 +44,7 @@ public class AsyncCliParserTest { assertEquals("output", simpleAsyncConfig.getOutputDir()); assertNull(simpleAsyncConfig.getFileList()); assertEquals(1, simpleAsyncConfig.getNumClients()); - assertEquals(30000L, simpleAsyncConfig.getTimeoutMs()); + assertEquals(30000L, simpleAsyncConfig.getTimeoutMillis()); assertEquals("-Xmx1g", simpleAsyncConfig.getXmx()); // TIKA-4663: default content handler is markdown assertEquals(BasicContentHandlerFactory.HANDLER_TYPE.MARKDOWN, simpleAsyncConfig.getHandlerType()); @@ -54,7 +54,7 @@ public class AsyncCliParserTest { assertEquals("output", simpleAsyncConfig.getOutputDir()); assertNull(simpleAsyncConfig.getFileList()); assertNull(simpleAsyncConfig.getNumClients()); - assertNull(simpleAsyncConfig.getTimeoutMs()); + assertNull(simpleAsyncConfig.getTimeoutMillis()); assertNull(simpleAsyncConfig.getXmx()); assertEquals(BasicContentHandlerFactory.HANDLER_TYPE.MARKDOWN, simpleAsyncConfig.getHandlerType()); @@ -63,7 +63,7 @@ public class AsyncCliParserTest { assertEquals("output", simpleAsyncConfig.getOutputDir()); assertNull(simpleAsyncConfig.getFileList()); assertNull(simpleAsyncConfig.getNumClients()); - assertNull(simpleAsyncConfig.getTimeoutMs()); + assertNull(simpleAsyncConfig.getTimeoutMillis()); assertNull(simpleAsyncConfig.getXmx()); simpleAsyncConfig = TikaAsyncCLI.parseCommandLine(new String[]{"--output", "output", "--input", "input"}); @@ -71,7 +71,7 @@ public class AsyncCliParserTest { assertEquals("output", simpleAsyncConfig.getOutputDir()); assertNull(simpleAsyncConfig.getFileList()); assertNull(simpleAsyncConfig.getNumClients()); - assertNull(simpleAsyncConfig.getTimeoutMs()); + assertNull(simpleAsyncConfig.getTimeoutMillis()); assertNull(simpleAsyncConfig.getXmx()); simpleAsyncConfig = TikaAsyncCLI.parseCommandLine(new String[]{"--output=output", "--input=input"}); @@ -79,7 +79,7 @@ public class AsyncCliParserTest { assertEquals("output", simpleAsyncConfig.getOutputDir()); assertNull(simpleAsyncConfig.getFileList()); assertNull(simpleAsyncConfig.getNumClients()); - assertNull(simpleAsyncConfig.getTimeoutMs()); + assertNull(simpleAsyncConfig.getTimeoutMillis()); assertNull(simpleAsyncConfig.getXmx()); } @@ -91,7 +91,7 @@ public class AsyncCliParserTest { assertEquals("output", simpleAsyncConfig.getOutputDir()); assertNull(simpleAsyncConfig.getFileList()); assertEquals(5, simpleAsyncConfig.getNumClients()); - assertEquals(30000L, simpleAsyncConfig.getTimeoutMs()); + assertEquals(30000L, simpleAsyncConfig.getTimeoutMillis()); assertEquals("1g", simpleAsyncConfig.getXmx()); assertEquals(BasicContentHandlerFactory.HANDLER_TYPE.XML, simpleAsyncConfig.getHandlerType()); } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java index 0424640a5b..1c8e01e662 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PerClientServerManager.java @@ -236,7 +236,7 @@ public class PerClientServerManager implements ServerManager { } @Override - public Socket connect(int socketTimeoutMs) throws IOException, ServerInitializationException { + public Socket connect(int socketTimeoutMillis) throws IOException, ServerInitializationException { if (serverSocket == null) { throw new IllegalStateException("Server not started. Call ensureRunning() first."); } @@ -248,7 +248,7 @@ public class PerClientServerManager implements ServerManager { while (true) { try { Socket socket = serverSocket.accept(); - socket.setSoTimeout(socketTimeoutMs); + socket.setSoTimeout(socketTimeoutMillis); socket.setTcpNoDelay(true); LOG.debug("clientId={}: accepted connection from server", clientId); return socket; 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 7c5f380408..1553099b7c 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 @@ -310,8 +310,8 @@ public class PipesClient implements Closeable { // Connect to server. Use the generous startup timeout as the read SO_TIMEOUT so the // server's post-connect initialization and READY handshake aren't bounded by the - // (possibly tight) per-request socketTimeoutMs. - Socket socket = serverManager.connect((int) pipesConfig.getStartupTimeoutMs()); + // (possibly tight) per-request socketTimeoutMillis. + Socket socket = serverManager.connect((int) pipesConfig.getStartupTimeoutMillis()); synchronized (connectionLock) { connectionTuple = new ConnectionTuple(socket, @@ -321,7 +321,7 @@ public class PipesClient implements Closeable { waitForStartup(); // Server is ready; subsequent reads use the normal per-request socket timeout. - socket.setSoTimeout((int) pipesConfig.getSocketTimeoutMs()); + socket.setSoTimeout((int) pipesConfig.getSocketTimeoutMillis()); } private void writeTask(FetchEmitTuple t) throws IOException { @@ -340,7 +340,7 @@ public class PipesClient implements Closeable { * 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 SO_TIMEOUT} ({@link PipesConfig#getSocketTimeoutMillis()}): 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. diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java index cbf44731e8..6dbbd18de5 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesConfig.java @@ -52,13 +52,13 @@ public class PipesConfig { public static final int DEFAULT_MAX_FILES_PROCESSED_PER_PROCESS = 10000; - public static final long DEFAULT_MAX_WAIT_FOR_CLIENT_MS = 60000; + public static final long DEFAULT_MAX_WAIT_FOR_CLIENT_MILLIS = 60000; - public static final long DEFAULT_SOCKET_TIMEOUT_MS = 60000; + public static final long DEFAULT_SOCKET_TIMEOUT_MILLIS = 60000; - public static final long DEFAULT_STARTUP_TIMEOUT_MS = 60000; + public static final long DEFAULT_STARTUP_TIMEOUT_MILLIS = 60000; - public static final long DEFAULT_HEARTBEAT_INTERVAL_MS = 1000; + public static final long DEFAULT_HEARTBEAT_INTERVAL_MILLIS = 1000; public static final boolean DEFAULT_USE_SHARED_SERVER = false; @@ -77,14 +77,14 @@ public class PipesConfig { private int maxIpcPayloadBytes = DEFAULT_MAX_IPC_PAYLOAD_BYTES; - private long socketTimeoutMs = DEFAULT_SOCKET_TIMEOUT_MS; - private long startupTimeoutMs = DEFAULT_STARTUP_TIMEOUT_MS; - private long heartbeatIntervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS; + private long socketTimeoutMillis = DEFAULT_SOCKET_TIMEOUT_MILLIS; + private long startupTimeoutMillis = DEFAULT_STARTUP_TIMEOUT_MILLIS; + private long heartbeatIntervalMillis = DEFAULT_HEARTBEAT_INTERVAL_MILLIS; private long shutdownClientAfterMillis = DEFAULT_SHUTDOWN_CLIENT_AFTER_MILLS; private int numClients = defaultNumClients(); - private long maxWaitForClientMillis = DEFAULT_MAX_WAIT_FOR_CLIENT_MS; + private long maxWaitForClientMillis = DEFAULT_MAX_WAIT_FOR_CLIENT_MILLIS; private int maxFilesProcessedPerProcess = DEFAULT_MAX_FILES_PROCESSED_PER_PROCESS; public static final int DEFAULT_STALE_FETCHER_TIMEOUT_SECONDS = 600; private int staleFetcherTimeoutSeconds = DEFAULT_STALE_FETCHER_TIMEOUT_SECONDS; @@ -152,7 +152,7 @@ public class PipesConfig { * This configuration is used by both PipesServer (forking process) and * AsyncProcessor (async processing). Some fields are specific to each: * <ul> - * <li>PipesServer uses: numClients, socketTimeoutMs, directEmitThresholdBytes, etc.</li> + * <li>PipesServer uses: numClients, socketTimeoutMillis, directEmitThresholdBytes, etc.</li> * <li>AsyncProcessor uses: emitWithinMillis, queueSize, numEmitters, etc.</li> * </ul> * Unused fields in each context are simply ignored. @@ -170,8 +170,8 @@ public class PipesConfig { return config; } - public long getSocketTimeoutMs() { - return socketTimeoutMs; + public long getSocketTimeoutMillis() { + return socketTimeoutMillis; } /** @@ -179,40 +179,40 @@ public class PipesConfig { * If no data is received within this time, the connection is considered timed out. * This is distinct from the parse/processing timeout, which lives on * {@link org.apache.tika.config.TimeoutLimits} under {@code parse-context.timeout-limits}. - * @param socketTimeoutMs + * @param socketTimeoutMillis */ - public void setSocketTimeoutMs(long socketTimeoutMs) { - this.socketTimeoutMs = socketTimeoutMs; + public void setSocketTimeoutMillis(long socketTimeoutMillis) { + this.socketTimeoutMillis = socketTimeoutMillis; } - public long getStartupTimeoutMs() { - return startupTimeoutMs; + public long getStartupTimeoutMillis() { + return startupTimeoutMillis; } /** * Timeout in milliseconds for the forked server to start up and send its READY handshake. - * Distinct from {@link #getSocketTimeoutMs()}: cold-starting the forked JVM (loading config, + * Distinct from {@link #getSocketTimeoutMillis()}: cold-starting the forked JVM (loading config, * parsers and plugins) can take far longer than a normal per-read timeout, so the handshake - * gets its own generous budget. Once the server is ready, reads switch to {@code socketTimeoutMs}. - * @param startupTimeoutMs + * gets its own generous budget. Once the server is ready, reads switch to {@code socketTimeoutMillis}. + * @param startupTimeoutMillis */ - public void setStartupTimeoutMs(long startupTimeoutMs) { - this.startupTimeoutMs = startupTimeoutMs; + public void setStartupTimeoutMillis(long startupTimeoutMillis) { + this.startupTimeoutMillis = startupTimeoutMillis; } - public long getHeartbeatIntervalMs() { - return heartbeatIntervalMs; + public long getHeartbeatIntervalMillis() { + return heartbeatIntervalMillis; } /** * Interval in milliseconds between heartbeat messages sent from server to client. - * Should be significantly less than socketTimeoutMs to ensure the client doesn't timeout. - * WARNING: Setting this >= socketTimeoutMs will cause socket timeouts during normal processing. + * Should be significantly less than socketTimeoutMillis to ensure the client doesn't timeout. + * WARNING: Setting this >= socketTimeoutMillis will cause socket timeouts during normal processing. * This only exists for testing. We encourage you never to use it. - * @param heartbeatIntervalMs + * @param heartbeatIntervalMillis */ - public void setHeartbeatIntervalMs(long heartbeatIntervalMs) { - this.heartbeatIntervalMs = heartbeatIntervalMs; + public void setHeartbeatIntervalMillis(long heartbeatIntervalMillis) { + this.heartbeatIntervalMillis = heartbeatIntervalMillis; } public long getShutdownClientAfterMillis() { diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java index 9a8ed6c5db..e84bfbd754 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ServerManager.java @@ -65,12 +65,12 @@ public interface ServerManager extends Closeable { * <p> * This method should be called after {@link #ensureRunning()}. * - * @param socketTimeoutMs the socket timeout in milliseconds + * @param socketTimeoutMillis the socket timeout in milliseconds * @return a connected Socket ready for communication * @throws IOException if connection fails * @throws ServerInitializationException if the server died before connecting */ - Socket connect(int socketTimeoutMs) throws IOException, ServerInitializationException; + Socket connect(int socketTimeoutMillis) throws IOException, ServerInitializationException; /** * Shuts down the server process and cleans up resources. diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java index 40b24cfb5d..50b03c3f18 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/SharedServerManager.java @@ -204,7 +204,7 @@ public class SharedServerManager implements ServerManager { } @Override - public Socket connect(int socketTimeoutMs) throws IOException, ServerInitializationException { + public Socket connect(int socketTimeoutMillis) throws IOException, ServerInitializationException { int port; byte[] token; synchronized (lock) { @@ -219,7 +219,7 @@ public class SharedServerManager implements ServerManager { Socket socket = new Socket(); try { socket.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), SOCKET_CONNECT_TIMEOUT_MS); - socket.setSoTimeout(socketTimeoutMs); + socket.setSoTimeout(socketTimeoutMillis); socket.setTcpNoDelay(true); // Send auth token before any protocol messages socket.getOutputStream().write(token); diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PipesMessage.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PipesMessage.java index a418990592..2f66bc0411 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PipesMessage.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/protocol/PipesMessage.java @@ -20,8 +20,6 @@ import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.EOFException; import java.io.IOException; -import java.nio.ByteBuffer; -import java.nio.ByteOrder; import java.util.Locale; /** @@ -140,16 +138,12 @@ public record PipesMessage(PipesMessageType type, byte[] payload) { } /** - * Creates a WORKING heartbeat with the last-progress timestamp in the payload. - * - * @param lastProgressMillis epoch millis of the last progress update + * Creates a WORKING heartbeat. Empty payload -- the client treats WORKING as a pure + * liveness signal (see {@code PipesClient.waitForServer}'s WORKING case) and never reads + * a timestamp back out of it. */ - public static PipesMessage working(long lastProgressMillis) { - byte[] payload = ByteBuffer.allocate(Long.BYTES) - .order(ByteOrder.BIG_ENDIAN) - .putLong(lastProgressMillis) - .array(); - return new PipesMessage(PipesMessageType.WORKING, payload); + public static PipesMessage working() { + return new PipesMessage(PipesMessageType.WORKING, EMPTY); } public static PipesMessage newRequest(byte[] payload) { @@ -171,18 +165,4 @@ public record PipesMessage(PipesMessageType type, byte[] payload) { public static PipesMessage crash(PipesMessageType crashType, byte[] payload) { return new PipesMessage(crashType, payload); } - - /** - * Extracts the last-progress timestamp from a WORKING message payload. - * - * @return epoch millis of the last progress update reported by the server - */ - public long lastProgressMillis() { - if (type != PipesMessageType.WORKING) { - throw new IllegalStateException("lastProgressMillis() only valid for WORKING messages"); - } - return ByteBuffer.wrap(payload) - .order(ByteOrder.BIG_ENDIAN) - .getLong(); - } } 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 68c3d06c9d..d49b038fdf 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 @@ -80,7 +80,7 @@ public class ConnectionHandler implements Runnable, Closeable { private final DataOutputStream output; private final SharedServerResources resources; private final PipesConfig pipesConfig; - private final long heartbeatIntervalMs; + private final long heartbeatIntervalMillis; private final ExecutorService executorService = Executors.newSingleThreadExecutor(); private final ExecutorCompletionService<PipesResult> executorCompletionService = @@ -105,7 +105,7 @@ public class ConnectionHandler implements Runnable, Closeable { this.output = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream())); this.resources = resources; this.pipesConfig = pipesConfig; - this.heartbeatIntervalMs = pipesConfig.getHeartbeatIntervalMs(); + this.heartbeatIntervalMillis = pipesConfig.getHeartbeatIntervalMillis(); this.protocolIO = new ServerProtocolIO(input, output); } @@ -286,9 +286,9 @@ public class ConnectionHandler implements Runnable, Closeable { // Send fire-and-forget heartbeat long elapsed = System.currentTimeMillis() - start.toEpochMilli(); - if (elapsed > heartbeatCounter * heartbeatIntervalMs) { + if (elapsed > heartbeatCounter * heartbeatIntervalMillis) { LOG.trace("handlerId={}: still processing, counter={}", handlerId, heartbeatCounter); - PipesMessage.working(parseTimeout.getLastProgressMillis()).write(output); + PipesMessage.working().write(output); heartbeatCounter++; } @@ -331,7 +331,7 @@ public class ConnectionHandler implements Runnable, Closeable { } private boolean checkProgressTimeout(ParseTimeout parseTimeout, long progressTimeoutMillis, String id) { - long timeSinceProgress = System.currentTimeMillis() - parseTimeout.getLastProgressMillis(); + long timeSinceProgress = parseTimeout.millisSinceLastProgress(); 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/PipesServer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java index e3d8f09e80..ad5fe40497 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 @@ -105,7 +105,7 @@ public class PipesServer implements AutoCloseable { * tell the difference between "I crashed" and "my parent went away". */ public static final int PARENT_GONE_EXIT_CODE = 23; - private final long heartbeatIntervalMs; + private final long heartbeatIntervalMillis; private final String pipesClientId; private Detector detector; @@ -144,7 +144,7 @@ public class PipesServer implements AutoCloseable { PipesConfig pipesConfig = PipesConfig.load(tikaJsonConfig); // Set socket timeout from config after loading PipesConfig - socket.setSoTimeout((int) pipesConfig.getSocketTimeoutMs()); + socket.setSoTimeout((int) pipesConfig.getSocketTimeoutMillis()); socket.setTcpNoDelay(true); MetadataFilter metadataFilter = tikaLoader.loadMetadataFilters(); @@ -187,7 +187,7 @@ public class PipesServer implements AutoCloseable { this.defaultMetadataWriteLimiterFactory = metadataWriteLimiterFactory; this.input = new DataInputStream(in); this.output = new DataOutputStream(out); - this.heartbeatIntervalMs = pipesConfig.getHeartbeatIntervalMs(); + this.heartbeatIntervalMillis = pipesConfig.getHeartbeatIntervalMillis(); validateHeartbeatInterval(pipesConfig); emitStrategy = pipesConfig.getEmitStrategy().getType(); @@ -234,18 +234,18 @@ public class PipesServer implements AutoCloseable { } /** - * Fails fast if heartbeatIntervalMs >= socketTimeoutMs: the client's liveness check + * Fails fast if heartbeatIntervalMillis >= socketTimeoutMillis: 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) { + long heartbeatIntervalMillis = pipesConfig.getHeartbeatIntervalMillis(); + long socketTimeoutMillis = pipesConfig.getSocketTimeoutMillis(); + if (heartbeatIntervalMillis >= socketTimeoutMillis) { 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); + heartbeatIntervalMillis, socketTimeoutMillis); // Allow override for testing only if (!"true".equals(System.getProperty("tika.pipes.allowInvalidHeartbeat"))) { @@ -292,7 +292,7 @@ public class PipesServer implements AutoCloseable { while (!Thread.currentThread().isInterrupted()) { try { java.net.Socket clientSocket = serverSocket.accept(); - clientSocket.setSoTimeout((int) pipesConfig.getSocketTimeoutMs()); + clientSocket.setSoTimeout((int) pipesConfig.getSocketTimeoutMillis()); clientSocket.setTcpNoDelay(true); // Validate auth token before creating handler @@ -483,8 +483,8 @@ 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(parseTimeout.getLastProgressMillis()).write(output); + if (elapsed > heartbeatCounter * heartbeatIntervalMillis) { + PipesMessage.working().write(output); heartbeatCounter++; } @@ -524,7 +524,7 @@ public class PipesServer implements AutoCloseable { } private boolean checkProgressTimeout(ParseTimeout parseTimeout, long progressTimeoutMillis, String id) { - long timeSinceProgress = System.currentTimeMillis() - parseTimeout.getLastProgressMillis(); + long timeSinceProgress = parseTimeout.millisSinceLastProgress(); if (timeSinceProgress > progressTimeoutMillis) { handleCrash(PipesMessageType.TIMEOUT, id, new RuntimeException("Server-side progress timeout: no progress for " + timeSinceProgress + "ms (limit: " + progressTimeoutMillis + "ms)")); @@ -566,7 +566,7 @@ public class PipesServer implements AutoCloseable { * self-terminates promptly if its parent disappears. Without this, an * orphaned PipesServer would only notice the parent is gone when the * next socket read fails -- which can take up to - * {@code socketTimeoutMs} (default 60s) and doesn't fire at all while + * {@code socketTimeoutMillis} (default 60s) and doesn't fire at all while * the server is mid-parse. {@code System.exit} here lets the * {@code AbstractExternalProcessParser} shutdown hook run, killing any * in-flight external subprocess (e.g. tesseract) cleanly. diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java index 34f8e44cb3..c925592b37 100644 --- a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/protocol/PipesMessageTest.java @@ -68,10 +68,10 @@ class PipesMessageTest { @Test void testWorkingMessageRoundTrip() throws IOException { - PipesMessage original = PipesMessage.working(42L); + PipesMessage original = PipesMessage.working(); PipesMessage roundTripped = roundTrip(original); assertEquals(PipesMessageType.WORKING, roundTripped.type()); - assertEquals(42L, roundTripped.lastProgressMillis()); + assertEquals(0, roundTripped.payload().length); } @Test 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 d6feaf0c0f..ad36ada247 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 @@ -407,7 +407,7 @@ public class PipesClientTest { @Test public void testSocketTimeout(@TempDir Path tmp) throws Exception { // Test socket timeout when heartbeats are sent too slowly - // Config has heartbeatIntervalMs=10000 (10 seconds) but socketTimeoutMs=3000 (3 seconds) + // Config has heartbeatIntervalMillis=10000 (10 seconds) but socketTimeoutMillis=3000 (3 seconds) // This simulates a server that appears unresponsive (different from parse timeout) // NOTE: This is an invalid configuration that would never be used in production, // but we allow it for testing via system property @@ -430,9 +430,9 @@ public class PipesClientTest { PipesConfig pipesConfig = PipesConfig.load(tikaJsonConfig); // Verify the misconfiguration that triggers socket timeout - assertEquals(3000, pipesConfig.getSocketTimeoutMs(), "Socket timeout should be 3 seconds"); - assertEquals(10000, pipesConfig.getHeartbeatIntervalMs(), "Heartbeat interval should be 10 seconds"); - assertTrue(pipesConfig.getHeartbeatIntervalMs() > pipesConfig.getSocketTimeoutMs(), + assertEquals(3000, pipesConfig.getSocketTimeoutMillis(), "Socket timeout should be 3 seconds"); + assertEquals(10000, pipesConfig.getHeartbeatIntervalMillis(), "Heartbeat interval should be 10 seconds"); + assertTrue(pipesConfig.getHeartbeatIntervalMillis() > pipesConfig.getSocketTimeoutMillis(), "Test requires heartbeat > socket timeout to trigger timeout"); // The config file includes -Dtika.pipes.allowInvalidHeartbeat=true in forkedJvmArgs @@ -447,8 +447,8 @@ public class PipesClientTest { PipesResult pipesResult = pipesClient.process(tuple); long elapsed = System.currentTimeMillis() - startTime; - // Should timeout due to socket timeout (no heartbeats received within socketTimeoutMs). - // Startup/handshake is bounded by startupTimeoutMs (not socketTimeoutMs), so a slow + // Should timeout due to socket timeout (no heartbeats received within socketTimeoutMillis). + // Startup/handshake is bounded by startupTimeoutMillis (not socketTimeoutMillis), so a slow // fork cold-start no longer misfires here as FAILED_TO_INITIALIZE. assertEquals(PipesResult.RESULT_STATUS.TIMEOUT, pipesResult.status(), "Should timeout when socket times out"); @@ -745,7 +745,7 @@ public class PipesClientTest { // Modify config to add very short heartbeat interval configContent = configContent.replace( "\"pipes\": {", - "\"pipes\": {\n \"heartbeatIntervalMs\": 100," + "\"pipes\": {\n \"heartbeatIntervalMillis\": 100," ); Files.writeString(tikaConfigPath, configContent, StandardCharsets.UTF_8); @@ -1003,7 +1003,7 @@ public class PipesClientTest { PipesMessage.ready().write(out); PipesMessage.read(in); // NEW_REQUEST -- ignored, this fake never parses anything while (!socket.isClosed()) { - PipesMessage.working(System.currentTimeMillis()).write(out); + PipesMessage.working().write(out); Thread.sleep(200); } } catch (Exception e) { @@ -1022,10 +1022,10 @@ public class PipesClientTest { } @Override - public Socket connect(int socketTimeoutMs) throws IOException { + public Socket connect(int socketTimeoutMillis) throws IOException { Socket socket = new Socket(); - socket.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), getPort()), socketTimeoutMs); - socket.setSoTimeout(socketTimeoutMs); + socket.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), getPort()), socketTimeoutMillis); + socket.setSoTimeout(socketTimeoutMillis); return socket; } diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-timeout-lt-heartbeat.json b/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-timeout-lt-heartbeat.json index 3133771d70..f4a2867777 100644 --- a/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-timeout-lt-heartbeat.json +++ b/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-timeout-lt-heartbeat.json @@ -18,8 +18,8 @@ }, "pipes": { "numClients": 1, - "socketTimeoutMs": 3000, - "heartbeatIntervalMs": 10000, + "socketTimeoutMillis": 3000, + "heartbeatIntervalMillis": 10000, "forkedJvmArgs": [ "-Xmx256m", "-Dtika.pipes.allowInvalidHeartbeat=true"
