TIKA-4497 -- allow per file timeouts in tika-pipes (#2343)

Author: tballison

CHANGES.txt

@@ -15,6 +15,8 @@ Release 4.0.0-BETA1 - ???
 
 Release 3.3.0 - ???
 
+  * Allow per file timeouts in tika-pipes (TIKA-4497).
+
   * Add matroska detector (TIKA-1180).
 
   * Allow multiple values for many Dublin Core keys (TIKA-4466).

tika-core/src/main/java/org/apache/tika/config/TikaTaskTimeout.java

@@ -16,9 +16,11 @@
  */
 package org.apache.tika.config;
 
+import java.io.Serializable;
+
 import org.apache.tika.parser.ParseContext;
 
-public class TikaTaskTimeout {
+public class TikaTaskTimeout implements Serializable {
 
     private final long timeoutMillis;
 
@@ -31,6 +33,9 @@ public long getTimeoutMillis() {
     }
 
     public static long getTimeoutMillis(ParseContext context, long defaultTimeoutMillis) {
+        if (context == null) {
+            return defaultTimeoutMillis;
+        }
         TikaTaskTimeout tikaTaskTimeout = context.get(TikaTaskTimeout.class);
         if (tikaTaskTimeout == null) {
             return defaultTimeoutMillis;

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java

@@ -47,6 +47,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.tika.config.TikaTaskTimeout;
 import org.apache.tika.metadata.Metadata;
 import org.apache.tika.metadata.TikaCoreProperties;
 import org.apache.tika.parser.ParseContext;
@@ -200,7 +201,8 @@ private PipesResult actuallyProcess(FetchEmitTuple t) throws InterruptedExceptio
                         ": PipesClient closed");
             }
             executorService.execute(futureTask);
-            return futureTask.get(pipesConfig.getTimeoutMillis(), TimeUnit.MILLISECONDS);
+            long timeout = TikaTaskTimeout.getTimeoutMillis(t.getParseContext(), pipesConfig.getTimeoutMillis());
+            return futureTask.get(timeout, TimeUnit.MILLISECONDS);
         } catch (InterruptedException e) {
             destroyForcibly();
             throw e;

tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesServer.java

@@ -41,6 +41,7 @@
 import org.xml.sax.SAXException;
 
 import org.apache.tika.config.TikaConfig;
+import org.apache.tika.config.TikaTaskTimeout;
 import org.apache.tika.detect.Detector;
 import org.apache.tika.exception.EncryptedDocumentException;
 import org.apache.tika.exception.TikaConfigException;
@@ -135,8 +136,9 @@ public static STATUS lookup(int val) {
     //if it is smaller than this value, write it back to the
     //PipesClient so that it can cache the extracts and then batch emit.
     private final long maxForEmitBatchBytes;
-    private final long serverParseTimeoutMillis;
+    private final long defaultServerParseTimeoutMillis;
     private final long serverWaitTimeoutMillis;
+    private volatile long serverParseTimeoutMillis;
     private Parser autoDetectParser;
     private Parser rMetaParser;
     private TikaConfig tikaConfig;
@@ -154,6 +156,7 @@ public PipesServer(Path tikaConfigPath, InputStream in, PrintStream out,
         this.input = new DataInputStream(in);
         this.output = new DataOutputStream(out);
         this.maxForEmitBatchBytes = maxForEmitBatchBytes;
+        this.defaultServerParseTimeoutMillis = serverParseTimeoutMillis;
         this.serverParseTimeoutMillis = serverParseTimeoutMillis;
         this.serverWaitTimeoutMillis = serverWaitTimeoutMillis;
         this.parsing = false;
@@ -189,9 +192,9 @@ public void run() {
             while (true) {
                 synchronized (lock) {
                     long elapsed = System.currentTimeMillis() - since;
-                    if (parsing && elapsed > serverParseTimeoutMillis) {
-                        LOG.warn("timeout server; elapsed {}  with {}", elapsed,
-                                serverParseTimeoutMillis);
+                    long timeout = serverParseTimeoutMillis;
+                    if (parsing && elapsed > timeout) {
+                        LOG.warn("timeout server; elapsed {}ms  with timeout={}", elapsed, timeout);
                         exit(TIMEOUT_EXIT_CODE);
                     } else if (!parsing && serverWaitTimeoutMillis > 0 &&
                             elapsed > serverWaitTimeoutMillis) {
@@ -341,6 +344,7 @@ private void parseOne() {
                         System.currentTimeMillis() - start);
             }
             start = System.currentTimeMillis();
+            updateTimeout(t);
             actuallyParse(t);
             if (LOG.isTraceEnabled()) {
                 LOG.trace("timer -- actually parsed: {} ms", System.currentTimeMillis() - start);
@@ -351,10 +355,32 @@ private void parseOne() {
             synchronized (lock) {
                 parsing = false;
                 since = System.currentTimeMillis();
+                serverParseTimeoutMillis = defaultServerParseTimeoutMillis;
             }
         }
     }
 
+    private void updateTimeout(FetchEmitTuple t) {
+        if (t == null) {
+            return;
+        }
+
+        ParseContext parseContext = t.getParseContext();
+        if (parseContext == null) {
+            return;
+        }
+
+        TikaTaskTimeout tikaTaskTimeout = parseContext.get(TikaTaskTimeout.class);
+        if (tikaTaskTimeout == null) {
+            return;
+        }
+
+        synchronized (lock) {
+            serverParseTimeoutMillis = tikaTaskTimeout.getTimeoutMillis();
+            LOG.debug("setting per file timeout {}ms", serverParseTimeoutMillis);
+        }
+    }
+
     private void actuallyParse(FetchEmitTuple t) {
 
         long start = System.currentTimeMillis();

tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java

@@ -16,6 +16,8 @@
  */
 package org.apache.tika.pipes.core;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 import java.io.IOException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -27,6 +29,7 @@
 import org.junit.jupiter.api.Test;
 import org.xml.sax.SAXException;
 
+import org.apache.tika.config.TikaTaskTimeout;
 import org.apache.tika.exception.TikaConfigException;
 import org.apache.tika.metadata.Metadata;
 import org.apache.tika.metadata.TikaCoreProperties;
@@ -62,9 +65,9 @@ public void testBasic() throws IOException, InterruptedException {
                 new FetchEmitTuple(testPdfFile, new FetchKey(fetcherName, testPdfFile),
                         new EmitKey(), new Metadata(), new ParseContext(), FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
         Assertions.assertNotNull(pipesResult.getEmitData().getMetadataList());
-        Assertions.assertEquals(1, pipesResult.getEmitData().getMetadataList().size());
+        assertEquals(1, pipesResult.getEmitData().getMetadataList().size());
         Metadata metadata = pipesResult.getEmitData().getMetadataList().get(0);
-        Assertions.assertEquals("testOverlappingText.pdf", metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY));
+        assertEquals("testOverlappingText.pdf", metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY));
     }
 
     @Test
@@ -76,9 +79,9 @@ public void testMetadataFilter() throws IOException, InterruptedException {
                 new FetchEmitTuple(testPdfFile, new FetchKey(fetcherName, testPdfFile),
                         new EmitKey(), new Metadata(), parseContext, FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
         Assertions.assertNotNull(pipesResult.getEmitData().getMetadataList());
-        Assertions.assertEquals(1, pipesResult.getEmitData().getMetadataList().size());
+        assertEquals(1, pipesResult.getEmitData().getMetadataList().size());
         Metadata metadata = pipesResult.getEmitData().getMetadataList().get(0);
-        Assertions.assertEquals("TESTOVERLAPPINGTEXT.PDF", metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY));
+        assertEquals("TESTOVERLAPPINGTEXT.PDF", metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY));
     }
 
     @Test
@@ -90,8 +93,23 @@ public void testMetadataListFilter() throws IOException, InterruptedException {
                 new FetchEmitTuple("mock/embedded.xml", new FetchKey(fetcherName, "mock/embedded.xml"),
                         new EmitKey(), new Metadata(), parseContext, FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
         Assertions.assertNotNull(pipesResult.getEmitData().getMetadataList());
-        Assertions.assertEquals(5, pipesResult.getEmitData().getMetadataList().size());
+        assertEquals(5, pipesResult.getEmitData().getMetadataList().size());
         Metadata metadata = pipesResult.getEmitData().getMetadataList().get(0);
-        Assertions.assertEquals(4, Integer.parseInt(metadata.get("X-TIKA:attachment_count")));
+        assertEquals(4, Integer.parseInt(metadata.get("X-TIKA:attachment_count")));
+    }
+
+    @Test
+    public void testTimeout() throws IOException, InterruptedException {
+        //TODO -- add unit test for timeout > default
+        //TODO -- figure out how to test pipes server timeout alone
+        //I did both manually during development, but unit tests are better. :D
+        ParseContext parseContext = new ParseContext();
+        parseContext.set(TikaTaskTimeout.class, new TikaTaskTimeout(1000));
+        MetadataListFilter metadataFilter = new CompositeMetadataListFilter(List.of(new AttachmentCountingListFilter()));
+        parseContext.set(MetadataListFilter.class, metadataFilter);
+        PipesResult pipesResult = pipesClient.process(
+                new FetchEmitTuple("mock/timeout-10s.xml", new FetchKey(fetcherName, "mock/timeout-10s.xml"),
+                        new EmitKey(), new Metadata(), parseContext, FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+        assertEquals(PipesResult.TIMEOUT.getStatus(), pipesResult.getStatus());
     }
 }

tika-pipes/tika-pipes-core/src/test/resources/test-documents/mock/timeout-10s.xml

@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+
+<!--
+  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.
+-->
+
+<mock>
+
+    <metadata action="add" name="dc:creator">Nikolai Lobachevsky</metadata>
+    <write element="p" times="30"> hello </write>
+    <fakeload millis="10000" cpu="10" mb="10"/>
+</mock>