You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ta...@apache.org on 2023/07/05 20:44:36 UTC
[tika] 01/01: TIKA-4097 -- handle orphaned temp files more robustly
This is an automated email from the ASF dual-hosted git repository.
tallison pushed a commit to branch TIKA-4097
in repository https://gitbox.apache.org/repos/asf/tika.git
commit 1b340a6c365934809a6f4a1bb2343d0cfa49e7c2
Author: tballison <ta...@apache.org>
AuthorDate: Wed Jul 5 16:44:27 2023 -0400
TIKA-4097 -- handle orphaned temp files more robustly
---
.../java/org/apache/tika/pipes/PipesClient.java | 31 ++++++++++++++++++++++
.../org/apache/tika/pipes/PipesConfigBase.java | 19 +++++++++++++
.../apache/tika/pipes/async/AsyncProcessor.java | 15 +++++++++++
3 files changed, 65 insertions(+)
diff --git a/tika-core/src/main/java/org/apache/tika/pipes/PipesClient.java b/tika-core/src/main/java/org/apache/tika/pipes/PipesClient.java
index 1e3147370..9552ba1f6 100644
--- a/tika-core/src/main/java/org/apache/tika/pipes/PipesClient.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/PipesClient.java
@@ -29,6 +29,8 @@ import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -41,6 +43,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.io.file.Counters;
+import org.apache.commons.io.file.PathUtils;
import org.apache.commons.io.input.UnsynchronizedByteArrayInputStream;
import org.apache.commons.io.output.UnsynchronizedByteArrayOutputStream;
import org.slf4j.Logger;
@@ -77,10 +81,18 @@ public class PipesClient implements Closeable {
private DataOutputStream output;
private DataInputStream input;
private int filesProcessed = 0;
+ //this is the client-specific subdirectory of the pipesConfig's getPipesTmpDir
+ final Path tmpDir;
public PipesClient(PipesConfigBase pipesConfig) {
this.pipesConfig = pipesConfig;
this.pipesClientId = CLIENT_COUNTER.getAndIncrement();
+ try {
+ tmpDir = Files.createTempDirectory(pipesConfig.getPipesTmpDir(),
+ "client-" + this.pipesClientId);
+ } catch (IOException e) {
+ throw new RuntimeException("Couldn't create temp dir?!", e);
+ }
}
public int getFilesProcessed() {
@@ -277,6 +289,19 @@ public class PipesClient implements Closeable {
if (process.isAlive()) {
LOG.error("Process still alive after {}ms", WAIT_ON_DESTROY_MS);
}
+ try {
+ if (Files.isDirectory(tmpDir)) {
+ LOG.debug("about to delete the full async temp directory: {}",
+ pipesConfig.getPipesTmpDir().toAbsolutePath());
+ Counters.PathCounters pathCounters =
+ PathUtils.deleteDirectory(pipesConfig.getPipesTmpDir());
+ LOG.debug("Successfully deleted {} temporary files in {} directories",
+ pathCounters.getFileCounter().get(),
+ pathCounters.getDirectoryCounter().get());
+ }
+ } catch (IllegalArgumentException | IOException e) {
+ LOG.warn("Failed to delete temporary directory: " + tmpDir.toAbsolutePath(), e);
+ }
}
private PipesResult readResults(FetchEmitTuple t, long start) throws IOException {
@@ -524,6 +549,11 @@ public class PipesClient implements Closeable {
origGCString = arg;
newGCLogString = arg.replace("${pipesClientId}", "id-" + pipesClientId);
}
+ if (arg.startsWith("-Djava.io.tmpdir=")) {
+ throw new IllegalArgumentException("Can't specify java.io.tmpdir in jvmargs. Set " +
+ "the overall tmpdir for all async process and its forked processes in the" +
+ "<pipesTmpDir/> attribute.");
+ }
}
if (origGCString != null && newGCLogString != null) {
@@ -552,6 +582,7 @@ public class PipesClient implements Closeable {
"-Dlog4j.configurationFile=classpath:pipes-fork-server-default-log4j2.xml");
}
commandLine.add("-DpipesClientId=" + pipesClientId);
+ commandLine.add("-Djava.io.tmpdir=" + ProcessUtils.escapeCommandLine(tmpDir.toAbsolutePath().toString()));
commandLine.addAll(configArgs);
commandLine.add("org.apache.tika.pipes.PipesServer");
commandLine.add(ProcessUtils.escapeCommandLine(
diff --git a/tika-core/src/main/java/org/apache/tika/pipes/PipesConfigBase.java b/tika-core/src/main/java/org/apache/tika/pipes/PipesConfigBase.java
index bf6a6bb69..6fd6cc9b4 100644
--- a/tika-core/src/main/java/org/apache/tika/pipes/PipesConfigBase.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/PipesConfigBase.java
@@ -16,6 +16,8 @@
*/
package org.apache.tika.pipes;
+import java.io.IOException;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
@@ -59,6 +61,8 @@ public class PipesConfigBase extends ConfigBase {
private Path tikaConfig;
private String javaPath = "java";
+ private Path pipesTmpDir = null;
+
public long getTimeoutMillis() {
return timeoutMillis;
}
@@ -171,4 +175,19 @@ public class PipesConfigBase extends ConfigBase {
public void setSleepOnStartupTimeoutMillis(long sleepOnStartupTimeoutMillis) {
this.sleepOnStartupTimeoutMillis = sleepOnStartupTimeoutMillis;
}
+
+ public void setPipesTmpDir(Path pipesTmpDir) {
+ this.pipesTmpDir = pipesTmpDir;
+ }
+
+ public Path getPipesTmpDir() throws IOException {
+ if (pipesTmpDir == null) {
+ pipesTmpDir = Files.createTempDirectory("tika-pipes-tmp-dir");
+ } else {
+ if (! Files.isDirectory(pipesTmpDir)) {
+ Files.createDirectories(pipesTmpDir);
+ }
+ }
+ return pipesTmpDir;
+ }
}
diff --git a/tika-core/src/main/java/org/apache/tika/pipes/async/AsyncProcessor.java b/tika-core/src/main/java/org/apache/tika/pipes/async/AsyncProcessor.java
index 3a6751f4f..5b602a94c 100644
--- a/tika-core/src/main/java/org/apache/tika/pipes/async/AsyncProcessor.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/async/AsyncProcessor.java
@@ -18,6 +18,7 @@ package org.apache.tika.pipes.async;
import java.io.Closeable;
import java.io.IOException;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
@@ -30,6 +31,8 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.commons.io.file.Counters;
+import org.apache.commons.io.file.PathUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -252,6 +255,18 @@ public class AsyncProcessor implements Closeable {
public void close() throws IOException {
executorService.shutdownNow();
asyncConfig.getPipesReporter().close();
+ if (Files.isDirectory(asyncConfig.getPipesTmpDir())) {
+ try {
+ LOG.debug("about to delete the full async temp directory: {}",
+ asyncConfig.getPipesTmpDir().toAbsolutePath());
+ Counters.PathCounters pathCounters = PathUtils.deleteDirectory(asyncConfig.getPipesTmpDir());
+ LOG.debug("Successfully deleted {} temporary files in {} directories",
+ pathCounters.getFileCounter().get(),
+ pathCounters.getDirectoryCounter().get());
+ } catch (IllegalArgumentException e) {
+ throw new IOException(e);
+ }
+ }
}
public long getTotalProcessed() {