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 <[email protected]> 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); + } }
