TIKA-4848: FileSystemEmitter writes to a tmp file and renames atomica…

CHANGES.txt

@@ -1,5 +1,10 @@
 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. Set "atomicWrites": false on the emitter to restore in-place
+     writes (TIKA-4848).
+
    * tika-eval: Profile/Compare accept the batch run's jsonl crash ledger
      (--pipesReport, -pa/-pb) and a run-info json (--runInfo, -ra/-rb), and
      read both from <extracts>/.run-info/ by default (refusing an ambiguous

docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc

@@ -143,6 +143,10 @@ Writes parsed results as files under `basePath`. The relative output path is der
 |`prettyPrint`
 |`false`
 |Pretty-print JSON output. Has no effect in `CONTENT_ONLY` mode (raw bytes are written).
+
+|`atomicWrites`
+|`true`
+|Write each output to a sibling `<name>.<uuid>.tmp` and rename it into place, so a reader of the output directory never sees a partial file. Set to `false` to write in place (one fewer rename per file; needed on filesystems where rename is slow or not atomic). Init-time only.
 |===
 
 [#file-system-iterator]

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.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 @@
  */
 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,9 +131,28 @@ public void emit(String emitKey, List<Metadata> metadataList, ParseContext parse
             }
         }
 
+        if (!config.atomicWrites()) {
+            writeInPlace(metadataList, output, config);
+            return;
+        }
+        Path tmp = tmpFor(output);
+        try {
+            try (Writer writer = Files.newBufferedWriter(tmp, StandardCharsets.UTF_8,
+                    StandardOpenOption.CREATE_NEW)) {
+                JsonMetadataList.toJson(metadataList, writer, config.prettyPrint());
+            }
+            publish(tmp, output, config.onExists());
+        } finally {
+            Files.deleteIfExists(tmp);
+        }
+    }
+
+    // atomicWrites=false: the pre-TIKA-4848 behavior; readers can observe a partial file
+    private static void writeInPlace(List<Metadata> metadataList, Path output,
+                                     FileSystemEmitterConfig config) throws IOException {
         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
+                    StandardOpenOption.CREATE_NEW)) {
                 JsonMetadataList.toJson(metadataList, writer, config.prettyPrint());
             } catch (FileAlreadyExistsException e) {
                 throw alreadyExistsException(output);
@@ -141,6 +164,35 @@ public void emit(String emitKey, List<Metadata> metadataList, ParseContext parse
         }
     }
 
+    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);
+        }
+    }
+
     @Override
     public void emit(String emitKey, InputStream inputStream, Metadata userMetadata, ParseContext parseContext) throws IOException {
 
@@ -174,22 +226,35 @@ public void emit(String emitKey, InputStream inputStream, Metadata userMetadata,
         if (!Files.isDirectory(output.getParent())) {
             Files.createDirectories(output.getParent());
         }
-        if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.REPLACE) {
+        if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.SKIP && Files.exists(output)) {
+            LOG.debug("Skipping existing file: {}", output);
+            return;
+        }
+        if (!config.atomicWrites()) {
+            copyInPlace(inputStream, output, config.onExists());
+            return;
+        }
+        Path tmp = tmpFor(output);
+        try {
+            Files.copy(inputStream, tmp);
+            publish(tmp, output, config.onExists());
+        } finally {
+            Files.deleteIfExists(tmp);
+        }
+    }
+
+    private static void copyInPlace(InputStream inputStream, Path output,
+                                    FileSystemEmitterConfig.ON_EXISTS onExists) throws IOException {
+        if (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) {
+            return;
+        }
+        try {
+            Files.copy(inputStream, output);
+        } catch (FileAlreadyExistsException e) {
+            if (onExists == FileSystemEmitterConfig.ON_EXISTS.EXCEPTION) {
                 throw alreadyExistsException(output);
             }
-        } else if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.SKIP) {
-            if (!Files.isRegularFile(output)) {
-                try {
-                    Files.copy(inputStream, output);
-                } catch (FileAlreadyExistsException e) {
-                    //swallow
-                }
-            }
         }
     }
 
