From b75e1bad1f311b7a357ccb0f46bdafcee55e5d38 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Wed, 23 Sep 2026 19:34:39 +0200 Subject: [PATCH 1/2] #2097 Tika ParserBolt: optional timeout for the parse with parser.tika.timeout The parse ran on the executor thread with no time limit, so a document which kept a parser busy held that thread for as long as it took, and the tuple timed out and was replayed into the same parse. The new key parser.tika.timeout sets the maximum time in milliseconds a document may take to parse. When it is set, the parse runs on a separate thread and the bolt waits for it up to that time. A document which takes longer is sent to the status stream as an ERROR with the message "parse timeout" and counted under error_parse_timeout. The parsing thread is interrupted, and the content handler stops the parse at its next SAX event; a parser which neither produces output nor checks for interrupts keeps its thread until it returns, so the bolt moves on to a new thread for the following documents. Parse errors and the text limit behave as before. The default stays -1 so the behaviour does not change unless the key is set. --- docs/src/main/asciidoc/configuration.adoc | 1 + external/tika/README.md | 2 + .../apache/stormcrawler/tika/ParserBolt.java | 142 +++++++++++- .../tika/ParserBoltTimeoutTest.java | 208 ++++++++++++++++++ 4 files changed, 352 insertions(+), 1 deletion(-) create mode 100644 external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index ec59cb4ab..6cff574bb 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -570,6 +570,7 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/tika[tika | parser.tika.config.file | tika-config.json | Name of the classpath resource holding the Tika configuration (JSON format since Tika 4). | parser.extract.embedded | false | Whether to parse embedded documents. Since Tika 4 embedded documents are no longer parsed unless this is set to `true`. | parser.tika.text.maxlength | -1 | Maximum number of characters of text extracted from a document, -1 (or any negative value) for no limit. When the limit is reached the parse stops, the text and outlinks extracted so far are kept and the metadata `parse.text.trimmed` is set to `true`. +| parser.tika.timeout | -1 | Maximum time in milliseconds a document may take to parse, 0 or less for no limit. When set, documents are parsed on a separate thread; a document which takes longer is sent to the status stream as an `ERROR` with the message `parse timeout`. A parser which neither produces output nor checks for interrupts keeps its thread after the timeout, and later documents are parsed on a new one. Keep the value below `topology.message.timeout.secs`. |=== NOTE: When using the Tika `ParserBolt` alongside `JSoupParserBolt`, set `jsoup.treat.non.html.as.error` to `false` so that non-HTML content is passed through to the Tika parser rather than being treated as an error. diff --git a/external/tika/README.md b/external/tika/README.md index b04ae1dda..b7ccb3b02 100644 --- a/external/tika/README.md +++ b/external/tika/README.md @@ -39,4 +39,6 @@ Embedded documents are only parsed when `parser.extract.embedded` is set to `tru The length of the text extracted from a document can be limited with `parser.tika.text.maxlength` (number of characters, default `-1`, any negative value means no limit). When the limit is reached the parse stops, the text and outlinks extracted so far are kept and the document is emitted with the metadata `parse.text.trimmed` set to `true`. +The time spent parsing a document can be limited with `parser.tika.timeout` (milliseconds, default `-1`, 0 or less means no limit). When it is set, documents are parsed on a separate thread and a document which takes longer is sent to the status stream as an `ERROR` with the message `parse timeout`. The parse is interrupted, and stops at its next output; a parser which is stuck without producing output and ignores the interrupt keeps its thread until it returns, and the following documents are parsed on a new thread. Keep the timeout below `topology.message.timeout.secs` so that the tuple is not replayed while it is being parsed. + Since Tika 4, Tika metadata keys use namespaced names, which surface as renamed `parse.*` keys, e.g. `parse.resourceName` is now `parse.tk:resource-name`. diff --git a/external/tika/src/main/java/org/apache/stormcrawler/tika/ParserBolt.java b/external/tika/src/main/java/org/apache/stormcrawler/tika/ParserBolt.java index 5c96aeb02..ca96b09b3 100644 --- a/external/tika/src/main/java/org/apache/stormcrawler/tika/ParserBolt.java +++ b/external/tika/src/main/java/org/apache/stormcrawler/tika/ParserBolt.java @@ -33,6 +33,13 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Pattern; import org.apache.commons.lang3.StringUtils; import org.apache.html.dom.HTMLDocumentImpl; @@ -72,6 +79,7 @@ import org.apache.tika.parser.html.HtmlMapper; import org.apache.tika.parser.html.IdentityHtmlMapper; import org.apache.tika.sax.BodyContentHandler; +import org.apache.tika.sax.ContentHandlerDecorator; import org.apache.tika.sax.Link; import org.apache.tika.sax.LinkContentHandler; import org.apache.tika.sax.TeeContentHandler; @@ -79,7 +87,9 @@ import org.jetbrains.annotations.NotNull; import org.slf4j.LoggerFactory; import org.w3c.dom.DocumentFragment; +import org.xml.sax.Attributes; import org.xml.sax.ContentHandler; +import org.xml.sax.SAXException; /** Uses Tika to parse the output of a fetch and extract text + metadata. */ public class ParserBolt extends BaseRichBolt { @@ -96,6 +106,14 @@ public class ParserBolt extends BaseRichBolt { */ public static final String TEXT_TRIMMED_KEY = "parse.text.trimmed"; + /** + * Configuration key for the maximum time in milliseconds a document may take to parse, a value + * of 0 or less for no limit. + */ + public static final String PARSE_TIMEOUT_PARAM = "parser.tika.timeout"; + + private static final AtomicInteger PARSE_THREAD_COUNT = new AtomicInteger(); + private Tika tika; /** ParseContext configured from the "parse-context" section of the Tika configuration. */ @@ -125,6 +143,11 @@ public class ParserBolt extends BaseRichBolt { private int textMaxLength = -1; + private long parseTimeout = -1; + + /** runs the parses when a timeout is set, replaced after each timeout */ + private ExecutorService parseExecutor; + @Override public void prepare( @NotNull Map conf, @@ -170,6 +193,11 @@ public void prepare( int maxLength = ConfUtils.getInt(conf, TEXT_MAX_LENGTH_PARAM, -1); textMaxLength = maxLength < 0 ? -1 : maxLength; + parseTimeout = ConfUtils.getLong(conf, PARSE_TIMEOUT_PARAM, -1); + if (parseTimeout > 0) { + parseExecutor = newParseExecutor(); + } + tika = instantiateTika(conf); this.collector = collector; @@ -314,8 +342,11 @@ public void execute(Tuple tuple) { String text; boolean textTrimmed = false; try (TikaInputStream tis = TikaInputStream.get(content)) { - tika.getParser().parse(tis, teeHandler, md, parseContext); + parseWithTimeout(tis, teeHandler, md, parseContext); text = textHandler.toString(); + } catch (TimeoutException e) { + handleException(url, null, metadata, tuple, "parse timeout"); + return; } catch (Throwable e) { if (!WriteLimitReachedException.isWriteLimitReached(e)) { handleException(url, e, metadata, tuple, "parse error"); @@ -396,6 +427,112 @@ public void execute(Tuple tuple) { eventCounter.scope("tuple_success").incrBy(1); } + /** + * Parses the document on the executor thread, or on a separate thread under {@link + * #PARSE_TIMEOUT_PARAM} if a timeout is set. + * + * @throws TimeoutException if the parse did not complete in time + */ + private void parseWithTimeout( + TikaInputStream tis, + ContentHandler handler, + org.apache.tika.metadata.Metadata md, + ParseContext parseContext) + throws Exception { + if (parseExecutor == null) { + parse(tis, handler, md, parseContext); + return; + } + Future future = + parseExecutor.submit( + () -> { + parse(tis, new InterruptibleContentHandler(handler), md, parseContext); + return null; + }); + try { + future.get(parseTimeout, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + if (cause instanceof Exception ex) { + throw ex; + } + if (cause instanceof Error err) { + throw err; + } + throw e; + } catch (TimeoutException e) { + // parsers rarely check for interrupts; the handler throws at the next + // SAX event but a parser stuck without producing output keeps its thread, + // so the next documents get a new one + future.cancel(true); + parseExecutor.shutdownNow(); + parseExecutor = newParseExecutor(); + throw e; + } catch (InterruptedException e) { + future.cancel(true); + Thread.currentThread().interrupt(); + throw e; + } + } + + /** Parses the document with the Tika parser. Overridden in tests. */ + void parse( + TikaInputStream tis, + ContentHandler handler, + org.apache.tika.metadata.Metadata md, + ParseContext parseContext) + throws Exception { + tika.getParser().parse(tis, handler, md, parseContext); + } + + private static ExecutorService newParseExecutor() { + return Executors.newSingleThreadExecutor( + r -> { + Thread t = new Thread(r, "tika-parse-" + PARSE_THREAD_COUNT.incrementAndGet()); + t.setDaemon(true); + return t; + }); + } + + /** Stops the parse at the next SAX event once the parsing thread has been interrupted. */ + private static class InterruptibleContentHandler extends ContentHandlerDecorator { + + InterruptibleContentHandler(ContentHandler handler) { + super(handler); + } + + private static void checkInterrupted() throws SAXException { + if (Thread.currentThread().isInterrupted()) { + throw new SAXException("Parse interrupted"); + } + } + + @Override + public void startElement(String uri, String localName, String name, Attributes atts) + throws SAXException { + checkInterrupted(); + super.startElement(uri, localName, name, atts); + } + + @Override + public void endElement(String uri, String localName, String name) throws SAXException { + checkInterrupted(); + super.endElement(uri, localName, name); + } + + @Override + public void characters(char[] ch, int start, int length) throws SAXException { + checkInterrupted(); + super.characters(ch, start, length); + } + + @Override + public void ignorableWhitespace(char[] ch, int start, int length) throws SAXException { + checkInterrupted(); + super.ignorableWhitespace(ch, start, length); + } + } + private static boolean isEmptyDocument(ParseData parseDoc) { byte[] content = parseDoc.getContent(); return (content == null || content.length == 0) @@ -548,6 +685,9 @@ private List toOutlinks(String parentURL, List links, Metadata pa @Override public void cleanup() { + if (parseExecutor != null) { + parseExecutor.shutdownNow(); + } if (parseFilters != null) { parseFilters.cleanup(); } diff --git a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java new file mode 100644 index 000000000..d995d9a12 --- /dev/null +++ b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java @@ -0,0 +1,208 @@ +/* + * 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.stormcrawler.tika; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.apache.storm.task.OutputCollector; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.TestUtil; +import org.apache.stormcrawler.parse.ParsingTester; +import org.apache.stormcrawler.persistence.Status; +import org.apache.tika.io.TikaInputStream; +import org.apache.tika.parser.ParseContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.xml.sax.ContentHandler; + +/** + * Checks that the ParserBolt stops waiting for a parse after parser.tika.timeout. + * + * @see #2097 + */ +class ParserBoltTimeoutTest extends ParsingTester { + + private static final byte[] CONTENT = "some text".getBytes(StandardCharsets.UTF_8); + + private static final String SLOW_URL = "https://example.org/slow.txt"; + + /** a parse of SLOW_URL keeps producing output or ignores interrupts, as configured */ + private static class SlowParserBolt extends ParserBolt { + volatile boolean ignoreInterrupts; + volatile Thread parseThread; + final CountDownLatch release = new CountDownLatch(1); + final CountDownLatch finished = new CountDownLatch(1); + + @Override + void parse( + TikaInputStream tis, + ContentHandler handler, + org.apache.tika.metadata.Metadata md, + ParseContext parseContext) + throws Exception { + parseThread = Thread.currentThread(); + String name = md.get(org.apache.tika.metadata.TikaCoreProperties.RESOURCE_NAME_KEY); + if (!"/slow.txt".equals(name)) { + super.parse(tis, handler, md, parseContext); + return; + } + try { + if (ignoreInterrupts) { + // like a parser which does not check for interrupts + while (release.getCount() > 0) { + try { + release.await(); + } catch (InterruptedException e) { + // ignored on purpose + } + } + } else { + char[] chars = "x".toCharArray(); + handler.startDocument(); + while (true) { + handler.characters(chars, 0, chars.length); + } + } + } finally { + finished.countDown(); + } + } + } + + private SlowParserBolt slowBolt; + + @BeforeEach + void setupParserBolt() { + slowBolt = new SlowParserBolt(); + setupParserBolt(slowBolt); + } + + @AfterEach + void releaseParse() { + slowBolt.release.countDown(); + } + + private void prepare(Long timeout) { + Map conf = new HashMap<>(); + if (timeout != null) { + conf.put(ParserBolt.PARSE_TIMEOUT_PARAM, timeout); + } + bolt.prepare(conf, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + } + + private void assertTimedOut() { + Assertions.assertTrue(output.getEmitted().isEmpty()); + List> status = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(1, status.size()); + Assertions.assertEquals(SLOW_URL, status.get(0).get(0)); + Assertions.assertEquals(Status.ERROR, status.get(0).get(2)); + Metadata metadata = (Metadata) status.get(0).get(1); + Assertions.assertEquals( + "parse timeout", metadata.getFirstValue(Constants.STATUS_ERROR_MESSAGE)); + Assertions.assertEquals(1, output.getAckedTuples().size()); + } + + @Test + void parseProducingOutputIsStopped() throws Exception { + prepare(200L); + parse(SLOW_URL, CONTENT, new Metadata()); + + assertTimedOut(); + // the handler throws at the next event once the thread is interrupted + Assertions.assertTrue(slowBolt.finished.await(10, TimeUnit.SECONDS)); + } + + @Test + void nextDocumentIsParsedAfterStuckParse() throws Exception { + prepare(1000L); + slowBolt.ignoreInterrupts = true; + parse(SLOW_URL, CONTENT, new Metadata()); + assertTimedOut(); + Thread stuck = slowBolt.parseThread; + Assertions.assertTrue(stuck.isAlive()); + + parse("https://example.org/fast.txt", CONTENT, new Metadata()); + List> emitted = output.getEmitted(); + Assertions.assertEquals(1, emitted.size()); + Assertions.assertEquals("some text", emitted.get(0).get(3).toString().strip()); + Assertions.assertNotSame(stuck, slowBolt.parseThread); + + slowBolt.release.countDown(); + Assertions.assertTrue(slowBolt.finished.await(10, TimeUnit.SECONDS)); + } + + @Test + void documentIsParsedUnderTimeout() throws IOException { + prepare(10_000L); + parse("https://example.org/fast.txt", CONTENT, new Metadata()); + + Assertions.assertTrue(output.getEmitted(Constants.StatusStreamName).isEmpty()); + List> emitted = output.getEmitted(); + Assertions.assertEquals(1, emitted.size()); + Assertions.assertEquals("some text", emitted.get(0).get(3).toString().strip()); + Assertions.assertNotSame(Thread.currentThread(), slowBolt.parseThread); + } + + @Test + void parseErrorIsReportedUnderTimeout() throws IOException { + prepare(10_000L); + // not a valid PDF, the parser fails + Metadata metadata = new Metadata(); + metadata.setValue("Content-Type", "application/pdf"); + parse("https://example.org/broken.pdf", "%PDF-1.4 broken".getBytes(), metadata); + + List> status = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(1, status.size()); + Metadata md = (Metadata) status.get(0).get(1); + Assertions.assertEquals("parse error", md.getFirstValue(Constants.STATUS_ERROR_MESSAGE)); + } + + @Test + void textLimitAppliesUnderTimeout() throws IOException { + Map conf = new HashMap<>(); + conf.put(ParserBolt.PARSE_TIMEOUT_PARAM, 10_000L); + conf.put(ParserBolt.TEXT_MAX_LENGTH_PARAM, 100); + bolt.prepare(conf, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + byte[] content = "word ".repeat(10_000).getBytes(StandardCharsets.UTF_8); + parse("https://example.org/big.txt", content, new Metadata()); + + List> emitted = output.getEmitted(); + Assertions.assertEquals(1, emitted.size()); + Assertions.assertEquals(100, emitted.get(0).get(3).toString().length()); + Metadata metadata = (Metadata) emitted.get(0).get(2); + Assertions.assertEquals("true", metadata.getFirstValue(ParserBolt.TEXT_TRIMMED_KEY)); + } + + @Test + void noTimeoutByDefault() throws IOException { + prepare(null); + parse("https://example.org/fast.txt", CONTENT, new Metadata()); + + Assertions.assertEquals(1, output.getEmitted().size()); + // parsed on the executor thread itself + Assertions.assertSame(Thread.currentThread(), slowBolt.parseThread); + } +} From 36670fc19dc037ce4cd9990ea398d9b68070b241 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Wed, 23 Sep 2026 20:16:37 +0200 Subject: [PATCH 2/2] #2097 ParserBoltTimeoutTest: use an explicit charset --- .../org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java index d995d9a12..38d8ce770 100644 --- a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java +++ b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltTimeoutTest.java @@ -172,7 +172,8 @@ void parseErrorIsReportedUnderTimeout() throws IOException { // not a valid PDF, the parser fails Metadata metadata = new Metadata(); metadata.setValue("Content-Type", "application/pdf"); - parse("https://example.org/broken.pdf", "%PDF-1.4 broken".getBytes(), metadata); + byte[] content = "%PDF-1.4 broken".getBytes(StandardCharsets.UTF_8); + parse("https://example.org/broken.pdf", content, metadata); List> status = output.getEmitted(Constants.StatusStreamName); Assertions.assertEquals(1, status.size());