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 2026/08/27 20:28:18 UTC
(tika) branch TIKA-4848-fs-emitter-atomic created (now da1aa89f5f)
This is an automated email from the ASF dual-hosted git repository.
tballison pushed a change to branch TIKA-4848-fs-emitter-atomic
in repository https://gitbox.apache.org/repos/asf/tika.git
at da1aa89f5f TIKA-4848: FileSystemEmitter writes to a tmp file and renames atomically so readers never see a partial output
This branch includes the following new commits:
new da1aa89f5f TIKA-4848: FileSystemEmitter writes to a tmp file and renames atomically so readers never see a partial output
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tika) 01/01: TIKA-4848: FileSystemEmitter writes to a tmp file and renames atomically so readers never see a partial output
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch TIKA-4848-fs-emitter-atomic
in repository https://gitbox.apache.org/repos/asf/tika.git
commit da1aa89f5f2ae670726d2e5c96a152395cc51a61
Author: tallison <ta...@apache.org>
AuthorDate: Thu Aug 27 16:27:56 2026 -0400
TIKA-4848: FileSystemEmitter writes to a tmp file and renames atomically so readers never see a partial output
---
CHANGES.txt | 4 +
.../tika/pipes/emitter/fs/FileSystemEmitter.java | 73 ++++++++++-----
.../pipes/emitter/fs/FileSystemEmitterTest.java | 102 ++++++++++++++++++++-
3 files changed, 153 insertions(+), 26 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 3b336d2e81..9699621a62 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,9 @@
Release 4.1.0 - unreleased
+ * FileSystemEmitter writes to a sibling ".tmp" file and renames it into
+ place, so readers of the output directory never see a partially written
+ file (TIKA-4848).
+
* Stop spooling OLE2 objects whose header over-reserves BAT capacity
(TIKA-4845).
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java
index 3e0f8d1a9b..211a84e263 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java
@@ -27,6 +27,7 @@ import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.util.List;
+import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,6 +54,9 @@ import org.apache.tika.utils.StringUtils;
*/
public class FileSystemEmitter extends AbstractStreamEmitter {
+ // in-progress writes; crawlers of the output dir should ignore these
+ static final String TMP_SUFFIX = ".tmp";
+
private static final Logger LOG = LoggerFactory.getLogger(FileSystemEmitter.class);
public static FileSystemEmitter build(ExtensionConfig pluginConfig) throws TikaConfigException, IOException {
@@ -127,17 +131,44 @@ public class FileSystemEmitter extends AbstractStreamEmitter {
}
}
- if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.EXCEPTION) {
- try (Writer writer = Files.newBufferedWriter(output, StandardCharsets.UTF_8,
- StandardOpenOption.CREATE_NEW)) { //CREATE_NEW forces an IOException if the file already exists
+ Path tmp = tmpFor(output);
+ try {
+ try (Writer writer = Files.newBufferedWriter(tmp, StandardCharsets.UTF_8,
+ StandardOpenOption.CREATE_NEW)) {
JsonMetadataList.toJson(metadataList, writer, config.prettyPrint());
- } catch (FileAlreadyExistsException e) {
- throw alreadyExistsException(output);
}
- } else {
- try (Writer writer = Files.newBufferedWriter(output, StandardCharsets.UTF_8)) {
- JsonMetadataList.toJson(metadataList, writer, config.prettyPrint());
+ publish(tmp, output, config.onExists());
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+
+ private static Path tmpFor(Path output) {
+ // sibling so the rename stays on one filesystem (and therefore atomic)
+ return output.resolveSibling(output.getFileName() + "." + UUID.randomUUID() + TMP_SUFFIX);
+ }
+
+ /**
+ * Moves the fully written {@code tmp} onto {@code output} with a single rename, so a
+ * concurrent reader never sees a partial file. Ownership of {@code tmp} passes to this
+ * method: it is gone on return, whether moved or discarded.
+ */
+ private static void publish(Path tmp, Path output, FileSystemEmitterConfig.ON_EXISTS onExists)
+ throws IOException {
+ if (onExists == FileSystemEmitterConfig.ON_EXISTS.REPLACE) {
+ Files.move(tmp, output, StandardCopyOption.REPLACE_EXISTING,
+ StandardCopyOption.ATOMIC_MOVE);
+ return;
+ }
+ // no REPLACE_EXISTING: Files.move refuses an existing target rather than clobbering it
+ try {
+ Files.move(tmp, output);
+ } catch (FileAlreadyExistsException e) {
+ Files.deleteIfExists(tmp);
+ if (onExists == FileSystemEmitterConfig.ON_EXISTS.EXCEPTION) {
+ throw alreadyExistsException(output);
}
+ LOG.debug("Skipping existing file: {}", output);
}
}
@@ -174,22 +205,16 @@ public class FileSystemEmitter extends AbstractStreamEmitter {
if (!Files.isDirectory(output.getParent())) {
Files.createDirectories(output.getParent());
}
- if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.REPLACE) {
- Files.copy(inputStream, output, StandardCopyOption.REPLACE_EXISTING);
- } else if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.EXCEPTION) {
- try {
- Files.copy(inputStream, output);
- } catch (FileAlreadyExistsException e) {
- throw alreadyExistsException(output);
- }
- } else if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.SKIP) {
- if (!Files.isRegularFile(output)) {
- try {
- Files.copy(inputStream, output);
- } catch (FileAlreadyExistsException e) {
- //swallow
- }
- }
+ if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.SKIP && Files.exists(output)) {
+ LOG.debug("Skipping existing file: {}", output);
+ return;
+ }
+ Path tmp = tmpFor(output);
+ try {
+ Files.copy(inputStream, tmp);
+ publish(tmp, output, config.onExists());
+ } finally {
+ Files.deleteIfExists(tmp);
}
}
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java
index 7695a193a9..3ba310b3d2 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java
@@ -16,13 +16,18 @@
*/
package org.apache.tika.pipes.emitter.fs;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
+import java.util.stream.Stream;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -33,6 +38,7 @@ import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.pipes.api.emitter.Emitter;
+import org.apache.tika.pipes.api.emitter.StreamEmitter;
import org.apache.tika.plugins.ExtensionConfig;
public class FileSystemEmitterTest {
@@ -44,6 +50,11 @@ public class FileSystemEmitterTest {
private Emitter createEmitter(Path basePath, Boolean allowAbsolutePaths)
throws TikaConfigException, IOException {
+ return createEmitter(basePath, allowAbsolutePaths, "REPLACE");
+ }
+
+ private StreamEmitter createEmitter(Path basePath, Boolean allowAbsolutePaths, String onExists)
+ throws TikaConfigException, IOException {
ObjectNode config = MAPPER.createObjectNode();
if (basePath != null) {
config.put("basePath", basePath.toAbsolutePath().toString());
@@ -51,9 +62,9 @@ public class FileSystemEmitterTest {
if (allowAbsolutePaths != null) {
config.put("allowAbsolutePaths", allowAbsolutePaths);
}
- config.put("onExists", "REPLACE");
+ config.put("onExists", onExists);
ExtensionConfig pluginConfig = new ExtensionConfig("test", "test", config.toString());
- return new FileSystemEmitterFactory().buildExtension(pluginConfig);
+ return (StreamEmitter) new FileSystemEmitterFactory().buildExtension(pluginConfig);
}
@Test
@@ -82,4 +93,91 @@ public class FileSystemEmitterTest {
assertThrows(IOException.class, () -> emitter.emit(
"../escaped.json", List.of(new Metadata()), new ParseContext()));
}
+
+ private Path seed(Path basePath, String name, String content) throws IOException {
+ Files.createDirectories(basePath);
+ Path existing = basePath.resolve(name);
+ Files.writeString(existing, content);
+ return existing;
+ }
+
+ private static long tmpFiles(Path dir) throws IOException {
+ try (Stream<Path> s = Files.list(dir)) {
+ return s.filter(p -> p.getFileName().toString().endsWith(FileSystemEmitter.TMP_SUFFIX))
+ .count();
+ }
+ }
+
+ @Test
+ public void testOnExistsExceptionLeavesOriginalIntact() throws Exception {
+ Path basePath = tempDir.resolve("base");
+ Path existing = seed(basePath, "a.json", "original");
+ StreamEmitter emitter = createEmitter(basePath, null, "EXCEPTION");
+ assertThrows(IOException.class, () ->
+ emitter.emit("a.json", List.of(new Metadata()), new ParseContext()));
+ assertThrows(IOException.class, () -> emitter.emit("a.json",
+ new ByteArrayInputStream("x".getBytes(StandardCharsets.UTF_8)), new Metadata(),
+ new ParseContext()));
+ assertEquals("original", Files.readString(existing));
+ assertEquals(0, tmpFiles(basePath), "tmp file leaked");
+ }
+
+ @Test
+ public void testOnExistsSkipLeavesOriginalIntact() throws Exception {
+ Path basePath = tempDir.resolve("base");
+ Path existing = seed(basePath, "a.json", "original");
+ StreamEmitter emitter = createEmitter(basePath, null, "SKIP");
+ emitter.emit("a.json", List.of(new Metadata()), new ParseContext());
+ emitter.emit("a.json", new ByteArrayInputStream("x".getBytes(StandardCharsets.UTF_8)),
+ new Metadata(), new ParseContext());
+ assertEquals("original", Files.readString(existing));
+ assertEquals(0, tmpFiles(basePath), "tmp file leaked");
+ }
+
+ @Test
+ public void testOnExistsReplaceOverwrites() throws Exception {
+ Path basePath = tempDir.resolve("base");
+ Path existing = seed(basePath, "a.json", "original");
+ StreamEmitter emitter = createEmitter(basePath, null, "REPLACE");
+ emitter.emit("a.json", List.of(new Metadata()), new ParseContext());
+ assertFalse(Files.readString(existing).equals("original"));
+ emitter.emit("a.json", new ByteArrayInputStream("x".getBytes(StandardCharsets.UTF_8)),
+ new Metadata(), new ParseContext());
+ assertEquals("x", Files.readString(existing));
+ assertEquals(0, tmpFiles(basePath), "tmp file leaked");
+ }
+
+ @Test
+ public void testReaderNeverSeesPartialFile() throws Exception {
+ // Regression for the AsyncResourceTest flake: a poller that reads as soon as the
+ // output exists must get the whole file, never an empty one mid-write.
+ Path basePath = tempDir.resolve("base");
+ Files.createDirectories(basePath);
+ StreamEmitter emitter = createEmitter(basePath, null, "REPLACE");
+ Path out = basePath.resolve("big.json");
+ Metadata m = new Metadata();
+ m.set("x", "y".repeat(1 << 20));
+ Thread writer = new Thread(() -> {
+ try {
+ for (int i = 0; i < 20; i++) {
+ emitter.emit("big.json", List.of(m), new ParseContext());
+ Files.delete(out);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ writer.start();
+ long minSeen = Long.MAX_VALUE;
+ while (writer.isAlive()) {
+ try {
+ minSeen = Math.min(minSeen, Files.size(out));
+ } catch (IOException e) {
+ //between delete and next publish
+ }
+ }
+ writer.join();
+ assertTrue(minSeen == Long.MAX_VALUE || minSeen > 1 << 20,
+ "observed partial file of size " + minSeen);
+ }
}