From c34143bfcd7d1fb9fb7816bac89ccf1854751e9b Mon Sep 17 00:00:00 2001 From: tallison Date: Wed, 23 Sep 2026 15:21:32 -0400 Subject: [PATCH 01/10] #2097 Tika ParserBolt: optional timeout for the parse with parser.tika.timeout, via Tika Pipes --- docs/src/main/asciidoc/configuration.adoc | 5 + external/tika/README.md | 6 + external/tika/pom.xml | 19 + .../apache/stormcrawler/tika/ParserBolt.java | 329 +++++++++++++++++- .../tika/ParserBoltPipesTimeoutTest.java | 174 +++++++++ 5 files changed, 519 insertions(+), 14 deletions(-) create mode 100644 external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index 24d2ebf83..30504f249 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -570,6 +570,11 @@ 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, the parse runs in a forked JVM (Tika Pipes); a parse that exceeds this is killed outright and reported to the status stream as an `ERROR` with the message `parse timeout`. Keep the value below `topology.message.timeout.secs`. +| parser.tika.pipes.numclients | - | Number of forked JVMs to keep under `parser.tika.timeout`. Unset uses Tika's own CPU-derived default. +| parser.tika.pipes.jvmargs | - | JVM arguments passed to each forked process under `parser.tika.timeout`, e.g. `-Xmx512m`. +| parser.tika.pipes.maxfilesperprocess | - | Restart a forked process after this many documents under `parser.tika.timeout`, to bound slow leaks in parsing libraries. +| parser.tika.pipes.plugins.dir | - | Directory holding Tika Pipes plugin zips. Only needed under `parser.tika.timeout` for documents over the 10MB inline-transfer threshold; unset uses Tika's default plugin directory resolution. |=== 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..0f9f36049 100644 --- a/external/tika/README.md +++ b/external/tika/README.md @@ -39,4 +39,10 @@ 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 set, the parse runs in a forked JVM via [Tika Pipes](https://tika.apache.org/docs/4.0.x/pipes/index.html): a document that takes longer than the timeout is killed outright (not merely asked to stop) and sent to the status stream as an `ERROR` with the message `parse timeout`, and the fork restarts before the next document. Keep the timeout below `topology.message.timeout.secs` so that the tuple is not replayed while it is being parsed. + +A handful of related keys tune the forked JVMs: `parser.tika.pipes.numclients` (how many to keep running, default is Tika's own CPU-derived count), `parser.tika.pipes.jvmargs` (e.g. `-Xmx512m`), `parser.tika.pipes.maxfilesperprocess` (restart a fork after this many documents, to bound slow leaks in parsing libraries), and `parser.tika.pipes.plugins.dir` (only needed for documents over the 10MB inline-transfer threshold; most crawled pages never hit it). + +`parser.htmlmapper.classname` is not applied to parses running under `parser.tika.timeout`: a live `HtmlMapper` instance cannot be sent to the forked JVM. Configure it in the `"parse-context"` section of the Tika configuration file instead if you need it there. + 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/pom.xml b/external/tika/pom.xml index c9d8fe4a7..284eb647d 100644 --- a/external/tika/pom.xml +++ b/external/tika/pom.xml @@ -60,6 +60,14 @@ under the License. ${tika.version} + + + org.apache.tika + tika-pipes-fork-parser + ${tika.version} + + @@ -93,6 +101,17 @@ under the License. test + + + org.apache.tika + tika-core + ${tika.version} + test-jar + test + + 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..c97b19959 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 @@ -21,6 +21,7 @@ import java.io.IOException; import java.io.InputStream; +import java.io.StringReader; import java.net.MalformedURLException; import java.net.URISyntaxException; import java.net.URL; @@ -34,6 +35,8 @@ import java.util.List; import java.util.Map; import java.util.regex.Pattern; +import javax.xml.parsers.ParserConfigurationException; +import javax.xml.parsers.SAXParserFactory; import org.apache.commons.lang3.StringUtils; import org.apache.html.dom.HTMLDocumentImpl; import org.apache.http.HttpHeaders; @@ -61,8 +64,10 @@ import org.apache.stormcrawler.util.MetadataTransfer; import org.apache.stormcrawler.util.URLUtil; import org.apache.tika.Tika; +import org.apache.tika.config.TimeoutLimits; import org.apache.tika.config.loader.TikaLoader; import org.apache.tika.exception.TikaConfigException; +import org.apache.tika.exception.TikaException; import org.apache.tika.exception.WriteLimitReachedException; import org.apache.tika.io.TikaInputStream; import org.apache.tika.metadata.TikaCoreProperties; @@ -71,6 +76,14 @@ import org.apache.tika.parser.Parser; import org.apache.tika.parser.html.HtmlMapper; import org.apache.tika.parser.html.IdentityHtmlMapper; +import org.apache.tika.pipes.api.ParseMode; +import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.core.PipesException; +import org.apache.tika.pipes.fork.PipesForkParser; +import org.apache.tika.pipes.fork.PipesForkParserConfig; +import org.apache.tika.pipes.fork.PipesForkParserException; +import org.apache.tika.pipes.fork.PipesForkResult; +import org.apache.tika.sax.BasicContentHandlerFactory; import org.apache.tika.sax.BodyContentHandler; import org.apache.tika.sax.Link; import org.apache.tika.sax.LinkContentHandler; @@ -80,6 +93,9 @@ import org.slf4j.LoggerFactory; import org.w3c.dom.DocumentFragment; import org.xml.sax.ContentHandler; +import org.xml.sax.InputSource; +import org.xml.sax.SAXException; +import org.xml.sax.XMLReader; /** Uses Tika to parse the output of a fetch and extract text + metadata. */ public class ParserBolt extends BaseRichBolt { @@ -96,6 +112,55 @@ 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. When set, the parse runs in a forked JVM (Tika Pipes) so + * that a parse which exceeds this is killed outright rather than merely asked to stop; the + * fork restarts before the next document. Keep this below {@code + * topology.message.timeout.secs}. + */ + public static final String PARSE_TIMEOUT_PARAM = "parser.tika.timeout"; + + /** + * Directory holding the Tika Pipes plugin zips (only needed under {@link + * #PARSE_TIMEOUT_PARAM} for documents over the 10MB inline-transfer threshold). Unset uses + * Tika's default plugin directory resolution. + */ + public static final String PIPES_PLUGINS_DIR_PARAM = "parser.tika.pipes.plugins.dir"; + + /** Number of forked JVMs to keep under {@link #PARSE_TIMEOUT_PARAM}, unset uses Tika's default. */ + public static final String PIPES_NUM_CLIENTS_PARAM = "parser.tika.pipes.numclients"; + + /** JVM arguments passed to each forked process under {@link #PARSE_TIMEOUT_PARAM}. */ + public static final String PIPES_JVM_ARGS_PARAM = "parser.tika.pipes.jvmargs"; + + /** Restart a forked process after this many documents under {@link #PARSE_TIMEOUT_PARAM}. */ + public static final String PIPES_MAX_FILES_PER_PROCESS_PARAM = + "parser.tika.pipes.maxfilesperprocess"; + + /** + * Safety cap on the markup a fork may return before the local {@link #textMaxLength} is + * applied, independent of it: this bounds XML tag/entity overhead, not visible text. + */ + private static final int PIPES_WRITE_LIMIT_CHARS = 20_000_000; + + private static final SAXParserFactory PIPES_CONTENT_PARSER_FACTORY = + newHardenedSaxParserFactory(); + + private static SAXParserFactory newHardenedSaxParserFactory() { + SAXParserFactory factory = SAXParserFactory.newInstance(); + factory.setNamespaceAware(true); + try { + factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true); + factory.setFeature("http://xml.org/sax/features/external-general-entities", false); + factory.setFeature("http://xml.org/sax/features/external-parameter-entities", false); + } catch (ParserConfigurationException | SAXException e) { + throw new IllegalStateException( + "Failed to configure the XML parser used to re-read Tika Pipes output", e); + } + return factory; + } + private Tika tika; /** ParseContext configured from the "parse-context" section of the Tika configuration. */ @@ -125,6 +190,17 @@ public class ParserBolt extends BaseRichBolt { private int textMaxLength = -1; + private long parseTimeout = -1; + + /** Runs parses under {@link #PARSE_TIMEOUT_PARAM} in a forked JVM; null otherwise. */ + private PipesForkParser pipesForkParser; + + /** On-disk copy of the Tika configuration, kept alive to merge into the forked JVM's config. */ + private Path resolvedTikaConfigPath; + + /** Whether {@link #resolvedTikaConfigPath} is a temp copy this bolt owns and must delete. */ + private boolean resolvedTikaConfigPathIsTemporary; + @Override public void prepare( @NotNull Map conf, @@ -170,8 +246,14 @@ 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); + tika = instantiateTika(conf); + if (parseTimeout > 0) { + pipesForkParser = buildPipesForkParser(conf); + } + this.collector = collector; this.eventCounter = @@ -314,8 +396,32 @@ public void execute(Tuple tuple) { String text; boolean textTrimmed = false; try (TikaInputStream tis = TikaInputStream.get(content)) { - tika.getParser().parse(tis, teeHandler, md, parseContext); + if (pipesForkParser != null) { + PipesParseOutcome outcome = parseWithPipes(tis, md, teeHandler, url); + md = outcome.metadata(); + textTrimmed = outcome.trimmed(); + if (textTrimmed) { + // the fork's own write limit or in-fork deadline cut the parse short, + // distinct from (and rarer than) the local textMaxLength limit below + LOG.info("Parse of {} was trimmed by the forked process", url); + eventCounter.scope("text_trimmed").incrBy(1); + } + } else { + tika.getParser().parse(tis, teeHandler, md, parseContext); + } text = textHandler.toString(); + } catch (ParseTimeoutException e) { + LOG.info("parse timeout -> {}", url); + handleException(url, null, metadata, tuple, "parse timeout"); + return; + } catch (ParseCrashException e) { + handleException(url, e, metadata, tuple, "parse crash"); + return; + } catch (ParsePipesInfraException e) { + // config/infra problem rather than a bad document: likely affects every + // document, not just this one, so it gets a distinct status and a loud log + handleException(url, e, metadata, tuple, "parse pipes error"); + return; } catch (Throwable e) { if (!WriteLimitReachedException.isWriteLimitReached(e)) { handleException(url, e, metadata, tuple, "parse error"); @@ -415,21 +521,23 @@ private Tika instantiateTika(Map conf) { "Tika configuration file " + tikaConfigFile + " not found on classpath"); } LOG.info("Instantiating Tika using custom configuration {}", tikaConfigUrl); - Path configPath = null; - boolean temporary = false; + Path configPath; try { if ("file".equals(tikaConfigUrl.getProtocol())) { configPath = Paths.get(tikaConfigUrl.toURI()); } else { - // TikaLoader can only read configurations from the filesystem: - // copy the resource to a temporary file and delete it as soon - // as the configuration has been loaded + // TikaLoader can only read configurations from the filesystem: copy the + // resource to a temporary file. Kept alive (not deleted here) so that a + // forked JVM under parser.tika.timeout can merge it too; cleaned up in + // cleanup() and marked deleteOnExit() as a safety net. configPath = Files.createTempFile("tika-config", ".json"); - temporary = true; + configPath.toFile().deleteOnExit(); + resolvedTikaConfigPathIsTemporary = true; try (InputStream is = tikaConfigUrl.openStream()) { Files.copy(is, configPath, StandardCopyOption.REPLACE_EXISTING); } } + resolvedTikaConfigPath = configPath; TikaLoader tikaLoader = TikaLoader.load(configPath, getClass().getClassLoader()); configuredParseContext = tikaLoader.loadParseContext(); Tika tika = new Tika(tikaLoader.loadDetectors(), tikaLoader.loadAutoDetectParser()); @@ -438,14 +546,193 @@ private Tika instantiateTika(Map conf) { } catch (IOException | TikaConfigException | URISyntaxException e) { throw new IllegalStateException( "Failed to instantiate Tika using custom configuration " + tikaConfigUrl, e); - } finally { - if (temporary && configPath != null) { - try { - Files.deleteIfExists(configPath); - } catch (IOException e) { - LOG.warn("Failed to delete temporary Tika configuration {}", configPath, e); - } + } + } + + /** + * Builds the {@link PipesForkParser} used under {@link #PARSE_TIMEOUT_PARAM}. The parse + * itself runs in a forked JVM so that {@link #parseTimeout} is enforced by killing the + * process outright (the parent-side {@code socketTimeoutMillis}), not by cooperative + * interruption; a stuck parser cannot keep the bolt's own thread blocked past the timeout. + */ + private PipesForkParser buildPipesForkParser(Map conf) { + PipesForkParserConfig pipesConfig = new PipesForkParserConfig(); + // XML, not HTML: ToXMLContentHandler always self-closes and escapes, so the + // returned content is well-formed and safe to re-read with a SAX parser locally; + // ToHTMLContentHandler leaves some elements unclosed per the HTML spec and is not. + pipesConfig.setHandlerType(BasicContentHandlerFactory.HANDLER_TYPE.XML); + pipesConfig.setWriteLimit(PIPES_WRITE_LIMIT_CHARS); + // one Metadata with the container's fields and container+embedded content + // concatenated, matching how a direct parse merges embedded content into one + // text/link/DOM stream today + pipesConfig.setParseMode(ParseMode.CONCATENATE); + pipesConfig.setMaxEmbeddedCount(extractEmbedded ? -1 : 0); + pipesConfig.setTimeoutLimits(new TimeoutLimits(parseTimeout, parseTimeout)); + // the real enforcement: the parent kills the forked process outright if it does + // not respond within this, regardless of what the parse is doing + pipesConfig.getPipesConfig().setSocketTimeoutMillis(parseTimeout); + if (resolvedTikaConfigPath != null) { + pipesConfig.setUserConfigPath(resolvedTikaConfigPath); + } + + int numClients = ConfUtils.getInt(conf, PIPES_NUM_CLIENTS_PARAM, -1); + if (numClients > 0) { + pipesConfig.setNumClients(numClients); + } + int maxFilesPerProcess = ConfUtils.getInt(conf, PIPES_MAX_FILES_PER_PROCESS_PARAM, -1); + if (maxFilesPerProcess > 0) { + pipesConfig.setMaxFilesPerProcess(maxFilesPerProcess); + } + String pluginsDir = ConfUtils.getString(conf, PIPES_PLUGINS_DIR_PARAM, null); + if (StringUtils.isNotBlank(pluginsDir)) { + pipesConfig.setPluginsDir(Paths.get(pluginsDir)); + } + List jvmArgs = ConfUtils.loadListFromConf(PIPES_JVM_ARGS_PARAM, conf); + if (!jvmArgs.isEmpty()) { + pipesConfig.setJvmArgs(jvmArgs); + } + + // parser.htmlmapper.classname (including its own IdentityHtmlMapper default) is applied + // by setting a live HtmlMapper instance on the ParseContext, which cannot cross the + // fork's IPC boundary; only components the pipes config understands (parse-context + // JSON, EmbeddedLimits, ContentHandlerFactory, ParseMode) do. The forked JVM falls back + // to Tika's own default HtmlMapper unless one is set via the "parse-context" section of + // parser.tika.config.file itself, which does travel with the merged config. + LOG.warn( + "parser.htmlmapper.classname ({}) is not applied to parses running under {}: a " + + "live HtmlMapper cannot be sent to the forked JVM; set it in the " + + "\"parse-context\" section of the Tika configuration file instead if " + + "needed there", + htmlMapperClass.getName(), + PARSE_TIMEOUT_PARAM); + + try { + return new PipesForkParser(pipesConfig); + } catch (IOException | TikaConfigException e) { + throw new IllegalStateException( + "Failed to initialise the Tika Pipes fork parser for " + PARSE_TIMEOUT_PARAM, + e); + } + } + + /** The parse under {@link #PARSE_TIMEOUT_PARAM} did not complete within {@link #parseTimeout}. */ + private static final class ParseTimeoutException extends Exception { + ParseTimeoutException(String message) { + super(message); + } + } + + /** The forked JVM crashed (OOM or otherwise) while parsing; it restarts for the next document. */ + private static final class ParseCrashException extends Exception { + ParseCrashException(String message) { + super(message); + } + } + + /** Tika Pipes itself is misconfigured or unavailable, independently of the document parsed. */ + private static final class ParsePipesInfraException extends Exception { + ParsePipesInfraException(String message, Throwable cause) { + super(message, cause); + } + } + + private record PipesParseOutcome(org.apache.tika.metadata.Metadata metadata, boolean trimmed) {} + + /** + * Parses {@code tis} in a forked JVM and replays the returned content into {@code handler} as + * if it had been parsed in-process, so that outlink extraction, text trimming and DOM-based + * parse filters downstream behave the same as the direct parse path. + */ + private PipesParseOutcome parseWithPipes( + TikaInputStream tis, + org.apache.tika.metadata.Metadata seedMetadata, + ContentHandler handler, + String url) + throws ParseTimeoutException, + ParseCrashException, + ParsePipesInfraException, + IOException, + SAXException { + PipesForkResult result; + try { + result = pipesForkParser.parse(tis, seedMetadata, new ParseContext()); + } catch (TikaException | PipesException e) { + throw new ParsePipesInfraException("Tika Pipes error for " + url, e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new ParsePipesInfraException( + "Interrupted while waiting for a Tika Pipes parse of " + url, e); + } + + if (result.isProcessCrash()) { + if (result.getStatus() == PipesResult.RESULT_STATUS.TIMEOUT) { + throw new ParseTimeoutException( + "Tika parse of " + url + " exceeded " + parseTimeout + "ms"); } + throw new ParseCrashException( + "Tika parse of " + + url + + " crashed the forked process: " + + result.getStatus() + + (result.getMessage() != null ? " - " + result.getMessage() : "")); + } + + if (!result.isSuccess()) { + throw new IOException( + "Tika Pipes parse of " + + url + + " failed: " + + result.getStatus() + + (result.getMessage() != null ? " - " + result.getMessage() : "")); + } + + org.apache.tika.metadata.Metadata resultMetadata = result.getMetadata(); + + if (result.getStatus() == PipesResult.RESULT_STATUS.PARSE_SUCCESS_WITH_EXCEPTION + && !isWriteLimitReached(resultMetadata)) { + // a genuine parser exception, not a text.maxlength trim: match the direct + // parse path, which discards the document entirely rather than keeping + // whatever partial content came out before the parser gave up + throw new IOException( + "Tika Pipes parse of " + + url + + " threw: " + + (resultMetadata != null + ? resultMetadata.get(TikaCoreProperties.CONTAINER_EXCEPTION) + : result.getMessage())); + } + + String xml = result.getContent(); + if (StringUtils.isNotBlank(xml)) { + reparseIntoHandler(xml, handler); + } + + boolean trimmed = + result.getStatus() == PipesResult.RESULT_STATUS.PARSE_SUCCESS_WITH_EXCEPTION + || result.getStatus() == PipesResult.RESULT_STATUS.PARTIAL_TIMEOUT; + return new PipesParseOutcome( + resultMetadata != null ? resultMetadata : seedMetadata, trimmed); + } + + private static boolean isWriteLimitReached(org.apache.tika.metadata.Metadata metadata) { + return metadata != null + && "true".equalsIgnoreCase(metadata.get(TikaCoreProperties.WRITE_LIMIT_REACHED)); + } + + /** + * Replays already-extracted XML content into {@code handler} via a local SAX parse. The + * content was produced by Tika's {@code ToXMLContentHandler}, which always self-closes and + * escapes, so it is well-formed and safe to re-read this way. + */ + private static void reparseIntoHandler(String xml, ContentHandler handler) + throws SAXException, IOException { + try { + XMLReader reader = PIPES_CONTENT_PARSER_FACTORY.newSAXParser().getXMLReader(); + reader.setContentHandler(handler); + reader.parse(new InputSource(new StringReader(xml))); + } catch (ParserConfigurationException e) { + throw new IllegalStateException( + "Failed to create the XML parser used to re-read Tika Pipes output", e); } } @@ -551,5 +838,19 @@ public void cleanup() { if (parseFilters != null) { parseFilters.cleanup(); } + if (pipesForkParser != null) { + try { + pipesForkParser.close(); + } catch (IOException e) { + LOG.warn("Failed to close the Tika Pipes fork parser", e); + } + } + if (resolvedTikaConfigPathIsTemporary && resolvedTikaConfigPath != null) { + try { + Files.deleteIfExists(resolvedTikaConfigPath); + } catch (IOException e) { + LOG.warn("Failed to delete temporary Tika configuration {}", resolvedTikaConfigPath, e); + } + } } } diff --git a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java new file mode 100644 index 000000000..bcda4a158 --- /dev/null +++ b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java @@ -0,0 +1,174 @@ +/* + * 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 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.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +/** + * Checks that parser.tika.timeout, backed by Tika Pipes, actually stops a stuck parse instead of + * merely asking it to stop. Uses Tika's own {@code MockParser} test fixture (from the tika-core + * test-jar) to drive a parse that spins forever and explicitly ignores {@code + * Thread.interrupt()}: the kind of parser a cooperative-interruption approach (checking the + * interrupt flag from a SAX callback) cannot touch, since it is never reached. Killing the forked + * process is the only thing that works here, and the point of this test is to prove that it does. + * + *

