jvmArgs =
+ new ArrayList<>(ConfUtils.loadListFromConf(PIPES_JVM_ARGS_PARAM, conf));
+ if (jvmArgs.stream().noneMatch(ParserBolt::setsHeap)) {
+ jvmArgs.add(PIPES_DEFAULT_HEAP);
+ }
+ return jvmArgs;
+ }
+
+ private static boolean setsHeap(String jvmArg) {
+ return jvmArg.startsWith("-Xmx")
+ || jvmArg.startsWith("-XX:MaxHeapSize")
+ || jvmArg.startsWith("-XX:MaxRAM");
+ }
+
+ /**
+ * Parses a one-word document so that a fork which cannot start fails the bolt here, instead of
+ * failing every document with a "parse pipes error" status. The fork starts lazily otherwise.
+ */
+ private void startFork() {
+ final PipesForkResult result;
+ try (TikaInputStream tis =
+ TikaInputStream.get("StormCrawler".getBytes(StandardCharsets.UTF_8))) {
+ result = pipesForkParser.parse(tis);
+ } catch (IOException | TikaException | PipesException e) {
+ throw new IllegalStateException(
+ "The Tika Pipes fork for " + PARSE_TIMEOUT_PARAM + " did not start", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(
+ "Interrupted while starting the Tika Pipes fork for " + PARSE_TIMEOUT_PARAM, e);
+ }
+ if (!result.isSuccess()) {
+ throw new IllegalStateException(
+ "The Tika Pipes fork for "
+ + PARSE_TIMEOUT_PARAM
+ + " failed a test parse: "
+ + describe(result));
+ }
+ }
+
+ private static String describe(PipesForkResult result) {
+ return result.getStatus()
+ + (result.getMessage() != null ? " - " + result.getMessage() : "");
+ }
+
+ /**
+ * 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, 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,
+ 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: " + describe(result));
+ }
+
+ if (!result.isSuccess()) {
+ throw new IOException("Tika Pipes parse of " + url + " failed: " + describe(result));
+ }
+
+ org.apache.tika.metadata.Metadata resultMetadata = result.getMetadata();
+
+ if (result.getStatus() == PipesResult.RESULT_STATUS.PARSE_SUCCESS_WITH_EXCEPTION
+ && !isWriteLimitReached(resultMetadata)) {
+ // 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
+ + " threw: "
+ + (resultMetadata != null
+ ? resultMetadata.get(TikaCoreProperties.CONTAINER_EXCEPTION)
+ : result.getMessage()));
+ }
+
+ boolean trimmed =
+ result.getStatus() == PipesResult.RESULT_STATUS.PARSE_SUCCESS_WITH_EXCEPTION
+ || result.getStatus() == PipesResult.RESULT_STATUS.PARTIAL_TIMEOUT;
+
+ String xml = result.getContent();
+ if (StringUtils.isNotBlank(xml)) {
+ try {
+ reparseIntoHandler(xml, handler);
+ } catch (SAXException e) {
+ if (WriteLimitReachedException.isWriteLimitReached(e)) {
+ trimmed = true;
+ } else if (!(trimmed && e instanceof SAXParseException)) {
+ throw e;
}
+ // a trimmed fork returns XML cut off mid-document: the handlers keep
+ // what they got before the cut, as with a trimmed direct parse
}
}
+
+ org.apache.tika.metadata.Metadata md =
+ resultMetadata != null ? resultMetadata : seedMetadata;
+ // the fork returns the content as metadata, it must not end up in parse.*
+ md.remove(TikaCoreProperties.TIKA_CONTENT.getName());
+ md.remove(TikaCoreProperties.TIKA_CONTENT_HANDLER_TYPE.getName());
+ return new PipesParseOutcome(md, trimmed);
+ }
+
+ private static boolean isWriteLimitReached(org.apache.tika.metadata.Metadata metadata) {
+ return metadata != null
+ && "true".equalsIgnoreCase(metadata.get(TikaCoreProperties.WRITE_LIMIT_REACHED));
+ }
+
+ /**
+ * 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 {
+ 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 +885,27 @@ 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);
+ }
+ }
+ deleteTemporaryTikaConfig();
+ }
+
+ private void deleteTemporaryTikaConfig() {
+ if (resolvedTikaConfigPathIsTemporary && resolvedTikaConfigPath != null) {
+ try {
+ Files.deleteIfExists(resolvedTikaConfigPath);
+ } catch (IOException e) {
+ LOG.warn(
+ "Failed to delete temporary Tika configuration {}",
+ resolvedTikaConfigPath,
+ e);
+ }
+ resolvedTikaConfigPathIsTemporary = false;
+ }
}
}
diff --git a/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesJvmArgsTest.java b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesJvmArgsTest.java
new file mode 100644
index 000000000..c6eeb3f2d
--- /dev/null
+++ b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesJvmArgsTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class ParserBoltPipesJvmArgsTest {
+
+ @Test
+ void defaultHeapWhenNoArgumentIsSet() {
+ Assertions.assertEquals(List.of("-Xmx512m"), ParserBolt.forkedJvmArgs(Map.of()));
+ }
+
+ @Test
+ void defaultHeapIsAddedToOtherArguments() {
+ Assertions.assertEquals(
+ List.of("-Djava.awt.headless=true", "-Xmx512m"),
+ ParserBolt.forkedJvmArgs(
+ Map.of(ParserBolt.PIPES_JVM_ARGS_PARAM, "-Djava.awt.headless=true")));
+ }
+
+ @Test
+ void configuredHeapIsKept() {
+ Assertions.assertEquals(
+ List.of("-Xmx1g"),
+ ParserBolt.forkedJvmArgs(Map.of(ParserBolt.PIPES_JVM_ARGS_PARAM, "-Xmx1g")));
+ }
+
+ @Test
+ void configuredMaxHeapSizeIsKept() {
+ Assertions.assertEquals(
+ List.of("-XX:MaxHeapSize=2g"),
+ ParserBolt.forkedJvmArgs(
+ Map.of(ParserBolt.PIPES_JVM_ARGS_PARAM, "-XX:MaxHeapSize=2g")));
+ }
+
+ @Test
+ void configuredRamPercentageIsKept() {
+ Assertions.assertEquals(
+ List.of("-XX:MaxRAMPercentage=25"),
+ ParserBolt.forkedJvmArgs(
+ Map.of(
+ ParserBolt.PIPES_JVM_ARGS_PARAM,
+ List.of("-XX:MaxRAMPercentage=25"))));
+ }
+}
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..0ce52b5b7
--- /dev/null
+++ b/external/tika/src/test/java/org/apache/stormcrawler/tika/ParserBoltPipesTimeoutTest.java
@@ -0,0 +1,238 @@
+/*
+ * 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;
+
+/**
+ * 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 } 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
+ */
+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);
+ conf.putIfAbsent(ParserBolt.PIPES_JVM_ARGS_PARAM, "-Xmx256m");
+ bolt.prepare(conf, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+ }
+
+ /** A 2s timeout must stop the bolt well before the hang's own 60s 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());
+ }
+
+ /** Parses normally within the timeout: text and outlinks are still emitted. */
+ @Test
+ @Timeout(30)
+ void documentIsParsedUnderTimeout() throws IOException {
+ prepare(new HashMap<>());
+
+ String url = "https://example.org/fast.html";
+ byte[] content =
+ ("thello 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"));
+ Metadata parseMetadata = (Metadata) emitted.get(0).get(2);
+ Assertions.assertEquals("t", parseMetadata.getFirstValue("parse.dc:title"));
+ // the fork returns the content as metadata, it must not be copied to parse.*
+ Assertions.assertNull(parseMetadata.getFirstValue("parse.tk:content"));
+ Assertions.assertNull(parseMetadata.getFirstValue("parse.tk:content-handler-type"));
+
+ 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));
+ }
+
+ /**
+ * The fork stops at parser.tika.text.maxlength and returns XML cut off mid-document: the
+ * document is still emitted, trimmed, instead of failing on the truncated XML.
+ */
+ @Test
+ @Timeout(30)
+ void textIsTrimmedUnderTimeout() throws IOException {
+ Map conf = new HashMap<>();
+ conf.put(ParserBolt.TEXT_MAX_LENGTH_PARAM, 5);
+ prepare(conf);
+
+ String url = "https://example.org/long.html";
+ byte[] content =
+ ("thello world
"
+ + "more text after the limit
")
+ .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.assertFalse(emitted.get(0).get(3).toString().contains("more text"));
+ Metadata parseMetadata = (Metadata) emitted.get(0).get(2);
+ Assertions.assertEquals("true", parseMetadata.getFirstValue(ParserBolt.TEXT_TRIMMED_KEY));
+ }
+
+ /** A fork which cannot start fails the bolt in prepare, not every document afterwards. */
+ @Test
+ @Timeout(60)
+ void forkThatCannotStartFailsPrepare() {
+ Map conf = new HashMap<>();
+ conf.put(ParserBolt.PIPES_JVM_ARGS_PARAM, "-XX:NoSuchOption");
+ Assertions.assertThrows(IllegalStateException.class, () -> prepare(conf));
+ }
+
+ /** A forked JVM dying mid-parse is reported as "parse crash" and the bolt carries on. */
+ @Test
+ @Timeout(60)
+ void crashedForkIsReportedUnderTimeout() throws IOException {
+ prepare(new HashMap<>());
+
+ String url = "https://example.org/crash.xml";
+ byte[] content =
+ (XML_DECLARATION + "").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 crash", md.getFirstValue(Constants.STATUS_ERROR_MESSAGE));
+
+ // the fork restarts for the next document
+ parse(
+ "https://example.org/fast.html",
+ "hello again
".getBytes(StandardCharsets.UTF_8),
+ new Metadata());
+ List> emitted = output.getEmitted();
+ Assertions.assertEquals(1, emitted.size());
+ Assertions.assertTrue(emitted.get(0).get(3).toString().contains("hello again"));
+ }
+
+ /** 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");
+ }
+}