#2097 Tika ParserBolt: optional timeout for the parse with parser.tika.timeout

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.

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`.

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,14 +79,17 @@
 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;
 import org.apache.tika.sax.XHTMLContentHandler;
 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<String, Object> 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<Outlink> toOutlinks(String parentURL, List<Link> links, Metadata pa
 
     @Override
     public void cleanup() {
+        if (parseExecutor != null) {
+            parseExecutor.shutdownNow();
+        }
         if (parseFilters != null) {
             parseFilters.cleanup();
         }