{@code MockParser} is dispatched to via {@code application/mock+xml}, which the tika-core + * test-jar registers by {@code } in its own {@code + * custom-mimetypes.xml}. Root-XML sniffing only refines a document magic detection has already + * classified as generic {@code application/xml}, so the {@code } content below must carry + * an {@code } declaration -- without one, detection never gets past magic bytes and + * falls back to {@code text/plain}. No content-type hint is needed once that declaration is + * present. + * + * @see #2097 + * @see #2182 + */ +class ParserBoltPipesTimeoutTest extends ParsingTester { + + private static final String XML_DECLARATION = ""; + + @BeforeEach + void setupParserBolt() { + bolt = new ParserBolt(); + setupParserBolt(bolt); + } + + private void prepare(Map extraConf) { + Map conf = new HashMap<>(extraConf); + conf.putIfAbsent(ParserBolt.PARSE_TIMEOUT_PARAM, 5_000L); + // a single, light forked JVM is enough for these tests and starts faster + conf.putIfAbsent(ParserBolt.PIPES_NUM_CLIENTS_PARAM, 1); + conf.putIfAbsent(ParserBolt.PIPES_JVM_ARGS_PARAM, "-Xmx256m"); + bolt.prepare(conf, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + } + + /** + * MockParser.hang(millis, interruptible=false) keeps sleeping through interruption for the + * full duration: exactly the kind of parser PR #2182's SAX-callback interrupt check can never + * reach, since it is not producing any SAX events at all. A short parser.tika.timeout must + * still stop the bolt well before the hang's own (much longer) duration elapses. + */ + @Test + @Timeout(30) + void parserThatIgnoresInterruptsIsKilledByTimeout() throws IOException { + Map conf = new HashMap<>(); + conf.put(ParserBolt.PARSE_TIMEOUT_PARAM, 2_000L); + prepare(conf); + + String url = "https://example.org/hang.xml"; + byte[] content = + (XML_DECLARATION + "") + .getBytes(StandardCharsets.UTF_8); + + long start = System.currentTimeMillis(); + parse(url, content, new Metadata()); + long elapsed = System.currentTimeMillis() - start; + + Assertions.assertTrue( + elapsed < 20_000, "the bolt should not have waited anywhere near the hang's" + " own 60s duration, took " + elapsed + "ms"); + + Assertions.assertTrue(output.getEmitted().isEmpty()); + List> status = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(1, status.size()); + Assertions.assertEquals(url, status.get(0).get(0)); + Assertions.assertEquals(Status.ERROR, status.get(0).get(2)); + Metadata md = (Metadata) status.get(0).get(1); + Assertions.assertEquals("parse timeout", md.getFirstValue(Constants.STATUS_ERROR_MESSAGE)); + Assertions.assertEquals(1, output.getAckedTuples().size()); + } + + /** A document that parses well within the timeout is emitted normally, text and outlinks included. */ + @Test + @Timeout(30) + void documentIsParsedUnderTimeout() throws IOException { + prepare(new HashMap<>()); + + String url = "https://example.org/fast.html"; + byte[] content = + ("t