@@ -220,7 +285,8 @@ private FileSystemEmitterConfig getConfig(ParseContext parseContext) throws Tika
                 // Merge runtime config into default config while preserving basePath and the
                 // init-time allowAbsolutePaths -- neither may be changed at runtime.
                 config = new FileSystemEmitterConfig(fileSystemEmitterConfig.basePath(), runtimeConfig.getFileExtension(), runtimeConfig.getOnExists(),
-                        runtimeConfig.isPrettyPrint(), fileSystemEmitterConfig.allowAbsolutePaths());
+                        runtimeConfig.isPrettyPrint(), fileSystemEmitterConfig.allowAbsolutePaths(),
+                        fileSystemEmitterConfig.atomicWrites());
                 checkConfig(config);
             }
         }

tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java

@@ -19,17 +19,21 @@
 import org.apache.tika.exception.TikaConfigException;
 import org.apache.tika.plugins.PluginJson;
 
-public record FileSystemEmitterConfig(String basePath, String fileExtension, ON_EXISTS onExists, boolean prettyPrint, boolean allowAbsolutePaths) {
+public record FileSystemEmitterConfig(String basePath, String fileExtension, ON_EXISTS onExists, boolean prettyPrint, boolean allowAbsolutePaths,
+        Boolean atomicWrites) {
 
     enum ON_EXISTS {
         SKIP, EXCEPTION, REPLACE
     }
 
-    /** onExists is optional; absent means EXCEPTION, the documented default. */
+    /** onExists absent means EXCEPTION, atomicWrites absent means true -- the documented defaults. */
     public FileSystemEmitterConfig {
         if (onExists == null) {
             onExists = ON_EXISTS.EXCEPTION;
         }
+        if (atomicWrites == null) {
+            atomicWrites = Boolean.TRUE;
+        }
     }
 
     public static FileSystemEmitterConfig load(final String json)

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.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,16 +50,29 @@ 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 {
+        return createEmitter(basePath, allowAbsolutePaths, onExists, null);
+    }
+
+    private StreamEmitter createEmitter(Path basePath, Boolean allowAbsolutePaths, String onExists,
+                                        Boolean atomicWrites) throws TikaConfigException, IOException {
         ObjectNode config = MAPPER.createObjectNode();
         if (basePath != null) {
             config.put("basePath", basePath.toAbsolutePath().toString());
         }
         if (allowAbsolutePaths != null) {
             config.put("allowAbsolutePaths", allowAbsolutePaths);
         }
-        config.put("onExists", "REPLACE");
+        config.put("onExists", onExists);
+        if (atomicWrites != null) {
+            config.put("atomicWrites", atomicWrites);
+        }
         ExtensionConfig pluginConfig = new ExtensionConfig("test", "test", config.toString());
-        return new FileSystemEmitterFactory().buildExtension(pluginConfig);
+        return (StreamEmitter) new FileSystemEmitterFactory().buildExtension(pluginConfig);
     }
 
     @Test
@@ -82,4 +101,115 @@ public void testPathTraversalBlocked() throws Exception {
         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);
+    }
+
+    @Test
+    public void testAtomicWritesOff() throws Exception {
+        Path basePath = tempDir.resolve("base");
+        Path existing = seed(basePath, "a.json", "original");
+        StreamEmitter exc = createEmitter(basePath, null, "EXCEPTION", false);
+        assertThrows(IOException.class, () ->
+                exc.emit("a.json", List.of(new Metadata()), new ParseContext()));
+        assertThrows(IOException.class, () -> exc.emit("a.json",
+                new ByteArrayInputStream("x".getBytes(StandardCharsets.UTF_8)), new Metadata(),
+                new ParseContext()));
+        StreamEmitter skip = createEmitter(basePath, null, "SKIP", false);
+        skip.emit("a.json", List.of(new Metadata()), new ParseContext());
+        skip.emit("a.json", new ByteArrayInputStream("x".getBytes(StandardCharsets.UTF_8)),
+                new Metadata(), new ParseContext());
+        assertEquals("original", Files.readString(existing));
+        StreamEmitter replace = createEmitter(basePath, null, "REPLACE", false);
+        replace.emit("a.json", new ByteArrayInputStream("x".getBytes(StandardCharsets.UTF_8)),
+                new Metadata(), new ParseContext());
+        assertEquals("x", Files.readString(existing));
+        replace.emit("b.json", List.of(new Metadata()), new ParseContext());
+        assertTrue(Files.isRegularFile(basePath.resolve("b.json")));
+        assertEquals(0, tmpFiles(basePath));
+    }
 }