This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4810-timeout-improvements in repository https://gitbox.apache.org/repos/asf/tika.git
commit a66d8679cb878b73ab867442e5ebf832a6b58310 Author: tallison <[email protected]> AuthorDate: Mon Aug 10 21:35:58 2026 -0400 TIKA-4813: unified in-process timeout model --- .../src/main/java/org/apache/tika/cli/TikaCLI.java | 5 +- .../java/org/apache/tika/config/ParseTimeout.java | 233 +++++++++++++++++++++ .../apache/tika/config/TikaProgressTracker.java | 8 + .../java/org/apache/tika/config/TimeoutLimits.java | 56 +++-- .../apache/tika/detect/FileCommandDetector.java | 6 +- .../exception/EmbeddedLimitReachedException.java | 30 ++- .../tika/exception/TikaTimeoutException.java | 71 ++++++- .../ParsingEmbeddedDocumentExtractor.java | 31 ++- .../apache/tika/metadata/TikaCoreProperties.java | 4 + .../org/apache/tika/parser/CompositeParser.java | 7 + .../java/org/apache/tika/parser/ParseRecord.java | 30 +++ .../tika/parser/external/ExternalParser.java | 7 +- .../sax/AbstractRecursiveParserWrapperHandler.java | 6 - .../org/apache/tika/utils/FileProcessResult.java | 39 ++++ .../java/org/apache/tika/utils/ProcessUtils.java | 146 ++++++++++++- .../org/apache/tika/config/ParseTimeoutTest.java | 222 ++++++++++++++++++++ .../tika/config/TikaProgressTrackerTest.java | 12 ++ ...arsingEmbeddedDocumentExtractorTimeoutTest.java | 171 +++++++++++++++ .../org/apache/tika/parser/mock/MockParser.java | 25 +++ .../org/apache/tika/utils/ProcessUtilsTest.java | 113 ++++++++++ .../apache/tika/detect/magika/MagikaDetector.java | 6 +- .../tika/detect/siegfried/SiegfriedDetector.java | 6 +- .../java/org/apache/tika/http/TikaHttpClient.java | 137 ++++++++++-- .../org/apache/tika/http/TikaHttpClientTest.java | 99 +++++++++ .../org/apache/tika/http/TikaTestHttpServer.java | 15 +- .../org/apache/tika/parser/gdal/GDALParser.java | 6 +- .../tika/inference/AbstractEmbeddingFilter.java | 15 +- .../tika/inference/OpenAIEmbeddingFilter.java | 7 +- .../tika/inference/OpenAIImageEmbeddingParser.java | 11 +- .../tika/parser/ocr/tess4j/Tess4JParser.java | 39 +++- .../apache/tika/parser/vlm/AbstractVLMParser.java | 7 +- .../tika/parser/RecursiveParserWrapperTest.java | 7 +- .../org/apache/tika/parser/dwg/DWGReadParser.java | 7 +- .../org/apache/tika/parser/dwg/DWGParserTest.java | 7 +- .../tika/parser/microsoft/libpst/LibPstParser.java | 7 +- .../apache/tika/parser/ocr/TesseractOCRParser.java | 17 +- .../tika/renderer/pdf/poppler/PopplerRenderer.java | 8 +- .../org/apache/tika/parser/pkg/UnrarParser.java | 9 +- .../apache/tika/parser/strings/StringsParser.java | 14 +- .../org/apache/tika/config/TimeoutLimitsTest.java | 23 +- 40 files changed, 1497 insertions(+), 172 deletions(-) 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 491add8b8a..4b1e3b0dae 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 @@ -728,9 +728,10 @@ public class TikaCLI { config.setParseMode(ParseMode.CONCATENATE); } - // Set timeout + // --fork-timeout maps to totalTaskTimeoutMillis; progressTimeoutMillis (stall + // detector) is a separate concept and keeps its own default. config.setTimeoutLimits(new TimeoutLimits( - TimeoutLimits.DEFAULT_TOTAL_TASK_TIMEOUT_MILLIS, forkTimeout)); + forkTimeout, TimeoutLimits.DEFAULT_PROGRESS_TIMEOUT_MILLIS)); // Set JVM args if provided if (forkJvmArgs != null && !forkJvmArgs.isEmpty()) { 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 new file mode 100644 index 0000000000..bcb77679d0 --- /dev/null +++ b/tika-core/src/main/java/org/apache/tika/config/ParseTimeout.java @@ -0,0 +1,233 @@ +/* + * 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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.tika.parser.ParseContext; + +/** + * Runtime timeout state for a parse task, shared with any embedded documents it recurses into. + * <p> + * One instance per top-level task, created from {@link TimeoutLimits} and looked up from the + * {@link ParseContext} at every nesting depth, so a budget request from an embedded document + * (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> + * Runtime-only state (not Serializable); never sent over the wire. + * <p> + * Two responsibilities: + * <ul> + * <li>{@link #budgetFor(long)} -- caps a requested budget at whatever remains of the task: + * {@code min(requested, remaining)}.</li> + * <li>{@link #checkpoint()} -- records progress. Bounded waits (e.g. + * {@link org.apache.tika.utils.ProcessUtils#execute}) checkpoint periodically so a long + * but legitimate external call isn't mistaken for a hang by the stall detector.</li> + * </ul> + * + * @since Apache Tika 4.0 + */ +public class ParseTimeout { + + private static final Logger LOG = LoggerFactory.getLogger(ParseTimeout.class); + + private final long startMillis; + private final long hardDeadlineMillis; + private final long progressTimeoutMillis; + private final AtomicLong lastProgressMillis; + + // 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; + this.progressTimeoutMillis = progressTimeoutMillis; + this.lastProgressMillis = new AtomicLong(startMillis); + } + + /** + * Starts a new timeout window anchored to now, using the total and progress + * timeouts from the given limits. + * <p> + * Rejects negative totals/progress (no coherent "less than no time"). Zero is + * accepted (e.g. a task resuming with none of its budget left). A progress timeout + * at or above a positive total is accepted but logged, since the stall detector + * could then never fire before the total deadline. + * + * @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) { + throw new IllegalArgumentException("totalTaskTimeoutMillis must not be negative, was " + total); + } + if (progress < 0) { + throw new IllegalArgumentException("progressTimeoutMillis must not be negative, was " + progress); + } + if (total > 0 && progress >= total) { + 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); + } + + /** + * Returns the ParseTimeout installed in the given context, creating and installing + * one (from {@link TimeoutLimits#get(ParseContext)}) if absent. Idempotent: the same + * instance is reused for every call with the same context, including nested + * embedded-document calls. + * + * @param context the ParseContext, may be null + * @return the task's ParseTimeout, or a detached default if context is null + */ + public static ParseTimeout getOrCreate(ParseContext context) { + if (context == null) { + return start(new TimeoutLimits()); + } + ParseTimeout timeout = context.get(ParseTimeout.class); + if (timeout == null) { + timeout = start(TimeoutLimits.get(context)); + context.set(ParseTimeout.class, timeout); + } + return timeout; + } + + /** + * Records a checkpoint on the ParseTimeout in the given context, if present. Unlike + * {@link #getOrCreate(ParseContext)}, this does not install one -- a checkpoint from + * code outside any tracked task is simply a no-op. + * + * @param context the ParseContext, may be null + */ + public static void checkpoint(ParseContext context) { + if (context == null) { + return; + } + ParseTimeout timeout = context.get(ParseTimeout.class); + if (timeout != null) { + timeout.checkpoint(); + } + } + + /** + * The single composition rule for nested timeouts: a requested budget is never + * granted more time than remains for the whole task. + * <p> + * Also the chokepoint for misconfiguration diagnostics -- every per-parser timeout + * flows through here, so validation lives once instead of per config class: + * <ul> + * <li>a non-positive request is treated as "unset" (falls back to remaining task + * time) instead of granting zero, which would fail instantly with no useful + * diagnostic;</li> + * <li>a request under one second is logged -- usually a seconds-vs-milliseconds + * mistake;</li> + * <li>a request larger than the task's original total is logged, since it can never + * be granted in full even at the task's start (unlike the ordinary case of being + * clipped by elapsed time, which is not logged).</li> + * </ul> + * Each logs at most once per task. + * + * @return {@code min(requestedMillis, remainingMillis())}, or just {@code remainingMillis()} + * if {@code requestedMillis} was non-positive + */ + public long budgetFor(long requestedMillis) { + if (requestedMillis <= 0) { + if (!warnedNonPositiveRequest.getAndSet(true)) { + LOG.warn("non-positive timeout requested ({}ms) -- treating as unset and using " + + "whatever remains of the task's total timeout instead", requestedMillis); + } + return remainingMillis(); + } + if (requestedMillis < 1000 && !warnedSubSecondRequest.getAndSet(true)) { + 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(); + 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", + requestedMillis, total); + } + return Math.min(requestedMillis, remainingMillis()); + } + + /** + * @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; + } + + /** + * @return milliseconds remaining before the task's total timeout, never negative + */ + public long remainingMillis() { + if (hardDeadlineMillis == Long.MAX_VALUE) { + return Long.MAX_VALUE; + } + return Math.max(0, hardDeadlineMillis - System.currentTimeMillis()); + } + + /** + * Records that progress happened. Never throws — cooperative cancellation + * on an exhausted deadline happens at embedded-document boundaries (see + * {@code ParseRecord}), not here. + */ + public void checkpoint() { + lastProgressMillis.set(System.currentTimeMillis()); + } + + /** + * @return epoch millis of the last checkpoint + */ + public long getLastProgressMillis() { + return lastProgressMillis.get(); + } + + /** + * @return the configured progress (stall-detection) timeout in milliseconds + */ + 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 index 2aa52a1a5e..abc76c382c 100644 --- a/tika-core/src/main/java/org/apache/tika/config/TikaProgressTracker.java +++ b/tika-core/src/main/java/org/apache/tika/config/TikaProgressTracker.java @@ -31,8 +31,14 @@ import org.apache.tika.parser.ParseContext; * {@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; @@ -62,6 +68,8 @@ public class TikaProgressTracker { if (tracker != null) { tracker.update(); } + // Delegate to the successor so old-API callers still feed stall detection/total-timeout accounting. + ParseTimeout.checkpoint(context); } /** 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 34fca0a13c..e0d07891bc 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 @@ -25,16 +25,19 @@ import org.apache.tika.parser.ParseContext; * Configuration for the two-tier task timeout system. * <p> * <ul> - * <li>{@code totalTaskTimeoutMillis} — bounds entire task wall-clock time + * <li>{@code totalTaskTimeoutMillis} — bounds entire task wall-clock time, including + * any embedded documents it recurses into * (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: 60,000 ms = 1 minute)</li> * </ul> * <p> - * Parsers that never call {@link TikaProgressTracker#update()} effectively get - * {@code progressTimeoutMillis} as their total timeout (same as the old single-timeout - * behavior). Parsers that <em>do</em> update progress can run up to - * {@code totalTaskTimeoutMillis}. + * These 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 + * {@code progressTimeoutMillis} raised to accommodate it — see + * {@link org.apache.tika.utils.ProcessUtils#execute}. * <p> * Example configuration: * <pre> @@ -60,6 +63,7 @@ public class TimeoutLimits implements Serializable { private long totalTaskTimeoutMillis = DEFAULT_TOTAL_TASK_TIMEOUT_MILLIS; private long progressTimeoutMillis = DEFAULT_PROGRESS_TIMEOUT_MILLIS; + private boolean throwOnDeadline = false; /** * No-arg constructor for Jackson deserialization. @@ -116,6 +120,19 @@ public class TimeoutLimits implements Serializable { this.progressTimeoutMillis = progressTimeoutMillis; } + /** + * Whether to throw when the task's total timeout is exhausted mid-parse, instead of + * skipping remaining embedded documents and returning content extracted so far. + * Default: {@code false}. + */ + public boolean isThrowOnDeadline() { + return throwOnDeadline; + } + + public void setThrowOnDeadline(boolean throwOnDeadline) { + this.throwOnDeadline = throwOnDeadline; + } + /** * Helper method to get TimeoutLimits from ParseContext with defaults. * @@ -130,34 +147,12 @@ public class TimeoutLimits implements Serializable { return limits != null ? limits : new TimeoutLimits(); } - /** - * Returns the per-process timeout to use for external process execution. - * <p> - * This checks for {@link TimeoutLimits} in the ParseContext and returns - * {@code max(0, progressTimeoutMillis - 100)} to give the monitoring loop - * a small window to detect the timeout before the process itself times out. - * Falls back to {@code defaultMs} if no TimeoutLimits is found. - * - * @param context the ParseContext (may be null) - * @param defaultMs default timeout if no TimeoutLimits in context - * @return timeout in milliseconds for external process execution - */ - public static long getProcessTimeoutMillis(ParseContext context, long defaultMs) { - if (context == null) { - return defaultMs; - } - TimeoutLimits limits = context.get(TimeoutLimits.class); - if (limits == null) { - return defaultMs; - } - return Math.max(0, limits.progressTimeoutMillis - 100); - } - @Override public String toString() { return "TimeoutLimits{" + "totalTaskTimeoutMillis=" + totalTaskTimeoutMillis + ", progressTimeoutMillis=" + progressTimeoutMillis + + ", throwOnDeadline=" + throwOnDeadline + '}'; } @@ -171,11 +166,12 @@ public class TimeoutLimits implements Serializable { } TimeoutLimits that = (TimeoutLimits) o; return totalTaskTimeoutMillis == that.totalTaskTimeoutMillis && - progressTimeoutMillis == that.progressTimeoutMillis; + progressTimeoutMillis == that.progressTimeoutMillis && + throwOnDeadline == that.throwOnDeadline; } @Override public int hashCode() { - return Objects.hash(totalTaskTimeoutMillis, progressTimeoutMillis); + return Objects.hash(totalTaskTimeoutMillis, progressTimeoutMillis, throwOnDeadline); } } diff --git a/tika-core/src/main/java/org/apache/tika/detect/FileCommandDetector.java b/tika-core/src/main/java/org/apache/tika/detect/FileCommandDetector.java index cec4104d07..73b4355d5b 100644 --- a/tika-core/src/main/java/org/apache/tika/detect/FileCommandDetector.java +++ b/tika-core/src/main/java/org/apache/tika/detect/FileCommandDetector.java @@ -99,16 +99,16 @@ public class FileCommandDetector implements Detector { return MediaType.OCTET_STREAM; } //spool the full file to disk, if there is no underlying file - return detectOnPath(tis.getPath(), metadata); + return detectOnPath(tis.getPath(), metadata, parseContext); } - private MediaType detectOnPath(Path path, Metadata metadata) throws IOException { + private MediaType detectOnPath(Path path, Metadata metadata, ParseContext parseContext) throws IOException { String[] args = new String[]{ProcessUtils.escapeCommandLine(fileCommandPath), "-b", "--mime-type", ProcessUtils.escapeCommandLine(path.toAbsolutePath().toString())}; ProcessBuilder builder = new ProcessBuilder(args); - FileProcessResult result = ProcessUtils.execute(builder, timeoutMs, 10000, 10000); + FileProcessResult result = ProcessUtils.execute(builder, parseContext, timeoutMs, 10000, 10000); if (result.isTimeout()) { metadata.set(ExternalProcess.IS_TIMEOUT, true); return MediaType.OCTET_STREAM; 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 40f571db71..d68dac0e58 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 @@ -29,24 +29,36 @@ public class EmbeddedLimitReachedException extends RuntimeException { public enum LimitType { MAX_DEPTH, - MAX_COUNT + MAX_COUNT, + /** The task's total timeout was exhausted; see {@code ParseRecord#isTaskDeadlineReached()}. */ + DEADLINE } private final LimitType limitType; - private final int limit; + 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 + */ + public EmbeddedLimitReachedException(LimitType limitType, long limit) { super(buildMessage(limitType, limit)); this.limitType = limitType; this.limit = limit; } - private static String buildMessage(LimitType limitType, int limit) { + private static String buildMessage(LimitType limitType, long limit) { switch (limitType) { case MAX_DEPTH: return "Max embedded depth reached: " + limit; case MAX_COUNT: return "Max embedded count reached: " + limit; + case DEADLINE: + return "Task deadline reached: totalTaskTimeoutMillis=" + limit; default: return "Embedded limit reached: " + limit; } @@ -56,7 +68,19 @@ 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() { return limit; } } diff --git a/tika-core/src/main/java/org/apache/tika/exception/TikaTimeoutException.java b/tika-core/src/main/java/org/apache/tika/exception/TikaTimeoutException.java index a53dbd6a31..bb5ffc2159 100644 --- a/tika-core/src/main/java/org/apache/tika/exception/TikaTimeoutException.java +++ b/tika-core/src/main/java/org/apache/tika/exception/TikaTimeoutException.java @@ -17,10 +17,75 @@ package org.apache.tika.exception; /** - * Runtime/unchecked version of {@link java.util.concurrent.TimeoutException} + * Thrown when a single operation (external process, HTTP call, pool borrow, etc.) + * exceeds its allotted timeout. + * <p> + * Checked, not a {@link RuntimeException}: expected to be caught at the nearest + * embedded-document boundary (see {@code ParsingEmbeddedDocumentExtractor}), recorded, + * and parsing of remaining siblings continued. This differs from the task's total + * deadline being exhausted, which does not throw at all -- see {@code ParseRecord}'s + * {@code taskDeadlineReached} handling. + * <p> + * When the caller got its budget via {@code ParseTimeout.budgetFor(long)}, use + * {@link #TikaTimeoutException(String, long, long)} so the message states both the + * requested and granted budget: {@code granted == requested} means the operation's own + * timeout was binding; {@code granted < requested} means the task's total timeout was + * binding and raising the operation's timeout won't help. + * + * @since Apache Tika 4.0 */ -public class TikaTimeoutException extends RuntimeException { +public class TikaTimeoutException extends TikaException { + + private static final long UNKNOWN = -1; + + private final long requestedMillis; + private final long grantedMillis; + public TikaTimeoutException(String message) { - super(message); + this(message, UNKNOWN, UNKNOWN); + } + + /** + * @param requestedMillis the timeout the caller's own configuration asked for + * @param grantedMillis the budget actually granted, e.g. by {@code ParseTimeout.budgetFor} + */ + public TikaTimeoutException(String message, long requestedMillis, long grantedMillis) { + super(buildMessage(message, requestedMillis, grantedMillis)); + this.requestedMillis = requestedMillis; + this.grantedMillis = grantedMillis; + } + + private static String buildMessage(String message, long requestedMillis, long grantedMillis) { + if (requestedMillis == UNKNOWN || grantedMillis == UNKNOWN) { + return message; + } + String clippedBy = grantedMillis < requestedMillis ? " (task remaining)" : " -- budget exhausted"; + return message + ": requested=" + requestedMillis + "ms, granted=" + grantedMillis + "ms" + clippedBy; + } + + /** + * @return the timeout the caller's own configuration requested, or {@code -1} if + * this exception was not constructed with that information + */ + public long getRequestedMillis() { + return requestedMillis; + } + + /** + * @return the budget actually granted, or {@code -1} if this exception was not + * constructed with that information + */ + public long getGrantedMillis() { + return grantedMillis; + } + + /** + * @return true if the granted budget was clipped below the requested timeout by the + * task's remaining time (the task's total timeout was binding, not the operation's + * own). False -- including when no requested/granted info was recorded -- means the + * operation's own timeout was binding. + */ + public boolean isClippedByRemaining() { + return requestedMillis != UNKNOWN && grantedMillis != UNKNOWN && grantedMillis < requestedMillis; } } 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 194d863d2d..1260c8404b 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 @@ -26,6 +26,7 @@ import org.xml.sax.ContentHandler; import org.xml.sax.SAXException; import org.xml.sax.helpers.AttributesImpl; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.CorruptedFileException; import org.apache.tika.exception.EmbeddedLimitReachedException; import org.apache.tika.exception.EncryptedDocumentException; @@ -87,15 +88,16 @@ public class ParsingEmbeddedDocumentExtractor implements EmbeddedDocumentExtract } /** - * Checks embedded document limits from ParseRecord. + * Checks embedded document limits from ParseRecord: the task deadline, then max + * count, then max depth. * <p> - * If throwOnMaxDepth or throwOnMaxCount is configured and the respective limit is hit, - * an EmbeddedLimitReachedException is thrown. Otherwise, returns false and sets the + * If throwing is configured for the limit hit, the corresponding + * EmbeddedLimitReachedException is thrown. Otherwise, returns false and sets the * appropriate limit flag on the ParseRecord. * <p> - * Note: The count limit is a hard stop (once hit, no more embedded docs are parsed). - * The depth limit only affects documents at that depth - sibling documents at - * shallower depths will still be parsed. + * Note: The deadline and count limits are hard stops (once hit, no more embedded + * docs are parsed). The depth limit only affects documents at that depth - sibling + * documents at shallower depths will still be parsed. * <p> * Subclasses that override parseEmbedded() should call this method to enforce limits. * @@ -104,6 +106,23 @@ public class ParsingEmbeddedDocumentExtractor implements EmbeddedDocumentExtract * @throws EmbeddedLimitReachedException if a limit is exceeded and throwing is configured */ protected boolean checkEmbeddedLimits(ParseRecord parseRecord) { + // Deadline is checked first: once the task is out of time, count/depth don't matter. + // Unlike a single embedded doc timing out (TikaTimeoutException, recorded, siblings + // continued -- see parseEmbedded's catch(TikaException) below), this is a + // document-level fact, not an exception path by default. + if (parseRecord.isTaskDeadlineReached()) { + return false; + } + ParseTimeout timeout = context.get(ParseTimeout.class); + if (timeout != null && timeout.remainingMillis() <= 0) { + parseRecord.setTaskDeadlineReached(true); + if (parseRecord.isThrowOnDeadline()) { + throw new EmbeddedLimitReachedException( + EmbeddedLimitReachedException.LimitType.DEADLINE, timeout.getHardDeadlineMillis() - timeout.getStartMillis()); + } + return false; + } + // Count limit is a hard stop - once we've hit max, no more embedded parsing if (parseRecord.isEmbeddedCountLimitReached()) { return false; diff --git a/tika-core/src/main/java/org/apache/tika/metadata/TikaCoreProperties.java b/tika-core/src/main/java/org/apache/tika/metadata/TikaCoreProperties.java index a977d8c377..78cd5fda35 100644 --- a/tika-core/src/main/java/org/apache/tika/metadata/TikaCoreProperties.java +++ b/tika-core/src/main/java/org/apache/tika/metadata/TikaCoreProperties.java @@ -167,6 +167,10 @@ public interface TikaCoreProperties { Property EMBEDDED_DEPTH_LIMIT_REACHED = Property.internalBoolean(TIKA_META_EXCEPTION_PREFIX + "embedded-depth-limit-reached"); + //total timeout exhausted mid-parse; remaining embedded docs were skipped, not attempted + Property TASK_DEADLINE_REACHED = + Property.internalBoolean(TIKA_META_EXCEPTION_PREFIX + "task-deadline-reached"); + /** * Use this to store exceptions caught during a parse that are * non-fatal, e.g. if a parser is in lenient mode and more diff --git a/tika-core/src/main/java/org/apache/tika/parser/CompositeParser.java b/tika-core/src/main/java/org/apache/tika/parser/CompositeParser.java index 5e744518ba..fc98af7921 100644 --- a/tika-core/src/main/java/org/apache/tika/parser/CompositeParser.java +++ b/tika-core/src/main/java/org/apache/tika/parser/CompositeParser.java @@ -29,6 +29,7 @@ import java.util.Set; import org.xml.sax.ContentHandler; import org.xml.sax.SAXException; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.TikaException; import org.apache.tika.exception.WriteLimitReachedException; import org.apache.tika.io.TikaInputStream; @@ -287,6 +288,9 @@ public class CompositeParser implements Parser { parserRecord = ParseRecord.newInstance(context); context.set(ParseRecord.class, parserRecord); } + // Installed once per top-level task; embedded/recursive parses share the same + // ParseContext, so remaining budget is one pool across the whole task. + ParseTimeout.getOrCreate(context); try { TaggedContentHandler taggedHandler = handler != null ? new TaggedContentHandler(handler) : null; @@ -341,6 +345,9 @@ public class CompositeParser implements Parser { if (record.isEmbeddedDepthLimitReached()) { metadata.set(TikaCoreProperties.EMBEDDED_DEPTH_LIMIT_REACHED, true); } + if (record.isTaskDeadlineReached()) { + metadata.set(TikaCoreProperties.TASK_DEADLINE_REACHED, true); + } for (Metadata m : record.getMetadataList()) { for (String n : m.names()) { diff --git a/tika-core/src/main/java/org/apache/tika/parser/ParseRecord.java b/tika-core/src/main/java/org/apache/tika/parser/ParseRecord.java index 2cf4218c97..01a8fd4cbf 100644 --- a/tika-core/src/main/java/org/apache/tika/parser/ParseRecord.java +++ b/tika-core/src/main/java/org/apache/tika/parser/ParseRecord.java @@ -22,6 +22,7 @@ import java.util.List; import java.util.Set; import org.apache.tika.config.EmbeddedLimits; +import org.apache.tika.config.TimeoutLimits; import org.apache.tika.metadata.Metadata; /** @@ -64,6 +65,8 @@ public class ParseRecord { private boolean throwOnMaxCount = false; private boolean embeddedDepthLimitReached = false; private boolean embeddedCountLimitReached = false; + private boolean throwOnDeadline = false; + private boolean taskDeadlineReached = false; /** * Creates a new ParseRecord configured from EmbeddedLimits in the ParseContext. @@ -81,6 +84,7 @@ public class ParseRecord { record.maxEmbeddedCount = limits.getMaxCount(); record.throwOnMaxDepth = limits.isThrowOnMaxDepth(); record.throwOnMaxCount = limits.isThrowOnMaxCount(); + record.throwOnDeadline = TimeoutLimits.get(context).isThrowOnDeadline(); return record; } @@ -255,4 +259,30 @@ public class ParseRecord { public boolean isEmbeddedCountLimitReached() { return embeddedCountLimitReached; } + + /** + * Sets whether an exception should be thrown when the task's total timeout is + * exhausted, rather than skipping remaining embedded documents cleanly. + */ + public void setThrowOnDeadline(boolean throwOnDeadline) { + this.throwOnDeadline = throwOnDeadline; + } + + public boolean isThrowOnDeadline() { + return throwOnDeadline; + } + + /** + * Sets the flag indicating the task's total timeout was exhausted -- a document-level + * fact (unlike a single embedded document timing out, recorded per-child via + * {@link #addException(Exception)}): once set, any not-yet-started embedded document + * is skipped rather than attempted. + */ + public void setTaskDeadlineReached(boolean taskDeadlineReached) { + this.taskDeadlineReached = taskDeadlineReached; + } + + public boolean isTaskDeadlineReached() { + return taskDeadlineReached; + } } 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 b6b9935fce..9dfcafc372 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 @@ -38,7 +38,6 @@ 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.TimeoutLimits; import org.apache.tika.exception.TikaException; import org.apache.tika.io.TemporaryResources; import org.apache.tika.io.TikaInputStream; @@ -187,11 +186,9 @@ public class ExternalParser implements Parser { } // Always capture both stdout and stderr in memory - long localTimeoutMillis = TimeoutLimits.getProcessTimeoutMillis( - context, config.getTimeoutMs()); FileProcessResult result = ProcessUtils.execute( - new ProcessBuilder(thisCommandLine), - localTimeoutMillis, config.getMaxStdOut(), config.getMaxStdErr()); + new ProcessBuilder(thisCommandLine), context, + config.getTimeoutMs(), config.getMaxStdOut(), config.getMaxStdErr()); // Set process metadata metadata.set(ExternalProcess.IS_TIMEOUT, result.isTimeout()); diff --git a/tika-core/src/main/java/org/apache/tika/sax/AbstractRecursiveParserWrapperHandler.java b/tika-core/src/main/java/org/apache/tika/sax/AbstractRecursiveParserWrapperHandler.java index 7290cfa19e..ac6c8d4eed 100644 --- a/tika-core/src/main/java/org/apache/tika/sax/AbstractRecursiveParserWrapperHandler.java +++ b/tika-core/src/main/java/org/apache/tika/sax/AbstractRecursiveParserWrapperHandler.java @@ -23,7 +23,6 @@ import org.xml.sax.SAXException; import org.xml.sax.helpers.DefaultHandler; import org.apache.tika.metadata.Metadata; -import org.apache.tika.metadata.Property; import org.apache.tika.metadata.TikaCoreProperties; /** @@ -35,11 +34,6 @@ import org.apache.tika.metadata.TikaCoreProperties; public abstract class AbstractRecursiveParserWrapperHandler extends DefaultHandler implements Serializable { - // Canonical definitions live in TikaCoreProperties; alias here so the key can never diverge. - public final static Property EMBEDDED_RESOURCE_LIMIT_REACHED = - TikaCoreProperties.EMBEDDED_RESOURCE_LIMIT_REACHED; - public final static Property EMBEDDED_DEPTH_LIMIT_REACHED = - TikaCoreProperties.EMBEDDED_DEPTH_LIMIT_REACHED; private static final int MAX_DEPTH = 100; private final ContentHandlerFactory contentHandlerFactory; private int embeddedDepth = 0; diff --git a/tika-core/src/main/java/org/apache/tika/utils/FileProcessResult.java b/tika-core/src/main/java/org/apache/tika/utils/FileProcessResult.java index f08ca472c8..1a24da73e5 100644 --- a/tika-core/src/main/java/org/apache/tika/utils/FileProcessResult.java +++ b/tika-core/src/main/java/org/apache/tika/utils/FileProcessResult.java @@ -27,6 +27,8 @@ public class FileProcessResult { long stderrLength = -1; boolean stderrTruncated = false; boolean stdoutTruncated = false; + long requestedTimeoutMillis = -1; + long grantedTimeoutMillis = -1; public String getStderr() { return stderr; @@ -100,6 +102,41 @@ public class FileProcessResult { this.stdoutTruncated = stdoutTruncated; } + /** + * @return the timeout the caller's own configuration requested, or {@code -1} if + * this result was not produced by a context-aware {@code ProcessUtils.execute} call + */ + public long getRequestedTimeoutMillis() { + return requestedTimeoutMillis; + } + + public void setRequestedTimeoutMillis(long requestedTimeoutMillis) { + this.requestedTimeoutMillis = requestedTimeoutMillis; + } + + /** + * @return the budget actually granted (after {@code ParseTimeout.budgetFor} + * clipping), or {@code -1} if this result was not produced by a context-aware + * {@code ProcessUtils.execute} call + */ + public long getGrantedTimeoutMillis() { + return grantedTimeoutMillis; + } + + public void setGrantedTimeoutMillis(long grantedTimeoutMillis) { + this.grantedTimeoutMillis = grantedTimeoutMillis; + } + + /** + * @return true if the granted budget was clipped below the requested timeout by the + * task's remaining time (task's total timeout was binding, not the process's own). + * False if not clipped, or if no requested/granted info is present. + */ + public boolean isClippedByRemaining() { + return requestedTimeoutMillis >= 0 && grantedTimeoutMillis >= 0 + && grantedTimeoutMillis < requestedTimeoutMillis; + } + @Override public String toString() { return "FileProcessResult{" + @@ -112,6 +149,8 @@ public class FileProcessResult { ", stderrLength=" + stderrLength + ", stderrTruncated=" + stderrTruncated + ", stdoutTruncated=" + stdoutTruncated + + ", requestedTimeoutMillis=" + requestedTimeoutMillis + + ", grantedTimeoutMillis=" + grantedTimeoutMillis + '}'; } } 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 eb983ec7de..f9bb4e4dac 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 @@ -28,10 +28,22 @@ import java.util.concurrent.TimeoutException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.tika.config.ParseTimeout; +import org.apache.tika.parser.ParseContext; + public class ProcessUtils { private static final Logger LOG = LoggerFactory.getLogger(ProcessUtils.class); + // How often a bounded subprocess wait checkpoints the task's ParseTimeout -- must be + // finer-grained than progressTimeoutMillis or a long wait looks like a false "hung" kill. + public static final long HEARTBEAT_INTERVAL_MILLIS = 1000; + + // Timeout for checkCommand's binary-existence probe (e.g. "myapp --version") -- kept + // short since an unresponsive binary isn't usable and a full minute would delay + // every task that happens to probe first. + public static final long DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS = 5000; + private static final ConcurrentHashMap<String, Process> PROCESS_MAP = new ConcurrentHashMap<>(); static { @@ -81,18 +93,52 @@ public class ProcessUtils { /** * This writes stdout and stderr to the FileProcessResult. + * <p> + * Equivalent to {@link #execute(ProcessBuilder, ParseContext, long, int, int)} with a + * null context: {@code requestedTimeoutMillis} is granted unclipped, no checkpointing. * * @param pb - * @param timeoutMillis + * @param requestedTimeoutMillis * @param maxStdoutBuffer * @param maxStdErrBuffer * @return * @throws IOException */ public static FileProcessResult execute(ProcessBuilder pb, - long timeoutMillis, + long requestedTimeoutMillis, int maxStdoutBuffer, int maxStdErrBuffer) throws IOException { + return execute(pb, null, requestedTimeoutMillis, maxStdoutBuffer, maxStdErrBuffer); + } + + /** + * Same as {@link #execute(ProcessBuilder, long, int, int)}, but bounds the wait to + * {@code min(requestedTimeoutMillis, ParseTimeout.remainingMillis())} (see + * {@link ParseTimeout#budgetFor(long)}) so no single call can outlast the task's + * total timeout regardless of its own configuration. The granted budget and original + * request are both recorded on the result (see + * {@link FileProcessResult#getRequestedTimeoutMillis()}, + * {@link FileProcessResult#isClippedByRemaining()}). + * <p> + * While waiting, checkpoints the {@link ParseTimeout} in {@code context} (if any) + * every {@value #HEARTBEAT_INTERVAL_MILLIS} ms, so a bounded external call can run + * longer than the progress (stall-detection) timeout without looking like a hang -- + * the wait itself is progress. A null {@code context} behaves like the + * four-argument overload. + * + * @param pb + * @param context may be null + * @param requestedTimeoutMillis the timeout the caller's own configuration asks for + * @param maxStdoutBuffer + * @param maxStdErrBuffer + * @return + * @throws IOException + */ + public static FileProcessResult execute(ProcessBuilder pb, ParseContext context, + long requestedTimeoutMillis, + int maxStdoutBuffer, int maxStdErrBuffer) + throws IOException { + long grantedTimeoutMillis = ParseTimeout.getOrCreate(context).budgetFor(requestedTimeoutMillis); Process p = null; String id = null; try { @@ -111,7 +157,7 @@ public class ProcessUtils { int exitValue = -1; boolean complete = false; try { - complete = p.waitFor(timeoutMillis, TimeUnit.MILLISECONDS); + complete = waitForWithHeartbeat(p, context, grantedTimeoutMillis); elapsed = System.currentTimeMillis() - start; if (complete) { exitValue = p.exitValue(); @@ -146,6 +192,8 @@ public class ProcessUtils { result.stderr = StringUtils.joinWith("\n", errGobbler.getLines()); result.stdoutTruncated = outGobbler.getIsTruncated(); result.stderrTruncated = errGobbler.getIsTruncated(); + result.requestedTimeoutMillis = requestedTimeoutMillis; + result.grantedTimeoutMillis = grantedTimeoutMillis; return result; } finally { if (p != null) { @@ -159,22 +207,46 @@ public class ProcessUtils { /** * This redirects stdout to stdoutRedirect path. + * <p> + * Equivalent to {@link #execute(ProcessBuilder, ParseContext, long, Path, int)} with + * a null context, i.e. {@code requestedTimeoutMillis} is granted unclipped and the + * wait does not checkpoint any task's progress timeout. * * @param pb - * @param timeoutMillis + * @param requestedTimeoutMillis * @param stdoutRedirect * @param maxStdErrBuffer * @return * @throws IOException */ public static FileProcessResult execute(ProcessBuilder pb, - long timeoutMillis, + long requestedTimeoutMillis, + Path stdoutRedirect, int maxStdErrBuffer) throws IOException { + return execute(pb, null, requestedTimeoutMillis, stdoutRedirect, maxStdErrBuffer); + } + + /** + * Same as {@link #execute(ProcessBuilder, long, Path, int)}, but bounds the wait to + * {@code min(requestedTimeoutMillis, ParseTimeout.remainingMillis())} and checkpoints + * while waiting -- see {@link #execute(ProcessBuilder, ParseContext, long, int, int)}. + * + * @param pb + * @param context may be null + * @param requestedTimeoutMillis the timeout the caller's own configuration asks for + * @param stdoutRedirect + * @param maxStdErrBuffer + * @return + * @throws IOException + */ + public static FileProcessResult execute(ProcessBuilder pb, ParseContext context, + long requestedTimeoutMillis, Path stdoutRedirect, int maxStdErrBuffer) throws IOException { if (!Files.isDirectory(stdoutRedirect.getParent())) { Files.createDirectories(stdoutRedirect.getParent()); } + long grantedTimeoutMillis = ParseTimeout.getOrCreate(context).budgetFor(requestedTimeoutMillis); pb.redirectOutput(stdoutRedirect.toFile()); Process p = null; String id = null; @@ -190,7 +262,7 @@ public class ProcessUtils { int exitValue = -1; boolean complete = false; try { - complete = p.waitFor(timeoutMillis, TimeUnit.MILLISECONDS); + complete = waitForWithHeartbeat(p, context, grantedTimeoutMillis); elapsed = System.currentTimeMillis() - start; if (complete) { exitValue = p.exitValue(); @@ -212,6 +284,8 @@ public class ProcessUtils { result.stderr = StringUtils.joinWith("\n", errGobbler.getLines()); result.stdoutTruncated = false; result.stderrTruncated = errGobbler.getIsTruncated(); + result.requestedTimeoutMillis = requestedTimeoutMillis; + result.grantedTimeoutMillis = grantedTimeoutMillis; return result; } finally { if (p != null) { @@ -228,25 +302,49 @@ public class ProcessUtils { * Checks to see if the command can be run. Typically used with * something like "myapp --version" to check to see if "myapp" * is installed and on the path. + * <p> + * Equivalent to {@link #checkCommandWithTimeout(String[], long, int...)} with + * {@link #DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS}. * * @param checkCmd The check command to run * @param errorValue What is considered an error value? Default is 127 (command not found). * @return true if the command ran successfully (exit code not in errorValue list) */ public static boolean checkCommand(String checkCmd, int... errorValue) { - return checkCommand(new String[]{checkCmd}, errorValue); + return checkCommandWithTimeout(new String[]{checkCmd}, DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS, errorValue); } /** * Checks to see if the command can be run. Typically used with * something like {@code new String[]{"myapp", "--version"}} to check to see if "myapp" * is installed and on the path. + * <p> + * Equivalent to {@link #checkCommandWithTimeout(String[], long, int...)} with + * {@link #DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS}. * * @param checkCmd The check command to run * @param errorValue What is considered an error value? Default is 127 (command not found). * @return true if the command ran successfully (exit code not in errorValue list) */ public static boolean checkCommand(String[] checkCmd, int... errorValue) { + return checkCommandWithTimeout(checkCmd, DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS, errorValue); + } + + /** + * Same as {@link #checkCommand(String[], int...)}, but with a caller-specified timeout + * instead of the {@value #DEFAULT_CHECK_COMMAND_TIMEOUT_MILLIS}ms default. + * <p> + * Deliberately a distinct method name, not a same-named overload: {@code checkCommand(cmd, 500)} + * would silently resolve to {@code checkCommand(String[], int...)} with + * {@code errorValue={500}} at the default timeout -- Java prefers the varargs-only + * overload over widening {@code int} to this method's {@code long} parameter, with no + * compile error to catch the mistake. + * + * @param timeoutMillis how long to wait for the command to exit + * @param errorValue What is considered an error value? Default is 127 (command not found). + * @return true if the command ran successfully (exit code not in errorValue list) + */ + public static boolean checkCommandWithTimeout(String[] checkCmd, long timeoutMillis, int... errorValue) { if (errorValue.length == 0) { errorValue = new int[]{127}; } @@ -260,7 +358,7 @@ public class ProcessUtils { Thread errThread = new Thread(errGobbler); outThread.start(); errThread.start(); - boolean finished = process.waitFor(60000, TimeUnit.MILLISECONDS); + boolean finished = process.waitFor(timeoutMillis, TimeUnit.MILLISECONDS); if (!finished) { throw new TimeoutException(); } @@ -293,4 +391,36 @@ public class ProcessUtils { } } + /** + * Waits for the process to exit, like {@link Process#waitFor(long, TimeUnit)}, but polls + * in {@value #HEARTBEAT_INTERVAL_MILLIS} ms increments and checkpoints {@code context}'s + * {@link ParseTimeout} after each increment that doesn't complete -- this is what lets a + * bounded external call run longer than the progress timeout without tripping the stall + * detector: the wait itself is progress. + * <p> + * Public so callers managing their own {@link Process} (not going through + * {@link #execute(ProcessBuilder, ParseContext, long, int, int)}) can still checkpoint + * while waiting. + * + * @param context may be null, in which case no checkpoint is recorded + * @param timeoutMillis total wait time in ms; zero or negative checks once without waiting + * @return true if the process exited before the timeout elapsed + */ + 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; + while (true) { + long remaining = deadline - System.currentTimeMillis(); + long pollMillis = remaining <= 0 ? 0 : Math.min(remaining, HEARTBEAT_INTERVAL_MILLIS); + if (p.waitFor(pollMillis, TimeUnit.MILLISECONDS)) { + return true; + } + if (remaining <= 0) { + return false; + } + ParseTimeout.checkpoint(context); + } + } + } 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 new file mode 100644 index 0000000000..74619474f2 --- /dev/null +++ b/tika-core/src/test/java/org/apache/tika/config/ParseTimeoutTest.java @@ -0,0 +1,222 @@ +/* + * 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.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +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 ParseTimeoutTest { + + @Test + public void testInitialTimestamp() { + long before = System.currentTimeMillis(); + 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); + } + + @Test + public void testCheckpointAdvancesTimestamp() throws Exception { + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits()); + long initial = timeout.getLastProgressMillis(); + + Thread.sleep(20); + timeout.checkpoint(); + + assertTrue(timeout.getLastProgressMillis() > initial); + } + + @Test + public void testConcurrentCheckpoints() throws Exception { + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits()); + 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++) { + timeout.checkpoint(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }).start(); + } + + startLatch.countDown(); + assertTrue(doneLatch.await(5, TimeUnit.SECONDS)); + assertTrue(timeout.getLastProgressMillis() > 0); + } + + @Test + public void testBudgetForNeverExceedsRequested() { + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits(3_600_000L, 60_000L)); + assertEquals(5_000L, timeout.budgetFor(5_000L)); + } + + @Test + public void testBudgetForClipsToRemaining() { + // total task budget of 1000ms; request far more than that -- must clip to ~1000ms + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits(1_000L, 1_000L)); + long budget = timeout.budgetFor(600_000L); + assertTrue(budget <= 1_000L, "budget should be clipped to remaining, was " + budget); + assertTrue(budget >= 0L); + } + + @Test + public void testBudgetForNeverGrowsAcrossACompositionChain() { + // A tight parser-level request must not be "rescued" back up to a looser value + // by an outer budget, regardless of order -- min() composes correctly either way. + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits(3_600_000L, 60_000L)); + long innerRequested = 500L; + assertEquals(innerRequested, timeout.budgetFor(innerRequested)); + } + + @Test + public void testRemainingMillisIsNeverNegative() throws Exception { + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits(1L, 1L)); + Thread.sleep(20); + assertEquals(0L, timeout.remainingMillis()); + assertEquals(0L, timeout.budgetFor(60_000L)); + } + + @Test + public void testHardDeadlineOverflowGuard() { + // 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.remainingMillis()); + assertTrue(timeout.budgetFor(60_000L) == 60_000L); + } + + @Test + public void testGetOrCreateInstallsAndReusesSameInstance() { + ParseContext context = new ParseContext(); + ParseTimeout first = ParseTimeout.getOrCreate(context); + ParseTimeout second = ParseTimeout.getOrCreate(context); + + assertNotNull(first); + assertSame(first, second, "getOrCreate must be idempotent per context"); + assertSame(first, context.get(ParseTimeout.class)); + } + + @Test + public void testGetOrCreateUsesTimeoutLimitsFromContext() { + ParseContext context = new ParseContext(); + context.set(TimeoutLimits.class, new TimeoutLimits(7_200_000L, 300_000L)); + + ParseTimeout timeout = ParseTimeout.getOrCreate(context); + + assertEquals(300_000L, timeout.getProgressTimeoutMillis()); + } + + @Test + public void testCheckpointStaticWithNullContext() { + // Should not throw + ParseTimeout.checkpoint(null); + } + + @Test + public void testCheckpointStaticWithNoTimeoutInstalled() { + // Should not throw, and should not install one as a side effect + ParseContext context = new ParseContext(); + ParseTimeout.checkpoint(context); + assertEquals(null, context.get(ParseTimeout.class)); + } + + @Test + public void testCheckpointStaticUpdatesInstalledTimeout() throws Exception { + ParseContext context = new ParseContext(); + ParseTimeout timeout = ParseTimeout.getOrCreate(context); + long initial = timeout.getLastProgressMillis(); + + Thread.sleep(20); + ParseTimeout.checkpoint(context); + + assertTrue(timeout.getLastProgressMillis() > initial); + } + + // ---- misconfiguration validation (design doc §9) ---------------------------------- + + @Test + public void testStartRejectsNegativeTotal() { + assertThrows(IllegalArgumentException.class, + () -> ParseTimeout.start(new TimeoutLimits(-1L, 1000L))); + } + + @Test + public void testStartRejectsNegativeProgress() { + assertThrows(IllegalArgumentException.class, + () -> ParseTimeout.start(new TimeoutLimits(1000L, -1L))); + } + + @Test + public void testStartAllowsZeroAsAnAlreadyExhaustedBudget() { + // Zero is a coherent state (no time left), unlike negative -- must not throw, + // and must behave as immediately exhausted. + ParseTimeout timeout = assertDoesNotThrow(() -> ParseTimeout.start(new TimeoutLimits(0L, 0L))); + assertEquals(0L, timeout.remainingMillis()); + } + + @Test + public void testBudgetForNonPositiveRequestFallsBackToRemainingInsteadOfZero() { + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits(60_000L, 60_000L)); + + // A misconfigured 0 or negative per-parser timeout must not silently grant a + // budget of zero (which would make every call fail instantly with no useful + // diagnostic) -- it falls back to whatever remains of the task instead. + assertTrue(timeout.budgetFor(0L) > 0L); + assertTrue(timeout.budgetFor(-100L) > 0L); + } + + @Test + public void testBudgetForSubSecondRequestIsStillHonored() { + // The sub-second warning is diagnostic only -- it must not alter the granted budget. + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits(60_000L, 60_000L)); + assertEquals(500L, timeout.budgetFor(500L)); + } + + @Test + public void testBudgetForExceedingTotalStillClipsToRemaining() { + // The "exceeds total" warning is diagnostic only -- min(requested, remaining) + // still applies exactly as it would without the warning. + ParseTimeout timeout = ParseTimeout.start(new TimeoutLimits(1_000L, 1_000L)); + long budget = timeout.budgetFor(600_000L); + assertTrue(budget <= 1_000L, "budget should still be clipped to remaining, was " + budget); + } +} 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 index 3c20a1e135..007f227a97 100644 --- a/tika-core/src/test/java/org/apache/tika/config/TikaProgressTrackerTest.java +++ b/tika-core/src/test/java/org/apache/tika/config/TikaProgressTrackerTest.java @@ -100,4 +100,16 @@ public class TikaProgressTrackerTest { // 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/extractor/ParsingEmbeddedDocumentExtractorTimeoutTest.java b/tika-core/src/test/java/org/apache/tika/extractor/ParsingEmbeddedDocumentExtractorTimeoutTest.java new file mode 100644 index 0000000000..83ba99dc62 --- /dev/null +++ b/tika-core/src/test/java/org/apache/tika/extractor/ParsingEmbeddedDocumentExtractorTimeoutTest.java @@ -0,0 +1,171 @@ +/* + * 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.extractor; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +import java.io.IOException; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; +import org.xml.sax.ContentHandler; +import org.xml.sax.SAXException; +import org.xml.sax.helpers.DefaultHandler; + +import org.apache.tika.config.ParseTimeout; +import org.apache.tika.config.TimeoutLimits; +import org.apache.tika.exception.EmbeddedLimitReachedException; +import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; +import org.apache.tika.io.TikaInputStream; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.mime.MediaType; +import org.apache.tika.parser.ParseContext; +import org.apache.tika.parser.ParseRecord; +import org.apache.tika.parser.Parser; + +/** + * Tests the two ways a recursive parse can run out of time, per the 4.0 timeout + * redesign: a single embedded document's own operation timing out (recorded, siblings + * continue) versus the task's total deadline being exhausted (a document-level fact, + * remaining children skipped cleanly, no exception unless explicitly configured). + */ +public class ParsingEmbeddedDocumentExtractorTimeoutTest { + + private static TikaInputStream tis() throws IOException { + return TikaInputStream.get(new byte[]{1, 2, 3}); + } + + private static ParseContext contextWithAmpleBudget() { + ParseContext context = new ParseContext(); + ParseRecord parseRecord = ParseRecord.newInstance(context); + context.set(ParseRecord.class, parseRecord); + context.set(ParseTimeout.class, ParseTimeout.start(new TimeoutLimits(60_000, 60_000))); + return context; + } + + private static ParseContext contextWithExhaustedBudget() { + ParseContext context = new ParseContext(); + ParseRecord parseRecord = ParseRecord.newInstance(context); + context.set(ParseRecord.class, parseRecord); + // total=0 -> remainingMillis() is already 0 by the time anything checks it + context.set(ParseTimeout.class, ParseTimeout.start(new TimeoutLimits(0, 0))); + return context; + } + + @Test + public void testChildTimeoutIsRecordedAndSiblingsContinue() throws Exception { + ParseContext context = contextWithAmpleBudget(); + AtomicInteger calls = new AtomicInteger(0); + context.set(Parser.class, new Parser() { + @Override + public Set<MediaType> getSupportedTypes(ParseContext context) { + return Set.of(); + } + + @Override + public void parse(TikaInputStream tis, ContentHandler handler, Metadata metadata, + ParseContext context) throws IOException, SAXException, TikaException { + if (calls.getAndIncrement() == 0) { + throw new TikaTimeoutException("op timed out", 1000, 1000); + } + // second call: "succeeds" -- does nothing + } + }); + + ParsingEmbeddedDocumentExtractor extractor = new ParsingEmbeddedDocumentExtractor(context); + ContentHandler handler = new DefaultHandler(); + + // First embedded document: its own operation times out. + extractor.parseEmbedded(tis(), handler, new Metadata(), context, false); + + ParseRecord parseRecord = context.get(ParseRecord.class); + assertEquals(1, parseRecord.getExceptions().size(), + "the child timeout should be recorded as an exception"); + assertTrue(parseRecord.getExceptions().get(0) instanceof TikaTimeoutException); + assertFalse(parseRecord.isTaskDeadlineReached(), + "a single child timing out is not a task-deadline event"); + + // Second embedded document (sibling): must still be attempted and succeed. + extractor.parseEmbedded(tis(), handler, new Metadata(), context, false); + assertEquals(2, calls.get(), "the second sibling must still be parsed"); + assertEquals(1, parseRecord.getExceptions().size(), + "no new exception should be recorded for the successful sibling"); + } + + @Test + public void testChildTimeoutMessageReportsRequestedAndGranted() { + TikaTimeoutException clipped = new TikaTimeoutException("timed out", 5000, 1200); + assertTrue(clipped.isClippedByRemaining()); + assertTrue(clipped.getMessage().contains("task remaining")); + + TikaTimeoutException exhausted = new TikaTimeoutException("timed out", 5000, 5000); + assertFalse(exhausted.isClippedByRemaining()); + assertTrue(exhausted.getMessage().contains("budget exhausted")); + } + + @Test + public void testDeadlineExhaustedSkipsRemainingChildrenCleanly() throws Exception { + ParseContext context = contextWithExhaustedBudget(); + context.set(Parser.class, new Parser() { + @Override + public Set<MediaType> getSupportedTypes(ParseContext context) { + return Set.of(); + } + + @Override + public void parse(TikaInputStream tis, ContentHandler handler, Metadata metadata, + ParseContext context) { + fail("a child must not be attempted once the task deadline is reached"); + } + }); + + ParsingEmbeddedDocumentExtractor extractor = new ParsingEmbeddedDocumentExtractor(context); + ContentHandler handler = new DefaultHandler(); + + assertFalse(extractor.shouldParseEmbedded(new Metadata())); + + ParseRecord parseRecord = context.get(ParseRecord.class); + assertTrue(parseRecord.isTaskDeadlineReached()); + + // parseEmbedded enforces the limit even if the caller skipped shouldParseEmbedded, + // and must not throw by default -- it returns having done nothing. + extractor.parseEmbedded(tis(), handler, new Metadata(), context, false); + // A second sibling: still skipped, same as a hard count limit. + extractor.parseEmbedded(tis(), handler, new Metadata(), context, false); + + assertTrue(parseRecord.getExceptions().isEmpty(), + "skipping for deadline is not itself recorded as an exception"); + } + + @Test + public void testThrowOnDeadlineThrowsEmbeddedLimitReachedException() { + ParseContext context = contextWithExhaustedBudget(); + context.get(ParseRecord.class).setThrowOnDeadline(true); + + ParsingEmbeddedDocumentExtractor extractor = new ParsingEmbeddedDocumentExtractor(context); + + EmbeddedLimitReachedException ex = assertThrows(EmbeddedLimitReachedException.class, + () -> extractor.shouldParseEmbedded(new Metadata())); + assertEquals(EmbeddedLimitReachedException.LimitType.DEADLINE, ex.getLimitType()); + } +} diff --git a/tika-core/src/test/java/org/apache/tika/parser/mock/MockParser.java b/tika-core/src/test/java/org/apache/tika/parser/mock/MockParser.java index bb3bb3a28d..63fb07c03f 100644 --- a/tika-core/src/test/java/org/apache/tika/parser/mock/MockParser.java +++ b/tika-core/src/test/java/org/apache/tika/parser/mock/MockParser.java @@ -51,6 +51,7 @@ import org.w3c.dom.NodeList; import org.xml.sax.ContentHandler; import org.xml.sax.SAXException; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.TikaException; import org.apache.tika.extractor.EmbeddedDocumentExtractor; import org.apache.tika.extractor.EmbeddedDocumentUtil; @@ -157,6 +158,8 @@ public class MockParser implements Parser { hang(action); } else if ("fakeload".equals(name)) { fakeload(action); + } else if ("checkpointedSleep".equals(name)) { + checkpointedSleep(action, context); } else if ("oom".equals(name)) { kabOOM(); } else if ("print_out".equals(name) || "print_err".equals(name)) { @@ -238,6 +241,28 @@ public class MockParser implements Parser { } + /** + * Sleeps for {@code millis}, checkpointing {@link ParseTimeout} every + * {@code intervalMillis} -- simulates a bounded external call that reports progress + * while it waits, e.g. {@code ProcessUtils.execute}'s heartbeat loop. + */ + private void checkpointedSleep(Node action, ParseContext context) { + NamedNodeMap attrs = action.getAttributes(); + long millis = Long.parseLong(attrs.getNamedItem("millis").getNodeValue()); + long intervalMillis = Long.parseLong(attrs.getNamedItem("intervalMillis").getNodeValue()); + long deadline = System.currentTimeMillis() + millis; + long remaining; + while ((remaining = deadline - System.currentTimeMillis()) > 0) { + try { + Thread.sleep(Math.min(intervalMillis, remaining)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + ParseTimeout.checkpoint(context); + } + } + private void throwIllegalChars() throws IOException { throw new IOException("Can't say \u0000 in xml or \u0001 or \u0002 or \u0003"); } 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 new file mode 100644 index 0000000000..86d9582bb2 --- /dev/null +++ b/tika-core/src/test/java/org/apache/tika/utils/ProcessUtilsTest.java @@ -0,0 +1,113 @@ +/* + * 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.utils; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeFalse; + +import org.junit.jupiter.api.Test; + +import org.apache.tika.config.ParseTimeout; +import org.apache.tika.config.TimeoutLimits; +import org.apache.tika.parser.ParseContext; + +/** + * These tests spawn the OS {@code sleep} command directly (unavailable on Windows) rather + * than mocking {@link Process}, because the property under test -- that a checkpoint fires + * while a real bounded wait is still in progress, not only after it returns -- is exactly + * the thing a mock would have to fake. + */ +public class ProcessUtilsTest { + + @Test + public void testHeartbeatFiresWhileProcessIsStillRunning() throws Exception { + assumeFalse(SystemUtils.IS_OS_WINDOWS); + + 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(() -> { + try { + ProcessUtils.execute(pb, context, 20_000, 1000, 1000); + } catch (Exception e) { + // surfaced via the join()+assert below if it prevents progress + } + }); + runner.start(); + + try { + // 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, + "expected a checkpoint to have fired while the process was still running"); + } finally { + runner.join(10_000); + } + } + + @Test + public void testNullContextDoesNotThrow() throws Exception { + assumeFalse(SystemUtils.IS_OS_WINDOWS); + + ProcessBuilder pb = new ProcessBuilder("sleep", "0"); + FileProcessResult result = ProcessUtils.execute(pb, null, 5_000, 1000, 1000); + + assertTrue(!result.isTimeout()); + } + + @Test + public void testTimeoutStillBoundsTheWait() throws Exception { + assumeFalse(SystemUtils.IS_OS_WINDOWS); + + long start = System.currentTimeMillis(); + ProcessBuilder pb = new ProcessBuilder("sleep", "5"); + FileProcessResult result = ProcessUtils.execute(pb, null, 800, 1000, 1000); + long elapsed = System.currentTimeMillis() - start; + + assertTrue(result.isTimeout(), "a 5s sleep with an 800ms budget must time out"); + assertTrue(elapsed < 4_000, + "the polling rewrite must still honor the timeout, not wait for the full sleep; took " + elapsed + "ms"); + } + + @Test + public void testCheckCommandDefaultTimeoutStillWorksForAFastCommand() { + assumeFalse(SystemUtils.IS_OS_WINDOWS); + + assertTrue(ProcessUtils.checkCommand(new String[]{"true"}), + "a fast, well-behaved command must still succeed under the new default"); + } + + @Test + public void testCheckCommandCustomTimeoutBoundsASlowCommand() { + assumeFalse(SystemUtils.IS_OS_WINDOWS); + + long start = System.currentTimeMillis(); + boolean result = ProcessUtils.checkCommandWithTimeout(new String[]{"sleep", "5"}, 500); + long elapsed = System.currentTimeMillis() - start; + + assertFalse(result, "a command that outlives its timeout must report failure"); + assertTrue(elapsed < 4_000, + "checkCommandWithTimeout must honor its own timeout, not the default; took " + elapsed + "ms"); + } +} diff --git a/tika-detectors/tika-detector-magika/src/main/java/org/apache/tika/detect/magika/MagikaDetector.java b/tika-detectors/tika-detector-magika/src/main/java/org/apache/tika/detect/magika/MagikaDetector.java index 67f64a5328..68988f1d3a 100644 --- a/tika-detectors/tika-detector-magika/src/main/java/org/apache/tika/detect/magika/MagikaDetector.java +++ b/tika-detectors/tika-detector-magika/src/main/java/org/apache/tika/detect/magika/MagikaDetector.java @@ -218,14 +218,14 @@ public class MagikaDetector implements Detector { return MediaType.OCTET_STREAM; } //spool the full file to disk if there is no underlying file - return detectOnPath(tis.getPath(), metadata); + return detectOnPath(tis.getPath(), metadata, parseContext); } public Config getDefaultConfig() { return defaultConfig; } - private MediaType detectOnPath(Path path, Metadata metadata) throws IOException { + private MediaType detectOnPath(Path path, Metadata metadata, ParseContext parseContext) throws IOException { String[] args = new String[]{ ProcessUtils.escapeCommandLine(defaultConfig.getMagikaPath()), @@ -233,7 +233,7 @@ public class MagikaDetector implements Detector { "--json" }; ProcessBuilder builder = new ProcessBuilder(args); - FileProcessResult result = ProcessUtils.execute(builder, defaultConfig.getTimeoutMs(), 10000000, 1000); + FileProcessResult result = ProcessUtils.execute(builder, parseContext, defaultConfig.getTimeoutMs(), 10000000, 1000); return processResult(result, metadata, defaultConfig.isUseMime()); } diff --git a/tika-detectors/tika-detector-siegfried/src/main/java/org/apache/tika/detect/siegfried/SiegfriedDetector.java b/tika-detectors/tika-detector-siegfried/src/main/java/org/apache/tika/detect/siegfried/SiegfriedDetector.java index cf0ac07dda..2092bc669d 100644 --- a/tika-detectors/tika-detector-siegfried/src/main/java/org/apache/tika/detect/siegfried/SiegfriedDetector.java +++ b/tika-detectors/tika-detector-siegfried/src/main/java/org/apache/tika/detect/siegfried/SiegfriedDetector.java @@ -193,19 +193,19 @@ public class SiegfriedDetector implements Detector { return MediaType.OCTET_STREAM; } //spool the full file to disk if there is no underlying file - return detectOnPath(tis.getPath(), metadata); + return detectOnPath(tis.getPath(), metadata, parseContext); } public Config getDefaultConfig() { return defaultConfig; } - private MediaType detectOnPath(Path path, Metadata metadata) throws IOException { + private MediaType detectOnPath(Path path, Metadata metadata, ParseContext parseContext) throws IOException { String[] args = new String[]{ProcessUtils.escapeCommandLine(defaultConfig.getSiegfriedPath()), "-json", ProcessUtils.escapeCommandLine(path.toAbsolutePath().toString())}; ProcessBuilder builder = new ProcessBuilder(args); - FileProcessResult result = ProcessUtils.execute(builder, defaultConfig.getTimeoutMs(), 1000000, 1000); + FileProcessResult result = ProcessUtils.execute(builder, parseContext, defaultConfig.getTimeoutMs(), 1000000, 1000); return processResult(result, metadata, defaultConfig.isUseMime()); } 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 ce5418a4f4..acd3fb6505 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 @@ -25,10 +25,15 @@ import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.apache.tika.config.ParseTimeout; import org.apache.tika.exception.TikaException; +import org.apache.tika.parser.ParseContext; /** * Lightweight HTTP client for Tika parser modules that call external REST @@ -48,6 +53,10 @@ public class TikaHttpClient implements Closeable { private static final String JSON_CONTENT_TYPE = "application/json; charset=utf-8"; + // How often a bounded HTTP wait checkpoints the task's ParseTimeout -- see + // org.apache.tika.utils.ProcessUtils.HEARTBEAT_INTERVAL_MILLIS for the same rationale. + private static final long HEARTBEAT_INTERVAL_MILLIS = 1000; + private final HttpClient httpClient; private final ExecutorService executor; private final int defaultTimeoutSeconds; @@ -81,68 +90,160 @@ public class TikaHttpClient implements Closeable { /** * POST a JSON body to {@code url} and return the response body as a string. + * <p> + * Equivalent to {@link #postJson(String, String, Map, int, ParseContext)} with a null + * context: {@code requestedTimeoutSeconds} is granted unclipped, no checkpointing. * - * @param url target URL - * @param jsonBody request body (UTF-8 JSON) - * @param headers additional HTTP headers (e.g. {@code Authorization}) - * @param timeoutSeconds read timeout; {@code 0} uses the default timeout + * @param url target URL + * @param jsonBody request body (UTF-8 JSON) + * @param headers additional HTTP headers (e.g. {@code Authorization}) + * @param requestedTimeoutSeconds read timeout; {@code 0} uses the default timeout * @return response body string * @throws IOException on network error * @throws TikaException on non-2xx HTTP status */ public String postJson(String url, String jsonBody, Map<String, String> headers, - int timeoutSeconds) throws IOException, TikaException { + int requestedTimeoutSeconds) throws IOException, TikaException { + return postJson(url, jsonBody, headers, requestedTimeoutSeconds, null); + } + + /** + * Same as {@link #postJson(String, String, Map, int)}, but bounds the wait to + * {@code min(requestedTimeoutSeconds, ParseTimeout.remainingMillis())} (see + * {@link ParseTimeout#budgetFor(long)}) so no single call can outlast the task's + * total timeout regardless of its own configuration. While waiting, checkpoints the + * {@link ParseTimeout} in {@code context} (if any) every + * {@value #HEARTBEAT_INTERVAL_MILLIS} ms -- see + * {@link org.apache.tika.utils.ProcessUtils#execute(ProcessBuilder, ParseContext, long, int, int)} + * for the same rationale applied to subprocess calls. A null {@code context} means + * the budget is granted unclipped. + * + * @param url target URL + * @param jsonBody request body (UTF-8 JSON) + * @param headers additional HTTP headers (e.g. {@code Authorization}) + * @param requestedTimeoutSeconds the timeout the caller's own configuration asks + * for; {@code 0} uses the default timeout + * @param context may be null + * @return response body string + * @throws IOException on network error + * @throws TikaException on non-2xx HTTP status + */ + public String postJson(String url, String jsonBody, Map<String, String> headers, + int requestedTimeoutSeconds, ParseContext context) throws IOException, TikaException { + int grantedTimeoutSeconds = grantedTimeoutSeconds(requestedTimeoutSeconds, context); HttpRequest.Builder builder = HttpRequest.newBuilder() .uri(URI.create(url)) - .timeout(Duration.ofSeconds(timeoutSeconds > 0 - ? timeoutSeconds : defaultTimeoutSeconds)) + .timeout(Duration.ofSeconds(grantedTimeoutSeconds)) .header("Content-Type", JSON_CONTENT_TYPE) .POST(HttpRequest.BodyPublishers.ofString(jsonBody, StandardCharsets.UTF_8)); headers.forEach(builder::header); - return send(builder.build()); + return send(builder.build(), context); } /** * GET {@code url} and return the response body as a string. * Useful for health-check probes at init time. + * <p> + * Equivalent to {@link #get(String, Map, int, ParseContext)} with a null context. + * + * @param url target URL + * @param headers additional HTTP headers + * @param requestedTimeoutSeconds read timeout; {@code 0} uses the default timeout + * @return response body string + * @throws IOException on network error + * @throws TikaException on non-2xx HTTP status + */ + public String get(String url, Map<String, String> headers, + int requestedTimeoutSeconds) throws IOException, TikaException { + return get(url, headers, requestedTimeoutSeconds, null); + } + + /** + * Same as {@link #get(String, Map, int)}, but bounds the wait to + * {@code min(requestedTimeoutSeconds, ParseTimeout.remainingMillis())} and + * checkpoints while waiting -- see + * {@link #postJson(String, String, Map, int, ParseContext)}. * - * @param url target URL - * @param headers additional HTTP headers - * @param timeoutSeconds read timeout; {@code 0} uses the default timeout + * @param url target URL + * @param headers additional HTTP headers + * @param requestedTimeoutSeconds the timeout the caller's own configuration asks + * for; {@code 0} uses the default timeout + * @param context may be null * @return response body string * @throws IOException on network error * @throws TikaException on non-2xx HTTP status */ public String get(String url, Map<String, String> headers, - int timeoutSeconds) throws IOException, TikaException { + int requestedTimeoutSeconds, ParseContext context) throws IOException, TikaException { + int grantedTimeoutSeconds = grantedTimeoutSeconds(requestedTimeoutSeconds, context); HttpRequest.Builder builder = HttpRequest.newBuilder() .uri(URI.create(url)) - .timeout(Duration.ofSeconds(timeoutSeconds > 0 - ? timeoutSeconds : defaultTimeoutSeconds)) + .timeout(Duration.ofSeconds(grantedTimeoutSeconds)) .GET(); headers.forEach(builder::header); - return send(builder.build()); + return send(builder.build(), context); + } + + /** + * Resolves the requested timeout (0 = client default) against the task's remaining + * budget, in whole seconds, floored at 1 -- {@code Duration.ofSeconds(0)} means "no + * timeout" to the JDK HTTP client, never the intent of an exhausted budget, which + * should fail fast instead. + */ + private int grantedTimeoutSeconds(int requestedTimeoutSeconds, ParseContext context) { + long requestedSeconds = requestedTimeoutSeconds > 0 ? requestedTimeoutSeconds : defaultTimeoutSeconds; + long grantedMillis = ParseTimeout.getOrCreate(context).budgetFor(requestedSeconds * 1000L); + return (int) Math.max(1L, grantedMillis / 1000L); } - private String send(HttpRequest request) throws IOException, TikaException { + private String send(HttpRequest request, ParseContext context) throws IOException, TikaException { + CompletableFuture<HttpResponse<String>> future = httpClient.sendAsync( + request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); try { - HttpResponse<String> response = httpClient.send( - request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); + HttpResponse<String> response = waitWithHeartbeat(future, context); if (response.statusCode() < 200 || response.statusCode() >= 300) { throw new TikaException("HTTP " + response.statusCode() + " from " + request.uri() + ": " + response.body()); } return response.body(); + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + if (cause instanceof IOException ioException) { + throw ioException; + } + throw new IOException("HTTP request failed: " + request.uri(), cause); } catch (InterruptedException e) { Thread.currentThread().interrupt(); + future.cancel(true); throw new IOException("HTTP request interrupted: " + request.uri(), e); } } + /** + * Waits for the future to complete, polling in {@value #HEARTBEAT_INTERVAL_MILLIS} ms + * increments and checkpointing {@code context}'s {@link ParseTimeout} on each + * increment that doesn't complete. The actual deadline is enforced by the JDK via the + * {@code HttpRequest}'s own {@code timeout(Duration)} (see + * {@link #send(HttpRequest, ParseContext)}), surfacing as + * {@link java.net.http.HttpTimeoutException} wrapped in the {@link ExecutionException} + * this method throws -- this loop only sets checkpoint cadence. + */ + private HttpResponse<String> waitWithHeartbeat(CompletableFuture<HttpResponse<String>> future, + ParseContext context) + throws InterruptedException, ExecutionException { + while (true) { + try { + return future.get(HEARTBEAT_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); + } catch (java.util.concurrent.TimeoutException e) { + ParseTimeout.checkpoint(context); + } + } + } + @Override public void close() { executor.shutdown(); 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 new file mode 100644 index 0000000000..a6660b71fa --- /dev/null +++ b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaHttpClientTest.java @@ -0,0 +1,99 @@ +/* + * 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.http; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import org.apache.tika.config.ParseTimeout; +import org.apache.tika.config.TimeoutLimits; +import org.apache.tika.http.TikaTestHttpServer.MockResponse; +import org.apache.tika.parser.ParseContext; + +public class TikaHttpClientTest { + + @Test + public void testHeartbeatFiresWhileWaitingForResponse() throws Exception { + try (TikaTestHttpServer server = new TikaTestHttpServer(); + TikaHttpClient client = TikaHttpClient.build(30)) { + server.enqueue(new MockResponse(200, "{\"ok\":true}", 6000)); + + 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 { + client.get(server.url(), Map.of(), 20, context); + } catch (Exception e) { + // surfaced via the assertion below if it prevented progress + } + }); + requester.start(); + + try { + // 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, + "expected a checkpoint to have fired while the request was still in flight"); + } finally { + requester.join(10_000); + } + } + } + + @Test + public void testNullContextDoesNotThrow() throws Exception { + try (TikaTestHttpServer server = new TikaTestHttpServer(); + TikaHttpClient client = TikaHttpClient.build(30)) { + server.enqueue(new MockResponse(200, "{\"ok\":true}")); + + String body = client.get(server.url(), Map.of(), 5); + + assertEquals("{\"ok\":true}", body); + } + } + + @Test + public void testRequestTimeoutStillBoundsTheWait() throws Exception { + try (TikaTestHttpServer server = new TikaTestHttpServer(); + TikaHttpClient client = TikaHttpClient.build(30)) { + server.enqueue(new MockResponse(200, "{\"ok\":true}", 5000)); + + long start = System.currentTimeMillis(); + boolean threw = false; + try { + client.get(server.url(), Map.of(), 1); + } catch (Exception e) { + threw = true; + } + long elapsed = System.currentTimeMillis() - start; + + assertTrue(threw, "a 5s server delay with a 1s request timeout must fail"); + assertTrue(elapsed < 4_000, + "the async polling rewrite must still honor the request timeout; took " + elapsed + "ms"); + } + } +} diff --git a/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaTestHttpServer.java b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaTestHttpServer.java index 2debc2a753..6df0e53719 100644 --- a/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaTestHttpServer.java +++ b/tika-parsers/tika-http-jdk/src/test/java/org/apache/tika/http/TikaTestHttpServer.java @@ -54,7 +54,11 @@ import java.util.concurrent.atomic.AtomicInteger; public class TikaTestHttpServer implements Closeable { /** A pre-programmed response to return for the next incoming request. */ - public record MockResponse(int status, String body) {} + public record MockResponse(int status, String body, long delayMillis) { + public MockResponse(int status, String body) { + this(status, body, 0); + } + } /** A captured incoming HTTP request. */ public record RecordedRequest(String method, String path, @@ -165,6 +169,15 @@ public class TikaTestHttpServer implements Closeable { resp = new MockResponse(500, "{\"error\":\"no response queued\"}"); } + if (resp.delayMillis() > 0) { + try { + Thread.sleep(resp.delayMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + byte[] responseBytes = resp.body().getBytes(StandardCharsets.UTF_8); String statusText = resp.status() == 200 ? "OK" : resp.status() == 500 ? "Internal Server Error" 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 fc4ff9cd6d..c1064b7c3b 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 @@ -40,7 +40,6 @@ import org.xml.sax.SAXException; import org.apache.tika.annotation.TikaComponent; import org.apache.tika.config.TikaProgressTracker; -import org.apache.tika.config.TimeoutLimits; import org.apache.tika.exception.TikaException; import org.apache.tika.io.TemporaryResources; import org.apache.tika.io.TikaInputStream; @@ -194,9 +193,8 @@ public class GDALParser implements Parser { String[] runCommand = processCommand(tis).split("\\s+", -1); - long localTimeoutMillis = TimeoutLimits.getProcessTimeoutMillis(context, timeoutMs); - FileProcessResult result = ProcessUtils.execute(new ProcessBuilder(runCommand), - localTimeoutMillis, maxStdOut, maxStdErr); + FileProcessResult result = ProcessUtils.execute(new ProcessBuilder(runCommand), context, + timeoutMs, maxStdOut, maxStdErr); metadata.set(ExternalProcess.IS_TIMEOUT, result.isTimeout()); metadata.set(ExternalProcess.EXIT_VALUE, result.getExitValue()); diff --git a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/AbstractEmbeddingFilter.java b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/AbstractEmbeddingFilter.java index 4dffd47929..7c191a41c7 100644 --- a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/AbstractEmbeddingFilter.java +++ b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/AbstractEmbeddingFilter.java @@ -77,12 +77,15 @@ public abstract class AbstractEmbeddingFilter extends MetadataFilter { * Implementations should set {@link Chunk#setVector(float[])} on * each chunk in the list. * - * @param chunks the text chunks to embed - * @param config the resolved config for this call + * @param chunks the text chunks to embed + * @param config the resolved config for this call + * @param parseContext the task's ParseContext -- pass to + * {@link org.apache.tika.http.TikaHttpClient} calls to bound the + * request by the task's remaining budget and checkpoint while waiting * @throws IOException on HTTP errors * @throws TikaException on API-level errors */ - protected abstract void embed(List<Chunk> chunks, InferenceConfig config) + protected abstract void embed(List<Chunk> chunks, InferenceConfig config, ParseContext parseContext) throws IOException, TikaException; @Override @@ -92,11 +95,11 @@ public abstract class AbstractEmbeddingFilter extends MetadataFilter { return; } for (Metadata metadata : metadataList) { - processOne(metadata); + processOne(metadata, parseContext); } } - private void processOne(Metadata metadata) throws TikaException { + private void processOne(Metadata metadata, ParseContext parseContext) throws TikaException { String content = metadata.get(defaultConfig.getContentField()); if (content == null) { LOG.debug("No content found at field '{}'; skipping embedding. " @@ -141,7 +144,7 @@ public abstract class AbstractEmbeddingFilter extends MetadataFilter { for (int i = 0; i < chunks.size(); i += batchSize) { List<Chunk> batch = chunks.subList( i, Math.min(i + batchSize, chunks.size())); - embed(batch, defaultConfig); + embed(batch, defaultConfig, parseContext); } ChunkSerializer.mergeInto(metadata, chunks, defaultConfig.getOutputField()); } catch (IOException e) { 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 addbdd4a22..528d5f23b6 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,8 +27,10 @@ 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.exception.TikaException; import org.apache.tika.http.TikaHttpClient; +import org.apache.tika.parser.ParseContext; import org.apache.tika.utils.StringUtils; /** @@ -77,7 +79,7 @@ public class OpenAIEmbeddingFilter extends AbstractEmbeddingFilter { } @Override - protected void embed(List<Chunk> chunks, InferenceConfig config) + protected void embed(List<Chunk> chunks, InferenceConfig config, ParseContext parseContext) throws IOException, TikaException { if (chunks.isEmpty()) { @@ -93,7 +95,8 @@ public class OpenAIEmbeddingFilter extends AbstractEmbeddingFilter { } String responseBody = httpClient.postJson(url, requestJson, headers, - config.getTimeoutSeconds()); + config.getTimeoutSeconds(), parseContext); + TikaProgressTracker.update(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 941477786b..5dedd7d73c 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 @@ -42,7 +42,6 @@ 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.TimeoutLimits; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; import org.apache.tika.http.TikaHttpClient; @@ -166,11 +165,7 @@ public class OpenAIImageEmbeddingParser implements Parser, Initializable, Closea String mimeType = detectMimeType(metadata); String base64Data = Base64.getEncoder().encodeToString(imageBytes); - long timeoutMillis = TimeoutLimits.getProcessTimeoutMillis( - parseContext, config.getTimeoutSeconds() * 1000L); - int timeoutSeconds = (int) (timeoutMillis / 1000L); - - float[] vector = callEmbeddingEndpoint(config, mimeType, base64Data, timeoutSeconds); + float[] vector = callEmbeddingEndpoint(config, mimeType, base64Data, config.getTimeoutSeconds(), parseContext); TikaProgressTracker.update(parseContext); Locators locators = buildLocators(metadata); @@ -196,7 +191,7 @@ public class OpenAIImageEmbeddingParser implements Parser, Initializable, Closea float[] callEmbeddingEndpoint(ImageEmbeddingConfig config, String mimeType, String base64Data, - int timeoutSeconds) + int timeoutSeconds, ParseContext parseContext) throws IOException, TikaException { String requestJson = buildRequest(config, mimeType, base64Data); @@ -207,7 +202,7 @@ public class OpenAIImageEmbeddingParser implements Parser, Initializable, Closea headers.put(apiKeyHeaderName, apiKeyPrefix + config.getApiKey()); } - String responseBody = httpClient.postJson(url, requestJson, headers, timeoutSeconds); + String responseBody = httpClient.postJson(url, requestJson, headers, timeoutSeconds, parseContext); return parseResponse(responseBody); } 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 657cb918ea..8fbe88099b 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 @@ -43,16 +43,18 @@ 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.ParseTimeout; import org.apache.tika.config.TikaProgressTracker; -import org.apache.tika.config.TimeoutLimits; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; import org.apache.tika.io.TikaInputStream; import org.apache.tika.metadata.Metadata; import org.apache.tika.mime.MediaType; import org.apache.tika.parser.ParseContext; import org.apache.tika.parser.Parser; import org.apache.tika.sax.XHTMLContentHandler; +import org.apache.tika.utils.ProcessUtils; import org.apache.tika.utils.StringUtils; /** @@ -165,12 +167,13 @@ public class Tess4JParser implements Parser, Initializable { xhtml.startDocument(); Tesseract tesseract = null; - long timeoutMillis = TimeoutLimits.getProcessTimeoutMillis( - parseContext, config.getTimeoutSeconds() * 1000L); + long requestedMillis = config.getTimeoutSeconds() * 1000L; + long timeoutMillis = ParseTimeout.getOrCreate(parseContext).budgetFor(requestedMillis); try { - tesseract = borrowTesseract(timeoutMillis); + tesseract = borrowTesseract(parseContext, timeoutMillis); if (tesseract == null) { - throw new TikaException("Timed out waiting for a Tesseract instance from the pool"); + throw new TikaTimeoutException("Timed out waiting for a Tesseract instance from the pool", + requestedMillis, timeoutMillis); } // Apply per-request config if different from defaults @@ -313,15 +316,33 @@ public class Tess4JParser implements Parser, Initializable { } /** - * Borrows a {@link Tesseract} instance from the pool, waiting up to the - * specified timeout. + * Borrows a {@link Tesseract} instance from the pool, waiting up to the specified + * timeout. Polls in {@link ProcessUtils#HEARTBEAT_INTERVAL_MILLIS} increments, + * checkpointing {@code parseContext}'s {@link ParseTimeout} between polls, so a busy + * pool doesn't trip the stall detector while a worker is still legitimately in use. * + * @param parseContext may be null, in which case no checkpoint is recorded * @param timeoutMillis maximum time to wait in milliseconds * @return a Tesseract instance, or null if the timeout elapsed * @throws InterruptedException if the thread was interrupted while waiting */ - private Tesseract borrowTesseract(long timeoutMillis) throws InterruptedException { - return pool.poll(timeoutMillis, TimeUnit.MILLISECONDS); + 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; + while (true) { + long remaining = deadline - System.currentTimeMillis(); + long pollMillis = remaining <= 0 ? 0 : + Math.min(remaining, ProcessUtils.HEARTBEAT_INTERVAL_MILLIS); + Tesseract tesseract = pool.poll(pollMillis, TimeUnit.MILLISECONDS); + if (tesseract != null) { + return tesseract; + } + if (remaining <= 0) { + return null; + } + ParseTimeout.checkpoint(parseContext); + } } /** 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 ceb47ad2c9..49486a9818 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 @@ -34,7 +34,6 @@ 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.TimeoutLimits; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; import org.apache.tika.extractor.ParentContentHandler; @@ -199,16 +198,12 @@ public abstract class AbstractVLMParser implements Parser, Initializable { byte[] fileBytes = readFully(tis); String base64Data = Base64.getEncoder().encodeToString(fileBytes); - long timeoutMillis = TimeoutLimits.getProcessTimeoutMillis( - parseContext, config.getTimeoutSeconds() * 1000L); - int timeoutSeconds = (int) (timeoutMillis / 1000L); - HttpCall call = buildHttpCall(config, base64Data, mimeType); String responseText; try { String responseBody = httpClient.postJson( - call.url(), call.json(), call.headers(), timeoutSeconds); + call.url(), call.json(), call.headers(), config.getTimeoutSeconds(), parseContext); responseText = extractResponseText(responseBody, metadata); TikaProgressTracker.update(parseContext); } catch (TikaException e) { diff --git a/tika-parsers/tika-parsers-standard/tika-parsers-standard-integration-tests/src/test/java/org/apache/tika/parser/RecursiveParserWrapperTest.java b/tika-parsers/tika-parsers-standard/tika-parsers-standard-integration-tests/src/test/java/org/apache/tika/parser/RecursiveParserWrapperTest.java index 955e050d39..6b9d18792c 100644 --- a/tika-parsers/tika-parsers-standard/tika-parsers-standard-integration-tests/src/test/java/org/apache/tika/parser/RecursiveParserWrapperTest.java +++ b/tika-parsers/tika-parsers-standard/tika-parsers-standard-integration-tests/src/test/java/org/apache/tika/parser/RecursiveParserWrapperTest.java @@ -42,7 +42,6 @@ import org.apache.tika.exception.TikaException; import org.apache.tika.io.TikaInputStream; import org.apache.tika.metadata.Metadata; import org.apache.tika.metadata.TikaCoreProperties; -import org.apache.tika.sax.AbstractRecursiveParserWrapperHandler; import org.apache.tika.sax.BasicContentHandlerFactory; import org.apache.tika.sax.ContentHandlerFactory; import org.apache.tika.sax.RecursiveParserWrapperHandler; @@ -292,7 +291,7 @@ public class RecursiveParserWrapperTest extends TikaTest { assertEquals(totalNoLimit, list.size()); limitReached = list.get(0) - .get(AbstractRecursiveParserWrapperHandler.EMBEDDED_RESOURCE_LIMIT_REACHED); + .get(TikaCoreProperties.EMBEDDED_RESOURCE_LIMIT_REACHED); assertNull(limitReached); } @@ -311,7 +310,7 @@ public class RecursiveParserWrapperTest extends TikaTest { assertEquals(maxEmbedded + 1, list.size()); limitReached = list.get(0) - .get(AbstractRecursiveParserWrapperHandler.EMBEDDED_RESOURCE_LIMIT_REACHED); + .get(TikaCoreProperties.EMBEDDED_RESOURCE_LIMIT_REACHED); assertEquals("true", limitReached); } @@ -324,7 +323,7 @@ public class RecursiveParserWrapperTest extends TikaTest { List<Metadata> list = handler.getMetadataList(); assertEquals(totalNoLimit, list.size()); limitReached = list.get(0) - .get(AbstractRecursiveParserWrapperHandler.EMBEDDED_RESOURCE_LIMIT_REACHED); + .get(TikaCoreProperties.EMBEDDED_RESOURCE_LIMIT_REACHED); assertNull(limitReached); } } diff --git a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-cad-module/src/main/java/org/apache/tika/parser/dwg/DWGReadParser.java b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-cad-module/src/main/java/org/apache/tika/parser/dwg/DWGReadParser.java index 3ef16fe2cf..50184d9b11 100644 --- a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-cad-module/src/main/java/org/apache/tika/parser/dwg/DWGReadParser.java +++ b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-cad-module/src/main/java/org/apache/tika/parser/dwg/DWGReadParser.java @@ -49,6 +49,7 @@ import org.xml.sax.SAXException; import org.apache.tika.annotation.TikaComponent; import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; import org.apache.tika.extractor.EmbeddedDocumentExtractor; import org.apache.tika.extractor.EmbeddedDocumentUtil; import org.apache.tika.io.TikaInputStream; @@ -122,7 +123,7 @@ public class DWGReadParser extends AbstractDWGParser { tmpFileOut.getCanonicalPath(), tmpFileIn.getCanonicalPath()); ProcessBuilder pb = new ProcessBuilder().command(command); LOG.debug("About to call DWGRead: {}", command); - FileProcessResult fpr = ProcessUtils.execute(pb, dwgc.getDwgReadTimeout(), 10000, 10000); + FileProcessResult fpr = ProcessUtils.execute(pb, context, dwgc.getDwgReadTimeout(), 10000, 10000); LOG.debug("DWGRead Exit code is: {}", fpr.getExitValue()); if (fpr.getExitValue() == 0) { if (dwgc.isCleanDwgReadOutput()) { @@ -165,8 +166,8 @@ public class DWGReadParser extends AbstractDWGParser { + "if json parsing fails consider reviewing dwgread json output to check it's valid"); } } else if (fpr.isTimeout()) { - throw new TikaException( - "DWGRead Failed - Timeout setting exceeded current setting of " + dwgc.getDwgReadTimeout() ); + throw new TikaTimeoutException("DWGRead timed out", + fpr.getRequestedTimeoutMillis(), fpr.getGrantedTimeoutMillis()); } else { throw new TikaException( diff --git a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-cad-module/src/test/java/org/apache/tika/parser/dwg/DWGParserTest.java b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-cad-module/src/test/java/org/apache/tika/parser/dwg/DWGParserTest.java index c652befed7..31d87afc68 100644 --- a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-cad-module/src/test/java/org/apache/tika/parser/dwg/DWGParserTest.java +++ b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-cad-module/src/test/java/org/apache/tika/parser/dwg/DWGParserTest.java @@ -19,6 +19,7 @@ package org.apache.tika.parser.dwg; 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.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -38,6 +39,7 @@ import org.xml.sax.ContentHandler; import org.apache.tika.TikaTest; import org.apache.tika.config.loader.TikaLoader; import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; import org.apache.tika.io.TikaInputStream; import org.apache.tika.metadata.DWG; import org.apache.tika.metadata.Metadata; @@ -315,7 +317,10 @@ public class DWGParserTest extends TikaTest { () -> getText("architectural_-_annotation_scaling_and_multileaders.dwg", parser), "Expected getText() to throw TikaException but it failed" ); - assertTrue(thrown.getMessage().contains("Timeout setting exceeded current setting of")); + // DWGReadParser now throws TikaTimeoutException, whose message carries the + // requested/granted budget instead of the old literal string this test checked for. + assertInstanceOf(TikaTimeoutException.class, thrown); + assertTrue(thrown.getMessage().contains("timed out")); } } diff --git a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-microsoft-module/src/main/java/org/apache/tika/parser/microsoft/libpst/LibPstParser.java b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-microsoft-module/src/main/java/org/apache/tika/parser/microsoft/libpst/LibPstParser.java index 968eeddfa2..16622ec31e 100644 --- a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-microsoft-module/src/main/java/org/apache/tika/parser/microsoft/libpst/LibPstParser.java +++ b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-microsoft-module/src/main/java/org/apache/tika/parser/microsoft/libpst/LibPstParser.java @@ -36,6 +36,7 @@ import org.apache.tika.config.Initializable; import org.apache.tika.config.JsonConfig; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; import org.apache.tika.io.TikaInputStream; import org.apache.tika.metadata.Metadata; import org.apache.tika.mime.MediaType; @@ -94,11 +95,13 @@ public class LibPstParser implements Parser, Initializable { try { ProcessBuilder pb = getProcessBuilder(pst, activeConfig, outDir, debugFile); XHTMLContentHandler xhtml = new XHTMLContentHandler(contentHandler, metadata, parseContext); - FileProcessResult fileProcessResult = ProcessUtils.execute(pb, activeConfig.getTimeoutSeconds() * 1000l, MAX_STDOUT, MAX_STDERR); + FileProcessResult fileProcessResult = ProcessUtils.execute(pb, parseContext, + activeConfig.getTimeoutSeconds() * 1000L, MAX_STDOUT, MAX_STDERR); xhtml.startDocument(); processContents(outDir, activeConfig, xhtml, metadata, parseContext); if (fileProcessResult.isTimeout()) { - throw new TikaException("Timeout exception: " + fileProcessResult.getProcessTimeMillis()); + throw new TikaTimeoutException("readpst timed out", + fileProcessResult.getRequestedTimeoutMillis(), fileProcessResult.getGrantedTimeoutMillis()); } if (fileProcessResult.getExitValue() != 0) { LOGGER.warn("libpst bad exit value {}", fileProcessResult.getExitValue()); 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 59b57e47a8..ac5a4b2d6d 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 @@ -62,10 +62,11 @@ 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.ParseTimeout; import org.apache.tika.config.TikaProgressTracker; -import org.apache.tika.config.TimeoutLimits; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; import org.apache.tika.extractor.ParentContentHandler; import org.apache.tika.io.TemporaryResources; import org.apache.tika.io.TikaInputStream; @@ -509,12 +510,12 @@ public class TesseractOCRParser extends AbstractExternalProcessParser implements Process process = null; String id = null; - long timeoutMillis = TimeoutLimits.getProcessTimeoutMillis(parseContext, - config.getTimeoutSeconds() * 1000); + long requestedMillis = config.getTimeoutSeconds() * 1000L; + long timeoutMillis = ParseTimeout.getOrCreate(parseContext).budgetFor(requestedMillis); try { process = pb.start(); id = register(process); - runOCRProcess(process, timeoutMillis); + runOCRProcess(process, parseContext, requestedMillis, timeoutMillis); TikaProgressTracker.update(parseContext); } finally { if (process != null) { @@ -526,8 +527,8 @@ public class TesseractOCRParser extends AbstractExternalProcessParser implements } } - private void runOCRProcess(Process process, long timeoutMillis) throws IOException, - TikaException { + private void runOCRProcess(Process process, ParseContext parseContext, long requestedMillis, + long timeoutMillis) throws IOException, TikaException { process.getOutputStream().close(); InputStream out = process.getInputStream(); InputStream err = process.getErrorStream(); @@ -540,9 +541,9 @@ public class TesseractOCRParser extends AbstractExternalProcessParser implements int exitValue = Integer.MIN_VALUE; try { - boolean finished = process.waitFor(timeoutMillis, TimeUnit.MILLISECONDS); + boolean finished = ProcessUtils.waitForWithHeartbeat(process, parseContext, timeoutMillis); if (!finished) { - throw new TikaException("TesseractOCRParser timeout"); + throw new TikaTimeoutException("TesseractOCRParser timeout", requestedMillis, timeoutMillis); } exitValue = process.exitValue(); } catch (InterruptedException e) { diff --git a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/main/java/org/apache/tika/renderer/pdf/poppler/PopplerRenderer.java b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/main/java/org/apache/tika/renderer/pdf/poppler/PopplerRenderer.java index 7bfc54474e..61c48c57a2 100644 --- a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/main/java/org/apache/tika/renderer/pdf/poppler/PopplerRenderer.java +++ b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/main/java/org/apache/tika/renderer/pdf/poppler/PopplerRenderer.java @@ -30,6 +30,7 @@ import java.util.regex.Pattern; import org.apache.tika.annotation.TikaComponent; import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; import org.apache.tika.io.TemporaryResources; import org.apache.tika.io.TikaInputStream; import org.apache.tika.metadata.Metadata; @@ -151,8 +152,11 @@ public class PopplerRenderer implements Renderer { ProcessBuilder builder = new ProcessBuilder(); builder.command(args); FileProcessResult result = ProcessUtils.execute( - builder, timeoutMs, 10, 1000); - if (result.getExitValue() != 0) { + builder, parseContext, timeoutMs, 10, 1000); + if (result.isTimeout()) { + throw new TikaTimeoutException("pdftoppm timed out", + result.getRequestedTimeoutMillis(), result.getGrantedTimeoutMillis()); + } else if (result.getExitValue() != 0) { throw new TikaException( "pdftoppm failed (exit " + result.getExitValue() + "): " + result.getStderr()); diff --git a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/UnrarParser.java b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/UnrarParser.java index 7958e57e87..e7a2b6e7e2 100644 --- a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/UnrarParser.java +++ b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pkg-module/src/main/java/org/apache/tika/parser/pkg/UnrarParser.java @@ -81,7 +81,7 @@ public class UnrarParser implements Parser { try (OutputStream os = Files.newOutputStream(tmp, StandardOpenOption.WRITE)) { IOUtils.copy(tis, os); } - FileProcessResult result = unrar(cwd, tmp); + FileProcessResult result = unrar(cwd, tmp, context); //delete the tmp rar file so that we don't recursively parse it in the next step try { Files.delete(tmp); @@ -89,7 +89,8 @@ public class UnrarParser implements Parser { //warn failed to delete tmp } if (result.isTimeout()) { - throw new TikaTimeoutException("timed out unrarring"); + throw new TikaTimeoutException("timed out unrarring", + result.getRequestedTimeoutMillis(), result.getGrantedTimeoutMillis()); } else if (result.getExitValue() != 0) { if (result.getStderr().contains("error in the encrypted file")) { throw new EncryptedDocumentException(); @@ -139,7 +140,7 @@ public class UnrarParser implements Parser { } } - private FileProcessResult unrar(Path cwd, Path tmp) throws IOException { + private FileProcessResult unrar(Path cwd, Path tmp, ParseContext context) throws IOException { //we could use the -l option to check for potentially bad file names //e.g. path traversals ProcessBuilder pb = new ProcessBuilder(); @@ -152,6 +153,6 @@ public class UnrarParser implements Parser { ProcessUtils.escapeCommandLine(tmp.toAbsolutePath().toString()) ); - return ProcessUtils.execute(pb, timeoutMillis, 10000, 1000); + return ProcessUtils.execute(pb, context, timeoutMillis, 10000, 1000); } } 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 8c1d76ee9e..e16220e2c2 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 @@ -26,8 +26,6 @@ import java.io.Reader; import java.util.ArrayList; import java.util.Collections; import java.util.Set; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.io.IOUtils; @@ -38,11 +36,12 @@ import org.apache.tika.annotation.TikaComponent; 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.config.TimeoutLimits; import org.apache.tika.detect.FileCommandDetector; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; import org.apache.tika.io.TemporaryResources; import org.apache.tika.io.TikaInputStream; import org.apache.tika.metadata.Metadata; @@ -206,15 +205,16 @@ public class StringsParser implements Parser, Initializable { // Reads content printed out by "strings" command Thread gobbler = logStream(out, xhtml, totalBytes); gobbler.start(); + long requestedMillis = config.getTimeoutSeconds() * 1000L; try { - long timeoutMillis = TimeoutLimits.getProcessTimeoutMillis(context, config.getTimeoutSeconds() * 1000L); - boolean completed = process.waitFor(timeoutMillis, TimeUnit.MILLISECONDS); + long timeoutMillis = ParseTimeout.getOrCreate(context).budgetFor(requestedMillis); + boolean completed = ProcessUtils.waitForWithHeartbeat(process, context, timeoutMillis); if (!completed) { - throw new TimeoutException("timed out"); + throw new TikaTimeoutException("strings process timed out", requestedMillis, timeoutMillis); } gobbler.join(10000); TikaProgressTracker.update(context); - } catch (InterruptedException | TimeoutException e) { + } catch (InterruptedException e) { throw new TikaException("strings process failed", e); } finally { process.destroyForcibly(); diff --git a/tika-serialization/src/test/java/org/apache/tika/config/TimeoutLimitsTest.java b/tika-serialization/src/test/java/org/apache/tika/config/TimeoutLimitsTest.java index d82f040fad..e84e3a110d 100644 --- a/tika-serialization/src/test/java/org/apache/tika/config/TimeoutLimitsTest.java +++ b/tika-serialization/src/test/java/org/apache/tika/config/TimeoutLimitsTest.java @@ -83,23 +83,16 @@ public class TimeoutLimitsTest extends TikaTest { } @Test - public void testGetProcessTimeoutMillis() { - // Test with null context - assertEquals(5000, TimeoutLimits.getProcessTimeoutMillis(null, 5000)); - - // Test with context that doesn't have TimeoutLimits + public void testBudgetForComposesViaParseTimeout() { + // getProcessTimeoutMillis() coupled the per-op budget to progressTimeoutMillis (a + // liveness setting); budgetFor() replaces it: honor the requested timeout, but never + // grant more than remains of totalTaskTimeoutMillis. Full coverage in ParseTimeoutTest. ParseContext context = new ParseContext(); - assertEquals(5000, TimeoutLimits.getProcessTimeoutMillis(context, 5000)); + context.set(TimeoutLimits.class, new TimeoutLimits(3600000, 60000)); - // Test with context that has TimeoutLimits - TimeoutLimits limits = new TimeoutLimits(3600000, 60000); - context.set(TimeoutLimits.class, limits); - assertEquals(59900, TimeoutLimits.getProcessTimeoutMillis(context, 5000)); - - // Test with very small progress timeout - TimeoutLimits smallLimits = new TimeoutLimits(3600000, 50); - context.set(TimeoutLimits.class, smallLimits); - assertEquals(0, TimeoutLimits.getProcessTimeoutMillis(context, 5000)); + long budget = ParseTimeout.getOrCreate(context).budgetFor(5000); + + assertEquals(5000, budget); } @Test