hello world

" + + "next page" + + "") + .getBytes(StandardCharsets.UTF_8); + parse(url, content, new Metadata()); + + Assertions.assertTrue(output.getEmitted(Constants.StatusStreamName).stream() + .noneMatch(t -> t.get(2) == Status.ERROR)); + List> emitted = output.getEmitted(); + Assertions.assertEquals(1, emitted.size()); + Assertions.assertTrue(emitted.get(0).get(3).toString().contains("hello world")); + + List> discovered = + output.getEmitted(Constants.StatusStreamName).stream() + .filter(t -> t.get(2) == Status.DISCOVERED) + .toList(); + Assertions.assertEquals(1, discovered.size()); + Assertions.assertEquals("http://example.com/next", discovered.get(0).get(0)); + } + + /** A genuine parse failure (not a timeout or a crash) is still reported as "parse error". */ + @Test + @Timeout(30) + void parseErrorIsReportedUnderTimeout() throws IOException { + prepare(new HashMap<>()); + + String url = "https://example.org/broken.xml"; + byte[] content = + (XML_DECLARATION + + "broken on" + + " purpose") + .getBytes(StandardCharsets.UTF_8); + parse(url, content, new 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)); + } + + /** parser.extract.embedded still gates embedded-document parsing under parser.tika.timeout. */ + @Test + @Timeout(30) + void embeddedNotParsedByDefaultUnderTimeout() throws IOException { + Map conf = new HashMap<>(); + conf.put("parser.extract.embedded", false); + prepare(conf); + + parse( + "https://stormcrawler.apache.org/test_recursive_embedded.docx", + "test_recursive_embedded.docx"); + List> outTuples = output.getEmitted(); + Assertions.assertEquals(1, outTuples.size()); + Assertions.assertFalse( + outTuples.get(0).get(3).toString().contains("Life, Liberty and the pursuit of Happiness"), + "embedded documents should not be parsed when parser.extract.embedded is false"); + } +} From 17c513b2ac7260c871a9dc260298ca9b6c668d63 Mon Sep 17 00:00:00 2001 From: tallison Date: Wed, 23 Sep 2026 16:14:50 -0400 Subject: [PATCH 02/10] follow ups --- THIRD-PARTY.txt | 15 ++++++ .../apache/stormcrawler/tika/ParserBolt.java | 47 +++++++++++-------- .../tika/ParserBoltPipesTimeoutTest.java | 45 +++++++++++------- 3 files changed, 70 insertions(+), 37 deletions(-) diff --git a/THIRD-PARTY.txt b/THIRD-PARTY.txt index 1dfebafb3..ab9d75d43 100644 --- a/THIRD-PARTY.txt +++ b/THIRD-PARTY.txt @@ -73,6 +73,7 @@ List of third-party dependencies grouped by their license type. * Apache Solr (module: solrj) (org.apache.solr:solr-solrj:10.0.0 - https://solr.apache.org/) * Apache Solr (module: solrj-jetty) (org.apache.solr:solr-solrj-jetty:10.0.0 - https://solr.apache.org/) * Apache Solr (module: solrj-zookeeper) (org.apache.solr:solr-solrj-zookeeper:10.0.0 - https://solr.apache.org/) + * Apache Tika Annotation Processor (org.apache.tika:tika-annotation-processor:4.0.0 - https://tika.apache.org) * Apache Tika Apple parser module (org.apache.tika:tika-parser-apple-module:4.0.0 - https://tika.apache.org/tika-parser-apple-module/) * Apache Tika audiovideo parser module (org.apache.tika:tika-parser-audiovideo-module:4.0.0 - https://tika.apache.org/tika-parser-audiovideo-module/) * Apache Tika cad parser module (org.apache.tika:tika-parser-cad-module:4.0.0 - https://tika.apache.org/tika-parser-cad-module/) @@ -81,10 +82,13 @@ List of third-party dependencies grouped by their license type. * Apache Tika crypto parser module (org.apache.tika:tika-parser-crypto-module:4.0.0 - https://tika.apache.org/tika-parser-crypto-module/) * Apache Tika data URI commons (org.apache.tika:tika-parser-datauri-commons:4.0.0 - https://tika.apache.org/tika-parser-datauri-commons/) * Apache Tika digest commons (org.apache.tika:tika-parser-digest-commons:4.0.0 - https://tika.apache.org/tika-parser-digest-commons/) + * Apache Tika eval core (org.apache.tika:tika-eval-core:4.0.0 - https://tika.apache.org/tika-eval-core/) * Apache Tika font parser module (org.apache.tika:tika-parser-font-module:4.0.0 - https://tika.apache.org/tika-parser-font-module/) * Apache Tika HTML encoding detector (org.apache.tika:tika-encoding-detector-html:4.0.0 - https://tika.apache.org/tika-encoding-detectors/tika-encoding-detector-html/) * Apache Tika html parser module (org.apache.tika:tika-parser-html-module:4.0.0 - https://tika.apache.org/tika-parser-html-module/) * Apache Tika image parser module (org.apache.tika:tika-parser-image-module:4.0.0 - https://tika.apache.org/tika-parser-image-module/) + * Apache Tika langdetect (built-in charsoup) (org.apache.tika:tika-langdetect-charsoup:4.0.0 - https://tika.apache.org/tika-langdetect-charsoup/) + * Apache Tika langdetect (charsoup core — no Tika dependencies) (org.apache.tika:tika-langdetect-charsoup-core:4.0.0 - https://tika.apache.org/tika-langdetect-charsoup-core/) * Apache Tika mail commons (org.apache.tika:tika-parser-mail-commons:4.0.0 - https://tika.apache.org/tika-parser-mail-commons/) * Apache Tika mail parser module (org.apache.tika:tika-parser-mail-module:4.0.0 - https://tika.apache.org/tika-parser-mail-module/) * Apache Tika Microsoft parser module (org.apache.tika:tika-parser-microsoft-module:4.0.0 - https://tika.apache.org/tika-parser-microsoft-module/) @@ -96,6 +100,12 @@ List of third-party dependencies grouped by their license type. * Apache Tika OCR parser module (org.apache.tika:tika-parser-ocr-module:4.0.0 - https://tika.apache.org/tika-parser-ocr-module/) * Apache Tika package parser module (org.apache.tika:tika-parser-pkg-module:4.0.0 - https://tika.apache.org/tika-parser-pkg-module/) * Apache Tika PDF parser module (org.apache.tika:tika-parser-pdf-module:4.0.0 - https://tika.apache.org/tika-parser-pdf-module/) + * Apache Tika pipes api (org.apache.tika:tika-pipes-api:4.0.0 - https://tika.apache.org/) + * Apache Tika pipes core (org.apache.tika:tika-pipes-core:4.0.0 - https://tika.apache.org/) + * Apache Tika Pipes File System (org.apache.tika:tika-pipes-file-system:4.0.0 - https://tika.apache.org/tika-pipes-file-system/) + * Apache Tika pipes fork parser (org.apache.tika:tika-pipes-fork-parser:4.0.0 - https://tika.apache.org/) + * Apache Tika Pipes iterators - base (org.apache.tika:tika-pipes-iterator-commons:4.0.0 - https://tika.apache.org/) + * Apache Tika plugins core (org.apache.tika:tika-plugins-core:4.0.0 - https://tika.apache.org) * Apache Tika serialization (org.apache.tika:tika-serialization:4.0.0 - https://tika.apache.org) * Apache Tika standard parser package (org.apache.tika:tika-parsers-standard-package:4.0.0 - https://tika.apache.org/tika-parsers/tika-parsers-standard/tika-parsers-standard-package/) * Apache Tika text parser module (org.apache.tika:tika-parser-text-module:4.0.0 - https://tika.apache.org/tika-parser-text-module/) @@ -176,7 +186,9 @@ List of third-party dependencies grouped by their license type. * jackson-databind (com.fasterxml.jackson.core:jackson-databind:2.22.0 - https://github.com/FasterXML/jackson) * Jackson dataformat: CBOR (com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.18.8 - https://github.com/FasterXML/jackson-dataformats-binary) * Jackson dataformat: Smile (com.fasterxml.jackson.dataformat:jackson-dataformat-smile:2.18.8 - https://github.com/FasterXML/jackson-dataformats-binary) + * Jackson dataformat: Smile (com.fasterxml.jackson.dataformat:jackson-dataformat-smile:2.22.1 - https://github.com/FasterXML/jackson-dataformats-binary) * Jackson-dataformat-YAML (com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.18.8 - https://github.com/FasterXML/jackson-dataformats-text) + * Jackson datatype: JSR310 (com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.22.1 - https://github.com/FasterXML/jackson-modules-java8/jackson-datatype-jsr310) * java-libpst (com.pff:java-libpst:0.9.3 - https://github.com/rjohnsondev/java-libpst) * JCL 1.2 implemented over SLF4J (org.slf4j:jcl-over-slf4j:2.0.17 - http://www.slf4j.org) * JCL 1.2 implemented over SLF4J (org.slf4j:jcl-over-slf4j:2.0.18 - http://www.slf4j.org) @@ -257,6 +269,7 @@ List of third-party dependencies grouped by their license type. * parso (com.epam:parso:2.0.14 - https://github.com/epam/parso) * PDFBox JBIG2 ImageIO plugin (org.apache.pdfbox:jbig2-imageio:3.0.5 - https://www.apache.org/jbig2-imageio/) * perfmark:perfmark-api (io.perfmark:perfmark-api:0.27.0 - https://github.com/perfmark/perfmark) + * PF4J (org.pf4j:pf4j:3.15.0 - https://pf4j.org/pf4j) * Playwright - Driver (com.microsoft.playwright:driver:1.63.0 - https://github.com/microsoft/playwright-java/driver) * Playwright - Main Library (com.microsoft.playwright:playwright:1.63.0 - https://github.com/microsoft/playwright-java/playwright) * Playwright - Node.js For All Platforms (com.microsoft.playwright:driver-bundle:1.63.0 - https://github.com/microsoft/playwright-java/driver-bundle) @@ -267,6 +280,7 @@ List of third-party dependencies grouped by their license type. * rome (com.rometools:rome:2.1.0 - http://rometools.com/rome) * rome-utils (com.rometools:rome-utils:2.1.0 - http://rometools.com/rome-utils) * server (org.opensearch:opensearch:2.19.6 - https://github.com/opensearch-project/OpenSearch.git) + * SLF4J 2 Provider for Log4j API (org.apache.logging.log4j:log4j-slf4j2-impl:2.26.1 - https://logging.apache.org/log4j/2.x/) * SmallRye Mutiny Zero (io.smallrye.reactive:mutiny-zero:1.3.1 - https://smallrye.io) * SnakeYAML (org.yaml:snakeyaml:2.7 - https://codeberg.org/snakeyaml/snakeyaml) * snappy-java (org.xerial.snappy:snappy-java:1.1.10.4 - https://github.com/xerial/snappy-java) @@ -391,6 +405,7 @@ List of third-party dependencies grouped by their license type. * Animal Sniffer Annotations (org.codehaus.mojo:animal-sniffer-annotations:1.24 - https://www.mojohaus.org/animal-sniffer/animal-sniffer-annotations) * dd-plist (com.googlecode.plist:dd-plist:1.30 - http://www.github.com/3breadt/dd-plist) + * Java SemVer (com.github.zafarkhaja:java-semver:0.10.2 - https://github.com/zafarkhaja/jsemver) * JOpt Simple (net.sf.jopt-simple:jopt-simple:5.0.4 - http://jopt-simple.github.io/jopt-simple) * jsoup Java HTML Parser (org.jsoup:jsoup:1.23.2 - https://jsoup.org/) * JTokkit (com.knuddels:jtokkit:1.1.0 - https://github.com/knuddelsgmbh/jtokkit) 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 c97b19959..cebb70d82 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 @@ -81,7 +81,6 @@ import org.apache.tika.pipes.core.PipesException; import org.apache.tika.pipes.fork.PipesForkParser; import org.apache.tika.pipes.fork.PipesForkParserConfig; -import org.apache.tika.pipes.fork.PipesForkParserException; import org.apache.tika.pipes.fork.PipesForkResult; import org.apache.tika.sax.BasicContentHandlerFactory; import org.apache.tika.sax.BodyContentHandler; @@ -113,22 +112,23 @@ 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. When set, the parse runs in a forked JVM (Tika Pipes) so - * that a parse which exceeds this is killed outright rather than merely asked to stop; the - * fork restarts before the next document. Keep this below {@code - * topology.message.timeout.secs}. + * Configuration key for the maximum time in milliseconds a document may take to parse, a value + * of 0 or less for no limit. When set, the parse runs in a forked JVM (Tika Pipes) so that a + * parse which exceeds this is killed outright rather than merely asked to stop; the fork + * restarts before the next document. Keep this below {@code topology.message.timeout.secs}. */ public static final String PARSE_TIMEOUT_PARAM = "parser.tika.timeout"; /** - * Directory holding the Tika Pipes plugin zips (only needed under {@link - * #PARSE_TIMEOUT_PARAM} for documents over the 10MB inline-transfer threshold). Unset uses - * Tika's default plugin directory resolution. + * Directory holding the Tika Pipes plugin zips (only needed under {@link #PARSE_TIMEOUT_PARAM} + * for documents over the 10MB inline-transfer threshold). Unset uses Tika's default plugin + * directory resolution. */ public static final String PIPES_PLUGINS_DIR_PARAM = "parser.tika.pipes.plugins.dir"; - /** Number of forked JVMs to keep under {@link #PARSE_TIMEOUT_PARAM}, unset uses Tika's default. */ + /** + * Number of forked JVMs to keep under {@link #PARSE_TIMEOUT_PARAM}, unset uses Tika's default. + */ public static final String PIPES_NUM_CLIENTS_PARAM = "parser.tika.pipes.numclients"; /** JVM arguments passed to each forked process under {@link #PARSE_TIMEOUT_PARAM}. */ @@ -550,10 +550,10 @@ private Tika instantiateTika(Map conf) { } /** - * Builds the {@link PipesForkParser} used under {@link #PARSE_TIMEOUT_PARAM}. The parse - * itself runs in a forked JVM so that {@link #parseTimeout} is enforced by killing the - * process outright (the parent-side {@code socketTimeoutMillis}), not by cooperative - * interruption; a stuck parser cannot keep the bolt's own thread blocked past the timeout. + * Builds the {@link PipesForkParser} used under {@link #PARSE_TIMEOUT_PARAM}. The parse itself + * runs in a forked JVM so that {@link #parseTimeout} is enforced by killing the process + * outright (the parent-side {@code socketTimeoutMillis}), not by cooperative interruption; a + * stuck parser cannot keep the bolt's own thread blocked past the timeout. */ private PipesForkParser buildPipesForkParser(Map conf) { PipesForkParserConfig pipesConfig = new PipesForkParserConfig(); @@ -615,14 +615,18 @@ private PipesForkParser buildPipesForkParser(Map conf) { } } - /** The parse under {@link #PARSE_TIMEOUT_PARAM} did not complete within {@link #parseTimeout}. */ + /** + * The parse under {@link #PARSE_TIMEOUT_PARAM} did not complete within {@link #parseTimeout}. + */ private static final class ParseTimeoutException extends Exception { ParseTimeoutException(String message) { super(message); } } - /** The forked JVM crashed (OOM or otherwise) while parsing; it restarts for the next document. */ + /** + * The forked JVM crashed (OOM or otherwise) while parsing; it restarts for the next document. + */ private static final class ParseCrashException extends Exception { ParseCrashException(String message) { super(message); @@ -720,9 +724,9 @@ private static boolean isWriteLimitReached(org.apache.tika.metadata.Metadata met } /** - * Replays already-extracted XML content into {@code handler} via a local SAX parse. The - * content was produced by Tika's {@code ToXMLContentHandler}, which always self-closes and - * escapes, so it is well-formed and safe to re-read this way. + * Replays already-extracted XML content into {@code handler} via a local SAX parse. The content + * was produced by Tika's {@code ToXMLContentHandler}, which always self-closes and escapes, so + * it is well-formed and safe to re-read this way. */ private static void reparseIntoHandler(String xml, ContentHandler handler) throws SAXException, IOException { @@ -849,7 +853,10 @@ public void cleanup() { try { Files.deleteIfExists(resolvedTikaConfigPath); } catch (IOException e) { - LOG.warn("Failed to delete temporary Tika configuration {}", resolvedTikaConfigPath, e); + LOG.warn( + "Failed to delete temporary Tika configuration {}", + resolvedTikaConfigPath, + e); } } } diff --git a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java index bcda4a158..270c1aa92 100644 --- a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java +++ b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java @@ -36,18 +36,17 @@ /** * Checks that parser.tika.timeout, backed by Tika Pipes, actually stops a stuck parse instead of * merely asking it to stop. Uses Tika's own {@code MockParser} test fixture (from the tika-core - * test-jar) to drive a parse that spins forever and explicitly ignores {@code - * Thread.interrupt()}: the kind of parser a cooperative-interruption approach (checking the - * interrupt flag from a SAX callback) cannot touch, since it is never reached. Killing the forked - * process is the only thing that works here, and the point of this test is to prove that it does. + * test-jar) to drive a parse that spins forever and explicitly ignores {@code Thread.interrupt()}: + * the kind of parser a cooperative-interruption approach (checking the interrupt flag from a SAX + * callback) cannot touch, since it is never reached. Killing the forked process is the only thing + * that works here, and the point of this test is to prove that it does. * *

{@code MockParser} is dispatched to via {@code application/mock+xml}, which the tika-core * test-jar registers by {@code } in its own {@code * custom-mimetypes.xml}. Root-XML sniffing only refines a document magic detection has already - * classified as generic {@code application/xml}, so the {@code } content below must carry - * an {@code } declaration -- without one, detection never gets past magic bytes and - * falls back to {@code text/plain}. No content-type hint is needed once that declaration is - * present. + * classified as generic {@code application/xml}, so the {@code } content below must carry an + * {@code } declaration -- without one, detection never gets past magic bytes and falls + * back to {@code text/plain}. No content-type hint is needed once that declaration is present. * * @see #2097 * @see #2182 @@ -72,10 +71,10 @@ private void prepare(Map extraConf) { } /** - * MockParser.hang(millis, interruptible=false) keeps sleeping through interruption for the - * full duration: exactly the kind of parser PR #2182's SAX-callback interrupt check can never - * reach, since it is not producing any SAX events at all. A short parser.tika.timeout must - * still stop the bolt well before the hang's own (much longer) duration elapses. + * MockParser.hang(millis, interruptible=false) keeps sleeping through interruption for the full + * duration: exactly the kind of parser PR #2182's SAX-callback interrupt check can never reach, + * since it is not producing any SAX events at all. A short parser.tika.timeout must still stop + * the bolt well before the hang's own (much longer) duration elapses. */ @Test @Timeout(30) @@ -94,7 +93,11 @@ void parserThatIgnoresInterruptsIsKilledByTimeout() throws IOException { long elapsed = System.currentTimeMillis() - start; Assertions.assertTrue( - elapsed < 20_000, "the bolt should not have waited anywhere near the hang's" + " own 60s duration, took " + elapsed + "ms"); + elapsed < 20_000, + "the bolt should not have waited anywhere near the hang's" + + " own 60s duration, took " + + elapsed + + "ms"); Assertions.assertTrue(output.getEmitted().isEmpty()); List> status = output.getEmitted(Constants.StatusStreamName); @@ -106,7 +109,10 @@ void parserThatIgnoresInterruptsIsKilledByTimeout() throws IOException { Assertions.assertEquals(1, output.getAckedTuples().size()); } - /** A document that parses well within the timeout is emitted normally, text and outlinks included. */ + /** + * A document that parses well within the timeout is emitted normally, text and outlinks + * included. + */ @Test @Timeout(30) void documentIsParsedUnderTimeout() throws IOException { @@ -120,8 +126,9 @@ void documentIsParsedUnderTimeout() throws IOException { .getBytes(StandardCharsets.UTF_8); parse(url, content, new Metadata()); - Assertions.assertTrue(output.getEmitted(Constants.StatusStreamName).stream() - .noneMatch(t -> t.get(2) == Status.ERROR)); + Assertions.assertTrue( + output.getEmitted(Constants.StatusStreamName).stream() + .noneMatch(t -> t.get(2) == Status.ERROR)); List> emitted = output.getEmitted(); Assertions.assertEquals(1, emitted.size()); Assertions.assertTrue(emitted.get(0).get(3).toString().contains("hello world")); @@ -168,7 +175,11 @@ void embeddedNotParsedByDefaultUnderTimeout() throws IOException { List> outTuples = output.getEmitted(); Assertions.assertEquals(1, outTuples.size()); Assertions.assertFalse( - outTuples.get(0).get(3).toString().contains("Life, Liberty and the pursuit of Happiness"), + outTuples + .get(0) + .get(3) + .toString() + .contains("Life, Liberty and the pursuit of Happiness"), "embedded documents should not be parsed when parser.extract.embedded is false"); } } From 5c8271afc159f9ea145d9702ab2289751042b35f Mon Sep 17 00:00:00 2001 From: tallison Date: Wed, 23 Sep 2026 16:36:14 -0400 Subject: [PATCH 03/10] tersify --- .../apache/stormcrawler/tika/ParserBolt.java | 70 +++++++------------ .../tika/ParserBoltPipesTimeoutTest.java | 33 +++------ 2 files changed, 36 insertions(+), 67 deletions(-) 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 cebb70d82..a7bdccf15 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 @@ -112,17 +112,15 @@ 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. When set, the parse runs in a forked JVM (Tika Pipes) so that a - * parse which exceeds this is killed outright rather than merely asked to stop; the fork - * restarts before the next document. Keep this below {@code topology.message.timeout.secs}. + * Configuration key for the maximum time in milliseconds a document may take to parse, 0 or + * less for no limit. Runs the parse in a forked JVM (Tika Pipes) and kills it outright if + * exceeded. Keep below {@code topology.message.timeout.secs}. */ public static final String PARSE_TIMEOUT_PARAM = "parser.tika.timeout"; /** - * Directory holding the Tika Pipes plugin zips (only needed under {@link #PARSE_TIMEOUT_PARAM} - * for documents over the 10MB inline-transfer threshold). Unset uses Tika's default plugin - * directory resolution. + * Directory holding Tika Pipes plugin zips, only needed under {@link #PARSE_TIMEOUT_PARAM} + * for documents over the 10MB inline-transfer threshold. Unset uses Tika's default. */ public static final String PIPES_PLUGINS_DIR_PARAM = "parser.tika.pipes.plugins.dir"; @@ -139,8 +137,8 @@ public class ParserBolt extends BaseRichBolt { "parser.tika.pipes.maxfilesperprocess"; /** - * Safety cap on the markup a fork may return before the local {@link #textMaxLength} is - * applied, independent of it: this bounds XML tag/entity overhead, not visible text. + * Safety cap on markup a fork may return, independent of {@link #textMaxLength}: bounds + * tag/entity overhead, not visible text. */ private static final int PIPES_WRITE_LIMIT_CHARS = 20_000_000; @@ -401,8 +399,7 @@ public void execute(Tuple tuple) { md = outcome.metadata(); textTrimmed = outcome.trimmed(); if (textTrimmed) { - // the fork's own write limit or in-fork deadline cut the parse short, - // distinct from (and rarer than) the local textMaxLength limit below + // fork's own write limit or deadline, not the textMaxLength trim below LOG.info("Parse of {} was trimmed by the forked process", url); eventCounter.scope("text_trimmed").incrBy(1); } @@ -418,8 +415,7 @@ public void execute(Tuple tuple) { handleException(url, e, metadata, tuple, "parse crash"); return; } catch (ParsePipesInfraException e) { - // config/infra problem rather than a bad document: likely affects every - // document, not just this one, so it gets a distinct status and a loud log + // infra problem, not a bad document -- affects every parse, distinct status handleException(url, e, metadata, tuple, "parse pipes error"); return; } catch (Throwable e) { @@ -526,10 +522,9 @@ private Tika instantiateTika(Map conf) { if ("file".equals(tikaConfigUrl.getProtocol())) { configPath = Paths.get(tikaConfigUrl.toURI()); } else { - // TikaLoader can only read configurations from the filesystem: copy the - // resource to a temporary file. Kept alive (not deleted here) so that a - // forked JVM under parser.tika.timeout can merge it too; cleaned up in - // cleanup() and marked deleteOnExit() as a safety net. + // TikaLoader needs a filesystem path; kept alive (not deleted here) so a + // forked JVM under parser.tika.timeout can merge it too -- cleaned up in + // cleanup(), with deleteOnExit() as a safety net. configPath = Files.createTempFile("tika-config", ".json"); configPath.toFile().deleteOnExit(); resolvedTikaConfigPathIsTemporary = true; @@ -550,26 +545,21 @@ private Tika instantiateTika(Map conf) { } /** - * Builds the {@link PipesForkParser} used under {@link #PARSE_TIMEOUT_PARAM}. The parse itself - * runs in a forked JVM so that {@link #parseTimeout} is enforced by killing the process - * outright (the parent-side {@code socketTimeoutMillis}), not by cooperative interruption; a - * stuck parser cannot keep the bolt's own thread blocked past the timeout. + * Builds the {@link PipesForkParser} used under {@link #PARSE_TIMEOUT_PARAM}: the parent + * kills the forked process outright via {@code socketTimeoutMillis}, not cooperative + * interruption. */ private PipesForkParser buildPipesForkParser(Map conf) { PipesForkParserConfig pipesConfig = new PipesForkParserConfig(); - // XML, not HTML: ToXMLContentHandler always self-closes and escapes, so the - // returned content is well-formed and safe to re-read with a SAX parser locally; - // ToHTMLContentHandler leaves some elements unclosed per the HTML spec and is not. + // XML, not HTML: ToXMLContentHandler always self-closes/escapes (safe to re-parse + // locally); ToHTMLContentHandler leaves some elements unclosed per the HTML spec. pipesConfig.setHandlerType(BasicContentHandlerFactory.HANDLER_TYPE.XML); pipesConfig.setWriteLimit(PIPES_WRITE_LIMIT_CHARS); - // one Metadata with the container's fields and container+embedded content - // concatenated, matching how a direct parse merges embedded content into one - // text/link/DOM stream today + // matches how the direct parse already merges embedded content into one stream pipesConfig.setParseMode(ParseMode.CONCATENATE); pipesConfig.setMaxEmbeddedCount(extractEmbedded ? -1 : 0); pipesConfig.setTimeoutLimits(new TimeoutLimits(parseTimeout, parseTimeout)); - // the real enforcement: the parent kills the forked process outright if it does - // not respond within this, regardless of what the parse is doing + // the real enforcement: kills the forked process outright if it doesn't respond in time pipesConfig.getPipesConfig().setSocketTimeoutMillis(parseTimeout); if (resolvedTikaConfigPath != null) { pipesConfig.setUserConfigPath(resolvedTikaConfigPath); @@ -592,12 +582,7 @@ private PipesForkParser buildPipesForkParser(Map conf) { pipesConfig.setJvmArgs(jvmArgs); } - // parser.htmlmapper.classname (including its own IdentityHtmlMapper default) is applied - // by setting a live HtmlMapper instance on the ParseContext, which cannot cross the - // fork's IPC boundary; only components the pipes config understands (parse-context - // JSON, EmbeddedLimits, ContentHandlerFactory, ParseMode) do. The forked JVM falls back - // to Tika's own default HtmlMapper unless one is set via the "parse-context" section of - // parser.tika.config.file itself, which does travel with the merged config. + // see the warning below: a live HtmlMapper can't cross the fork's IPC boundary LOG.warn( "parser.htmlmapper.classname ({}) is not applied to parses running under {}: a " + "live HtmlMapper cannot be sent to the forked JVM; set it in the " @@ -643,9 +628,8 @@ private static final class ParsePipesInfraException extends Exception { private record PipesParseOutcome(org.apache.tika.metadata.Metadata metadata, boolean trimmed) {} /** - * Parses {@code tis} in a forked JVM and replays the returned content into {@code handler} as - * if it had been parsed in-process, so that outlink extraction, text trimming and DOM-based - * parse filters downstream behave the same as the direct parse path. + * Parses {@code tis} in a forked JVM, then replays the returned content into {@code handler} + * as if parsed in-process, so outlink/text/DOM handling downstream is unchanged. */ private PipesParseOutcome parseWithPipes( TikaInputStream tis, @@ -694,9 +678,8 @@ private PipesParseOutcome parseWithPipes( if (result.getStatus() == PipesResult.RESULT_STATUS.PARSE_SUCCESS_WITH_EXCEPTION && !isWriteLimitReached(resultMetadata)) { - // a genuine parser exception, not a text.maxlength trim: match the direct - // parse path, which discards the document entirely rather than keeping - // whatever partial content came out before the parser gave up + // not a text.maxlength trim: match the direct path, which discards the whole + // document rather than keep partial content throw new IOException( "Tika Pipes parse of " + url @@ -724,9 +707,8 @@ private static boolean isWriteLimitReached(org.apache.tika.metadata.Metadata met } /** - * Replays already-extracted XML content into {@code handler} via a local SAX parse. The content - * was produced by Tika's {@code ToXMLContentHandler}, which always self-closes and escapes, so - * it is well-formed and safe to re-read this way. + * Re-parses already-extracted XML into {@code handler}. Produced by {@code + * ToXMLContentHandler}, which always self-closes/escapes, so it's safe to re-read this way. */ private static void reparseIntoHandler(String xml, ContentHandler handler) throws SAXException, IOException { diff --git a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java index 270c1aa92..58c4be2d2 100644 --- a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java +++ b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java @@ -34,19 +34,14 @@ import org.junit.jupiter.api.Timeout; /** - * Checks that parser.tika.timeout, backed by Tika Pipes, actually stops a stuck parse instead of - * merely asking it to stop. Uses Tika's own {@code MockParser} test fixture (from the tika-core - * test-jar) to drive a parse that spins forever and explicitly ignores {@code Thread.interrupt()}: - * the kind of parser a cooperative-interruption approach (checking the interrupt flag from a SAX - * callback) cannot touch, since it is never reached. Killing the forked process is the only thing - * that works here, and the point of this test is to prove that it does. + * Proves parser.tika.timeout (Tika Pipes) kills a stuck parse outright, unlike cooperative + * interruption: MockParser.hang(interruptible=false) never checks Thread.interrupt() and + * produces no SAX events, so a callback-based interrupt check would never even run. * - *

{@code MockParser} is dispatched to via {@code application/mock+xml}, which the tika-core - * test-jar registers by {@code } in its own {@code - * custom-mimetypes.xml}. Root-XML sniffing only refines a document magic detection has already - * classified as generic {@code application/xml}, so the {@code } content below must carry an - * {@code } declaration -- without one, detection never gets past magic bytes and falls - * back to {@code text/plain}. No content-type hint is needed once that declaration is present. + *

{@code } content needs an {@code } declaration: MockParser is dispatched + * via {@code application/mock+xml} (registered as root-XML "mock" in tika-core's own + * custom-mimetypes.xml), and root-XML sniffing only refines bytes already magic-classified as + * {@code application/xml}. Without the declaration it falls back to text/plain. * * @see #2097 * @see #2182 @@ -64,18 +59,13 @@ void setupParserBolt() { private void prepare(Map extraConf) { Map conf = new HashMap<>(extraConf); conf.putIfAbsent(ParserBolt.PARSE_TIMEOUT_PARAM, 5_000L); - // a single, light forked JVM is enough for these tests and starts faster + // one light fork is enough here and starts faster conf.putIfAbsent(ParserBolt.PIPES_NUM_CLIENTS_PARAM, 1); conf.putIfAbsent(ParserBolt.PIPES_JVM_ARGS_PARAM, "-Xmx256m"); bolt.prepare(conf, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); } - /** - * MockParser.hang(millis, interruptible=false) keeps sleeping through interruption for the full - * duration: exactly the kind of parser PR #2182's SAX-callback interrupt check can never reach, - * since it is not producing any SAX events at all. A short parser.tika.timeout must still stop - * the bolt well before the hang's own (much longer) duration elapses. - */ + /** A 2s timeout must stop the bolt well before the hang's own 60s duration elapses. */ @Test @Timeout(30) void parserThatIgnoresInterruptsIsKilledByTimeout() throws IOException { @@ -109,10 +99,7 @@ void parserThatIgnoresInterruptsIsKilledByTimeout() throws IOException { Assertions.assertEquals(1, output.getAckedTuples().size()); } - /** - * A document that parses well within the timeout is emitted normally, text and outlinks - * included. - */ + /** Parses normally within the timeout: text and outlinks are still emitted. */ @Test @Timeout(30) void documentIsParsedUnderTimeout() throws IOException { From f4342bb2afd1eaf3747b089fb7d6ad02c63f463a Mon Sep 17 00:00:00 2001 From: tallison Date: Wed, 23 Sep 2026 17:42:35 -0400 Subject: [PATCH 04/10] fix formatting --- .../org/apache/stormcrawler/tika/ParserBolt.java | 13 ++++++------- .../tika/ParserBoltPipesTimeoutTest.java | 8 ++++---- 2 files changed, 10 insertions(+), 11 deletions(-) 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 a7bdccf15..b256805b1 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 @@ -119,8 +119,8 @@ public class ParserBolt extends BaseRichBolt { public static final String PARSE_TIMEOUT_PARAM = "parser.tika.timeout"; /** - * Directory holding Tika Pipes plugin zips, only needed under {@link #PARSE_TIMEOUT_PARAM} - * for documents over the 10MB inline-transfer threshold. Unset uses Tika's default. + * Directory holding Tika Pipes plugin zips, only needed under {@link #PARSE_TIMEOUT_PARAM} for + * documents over the 10MB inline-transfer threshold. Unset uses Tika's default. */ public static final String PIPES_PLUGINS_DIR_PARAM = "parser.tika.pipes.plugins.dir"; @@ -545,9 +545,8 @@ private Tika instantiateTika(Map conf) { } /** - * Builds the {@link PipesForkParser} used under {@link #PARSE_TIMEOUT_PARAM}: the parent - * kills the forked process outright via {@code socketTimeoutMillis}, not cooperative - * interruption. + * Builds the {@link PipesForkParser} used under {@link #PARSE_TIMEOUT_PARAM}: the parent kills + * the forked process outright via {@code socketTimeoutMillis}, not cooperative interruption. */ private PipesForkParser buildPipesForkParser(Map conf) { PipesForkParserConfig pipesConfig = new PipesForkParserConfig(); @@ -628,8 +627,8 @@ private static final class ParsePipesInfraException extends Exception { private record PipesParseOutcome(org.apache.tika.metadata.Metadata metadata, boolean trimmed) {} /** - * Parses {@code tis} in a forked JVM, then replays the returned content into {@code handler} - * as if parsed in-process, so outlink/text/DOM handling downstream is unchanged. + * Parses {@code tis} in a forked JVM, then replays the returned content into {@code handler} as + * if parsed in-process, so outlink/text/DOM handling downstream is unchanged. */ private PipesParseOutcome parseWithPipes( TikaInputStream tis, diff --git a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java index 58c4be2d2..cdb1586c7 100644 --- a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java +++ b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java @@ -35,11 +35,11 @@ /** * Proves parser.tika.timeout (Tika Pipes) kills a stuck parse outright, unlike cooperative - * interruption: MockParser.hang(interruptible=false) never checks Thread.interrupt() and - * produces no SAX events, so a callback-based interrupt check would never even run. + * interruption: MockParser.hang(interruptible=false) never checks Thread.interrupt() and produces + * no SAX events, so a callback-based interrupt check would never even run. * - *

{@code } content needs an {@code } declaration: MockParser is dispatched - * via {@code application/mock+xml} (registered as root-XML "mock" in tika-core's own + *

{@code } content needs an {@code } declaration: MockParser is dispatched via + * {@code application/mock+xml} (registered as root-XML "mock" in tika-core's own * custom-mimetypes.xml), and root-XML sniffing only refines bytes already magic-classified as * {@code application/xml}. Without the declaration it falls back to text/plain. * From c7a78d6e74bbf6a58b502cc3f8aaf8502f9440e6 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Thu, 24 Sep 2026 10:58:37 +0200 Subject: [PATCH 05/10] #2097 Tika Pipes: fix metadata leak, trimmed output and fork setup - strip the content the fork returns as metadata (tk:content), it was copied to parse.* for every document - keep the text and links of a trimmed fork output instead of failing the document on the XML cut off mid-document, and stop the fork at parser.tika.text.maxlength - one fork per bolt instance: execute() parses one document at a time, so drop parser.tika.pipes.numclients - start the fork with the worker's own java instead of the one on the PATH - exclude the log4j backend brought by tika-pipes-fork-parser, the Storm worker provides it - the fork always maps HTML with Tika's DefaultHtmlMapper, document it and only warn when another mapper is configured - delete the temporary Tika configuration right away when no fork needs it - plugins dir is not needed for documents over 10MB, fix the docs - tests for the metadata, a trimmed parse and a crashed fork --- THIRD-PARTY.txt | 1 - .../resources/archetype-resources/pom.xml | 37 ++++++++ docs/src/main/asciidoc/configuration.adoc | 5 +- .../resources/archetype-resources/pom.xml | 37 ++++++++ .../resources/archetype-resources/pom.xml | 37 ++++++++ external/tika/README.md | 6 +- external/tika/pom.xml | 12 +++ .../apache/stormcrawler/tika/ParserBolt.java | 90 ++++++++++++------- .../tika/ParserBoltPipesTimeoutTest.java | 63 ++++++++++++- 9 files changed, 245 insertions(+), 43 deletions(-) diff --git a/THIRD-PARTY.txt b/THIRD-PARTY.txt index ab9d75d43..d65dc091c 100644 --- a/THIRD-PARTY.txt +++ b/THIRD-PARTY.txt @@ -280,7 +280,6 @@ List of third-party dependencies grouped by their license type. * rome (com.rometools:rome:2.1.0 - http://rometools.com/rome) * rome-utils (com.rometools:rome-utils:2.1.0 - http://rometools.com/rome-utils) * server (org.opensearch:opensearch:2.19.6 - https://github.com/opensearch-project/OpenSearch.git) - * SLF4J 2 Provider for Log4j API (org.apache.logging.log4j:log4j-slf4j2-impl:2.26.1 - https://logging.apache.org/log4j/2.x/) * SmallRye Mutiny Zero (io.smallrye.reactive:mutiny-zero:1.3.1 - https://smallrye.io) * SnakeYAML (org.yaml:snakeyaml:2.7 - https://codeberg.org/snakeyaml/snakeyaml) * snappy-java (org.xerial.snappy:snappy-java:1.1.10.4 - https://github.com/xerial/snappy-java) diff --git a/archetype/src/main/resources/archetype-resources/pom.xml b/archetype/src/main/resources/archetype-resources/pom.xml index 1a7d48eaa..a37373194 100644 --- a/archetype/src/main/resources/archetype-resources/pom.xml +++ b/archetype/src/main/resources/archetype-resources/pom.xml @@ -88,6 +88,43 @@ under the License. + + + META-INF/extensions.idx + + + META-INF/tika/detectors.idx + + + META-INF/tika/encoding-detectors.idx + + + META-INF/tika/language-detectors.idx + + + META-INF/tika/metadata-filters.idx + + + META-INF/tika/parse-context.idx + + + META-INF/tika/parsers.idx + + + META-INF/tika/renderers.idx + + + META-INF/tika/translators.idx + diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index 30504f249..536ef8a4f 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -570,11 +570,10 @@ 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, the parse runs in a forked JVM (Tika Pipes); a parse that exceeds this is killed outright and reported to the status stream as an `ERROR` with the message `parse timeout`. Keep the value below `topology.message.timeout.secs`. -| parser.tika.pipes.numclients | - | Number of forked JVMs to keep under `parser.tika.timeout`. Unset uses Tika's own CPU-derived default. +| parser.tika.timeout | -1 | Maximum time in milliseconds a document may take to parse, 0 or less for no limit. When set, the parse runs in a forked JVM (Tika Pipes), one per bolt instance; a parse that exceeds this is killed outright and reported to the status stream as an `ERROR` with the message `parse timeout`. Keep the value below `topology.message.timeout.secs`. `parser.htmlmapper.classname` is ignored in this mode, see the README of the module. | parser.tika.pipes.jvmargs | - | JVM arguments passed to each forked process under `parser.tika.timeout`, e.g. `-Xmx512m`. | parser.tika.pipes.maxfilesperprocess | - | Restart a forked process after this many documents under `parser.tika.timeout`, to bound slow leaks in parsing libraries. -| parser.tika.pipes.plugins.dir | - | Directory holding Tika Pipes plugin zips. Only needed under `parser.tika.timeout` for documents over the 10MB inline-transfer threshold; unset uses Tika's default plugin directory resolution. +| parser.tika.pipes.plugins.dir | - | Directory holding Tika Pipes plugin zips under `parser.tika.timeout`, not needed by default. Unset uses Tika's default plugin directory resolution. |=== 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/opensearch/archetype/src/main/resources/archetype-resources/pom.xml b/external/opensearch/archetype/src/main/resources/archetype-resources/pom.xml index 94cefffc6..115554aea 100644 --- a/external/opensearch/archetype/src/main/resources/archetype-resources/pom.xml +++ b/external/opensearch/archetype/src/main/resources/archetype-resources/pom.xml @@ -89,6 +89,43 @@ under the License. + + + META-INF/extensions.idx + + + META-INF/tika/detectors.idx + + + META-INF/tika/encoding-detectors.idx + + + META-INF/tika/language-detectors.idx + + + META-INF/tika/metadata-filters.idx + + + META-INF/tika/parse-context.idx + + + META-INF/tika/parsers.idx + + + META-INF/tika/renderers.idx + + + META-INF/tika/translators.idx + diff --git a/external/solr/archetype/src/main/resources/archetype-resources/pom.xml b/external/solr/archetype/src/main/resources/archetype-resources/pom.xml index 14967e13f..107a22859 100644 --- a/external/solr/archetype/src/main/resources/archetype-resources/pom.xml +++ b/external/solr/archetype/src/main/resources/archetype-resources/pom.xml @@ -89,6 +89,43 @@ under the License. + + + META-INF/extensions.idx + + + META-INF/tika/detectors.idx + + + META-INF/tika/encoding-detectors.idx + + + META-INF/tika/language-detectors.idx + + + META-INF/tika/metadata-filters.idx + + + META-INF/tika/parse-context.idx + + + META-INF/tika/parsers.idx + + + META-INF/tika/renderers.idx + + + META-INF/tika/translators.idx + diff --git a/external/tika/README.md b/external/tika/README.md index 0f9f36049..d1a77e446 100644 --- a/external/tika/README.md +++ b/external/tika/README.md @@ -39,10 +39,10 @@ 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 set, the parse runs in a forked JVM via [Tika Pipes](https://tika.apache.org/docs/4.0.x/pipes/index.html): a document that takes longer than the timeout is killed outright (not merely asked to stop) and sent to the status stream as an `ERROR` with the message `parse timeout`, and the fork restarts before the next document. Keep the timeout below `topology.message.timeout.secs` so that the tuple is not replayed while it is being parsed. +The time spent parsing a document can be limited with `parser.tika.timeout` (milliseconds, default `-1`, 0 or less means no limit). When set, the parse runs in a forked JVM via [Tika Pipes](https://tika.apache.org/docs/4.0.x/pipes/index.html), one per bolt instance, started with the same `java` and classpath as the Storm worker: a document that takes longer than the timeout is killed outright (not merely asked to stop) and sent to the status stream as an `ERROR` with the message `parse timeout`, and the fork restarts before the next document. A forked JVM that dies while parsing, e.g. running out of memory, is reported as an `ERROR` with the message `parse crash` and restarted as well; the Storm worker is not affected. Keep the timeout below `topology.message.timeout.secs` so that the tuple is not replayed while it is being parsed. -A handful of related keys tune the forked JVMs: `parser.tika.pipes.numclients` (how many to keep running, default is Tika's own CPU-derived count), `parser.tika.pipes.jvmargs` (e.g. `-Xmx512m`), `parser.tika.pipes.maxfilesperprocess` (restart a fork after this many documents, to bound slow leaks in parsing libraries), and `parser.tika.pipes.plugins.dir` (only needed for documents over the 10MB inline-transfer threshold; most crawled pages never hit it). +Each forked JVM needs memory on top of the worker, so size the hosts for the number of `ParserBolt` executors. A handful of related keys tune the forked JVMs: `parser.tika.pipes.jvmargs` (e.g. `-Xmx512m`), `parser.tika.pipes.maxfilesperprocess` (restart a fork after this many documents, to bound slow leaks in parsing libraries), and `parser.tika.pipes.plugins.dir` (Tika Pipes plugins, not needed by default). Documents over 10MB are handed to the forked JVM through a temporary file in the worker's `java.io.tmpdir`. -`parser.htmlmapper.classname` is not applied to parses running under `parser.tika.timeout`: a live `HtmlMapper` instance cannot be sent to the forked JVM. Configure it in the `"parse-context"` section of the Tika configuration file instead if you need it there. +`parser.htmlmapper.classname` is not applied to parses running under `parser.tika.timeout`: an `HtmlMapper` cannot be passed to the forked JVM, which always uses Tika's `DefaultHtmlMapper`. That mapper drops the elements it does not consider safe from the DOM given to the parse filters, so XPath expressions written against the default `IdentityHtmlMapper` output may need adjusting. 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/pom.xml b/external/tika/pom.xml index 284eb647d..f2feb2b7e 100644 --- a/external/tika/pom.xml +++ b/external/tika/pom.xml @@ -66,6 +66,18 @@ under the License. org.apache.tika tika-pipes-fork-parser ${tika.version} + + + + org.apache.logging.log4j + log4j-core + + + org.apache.logging.log4j + log4j-slf4j2-impl + + - META-INF/extensions.idx + META-INF/extensions.idx - META-INF/tika/detectors.idx + META-INF/tika/detectors.idx - META-INF/tika/encoding-detectors.idx + META-INF/tika/encoding-detectors.idx - META-INF/tika/language-detectors.idx + META-INF/tika/language-detectors.idx - META-INF/tika/metadata-filters.idx + META-INF/tika/metadata-filters.idx - META-INF/tika/parse-context.idx + META-INF/tika/parse-context.idx - META-INF/tika/parsers.idx + META-INF/tika/parsers.idx - META-INF/tika/renderers.idx + META-INF/tika/renderers.idx - META-INF/tika/translators.idx + META-INF/tika/translators.idx