This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new 6a2cad3f49 TIKA-4848: FileSystemEmitter writes to a tmp file and
renames atomica… (#3085)
6a2cad3f49 is described below
commit 6a2cad3f4958add3db4b0f52136e8c6400f51bbf
Author: Tim Allison <[email protected]>
AuthorDate: Thu Aug 27 17:19:12 2026 -0400
TIKA-4848: FileSystemEmitter writes to a tmp file and renames atomica…
(#3085)
---
CHANGES.txt | 5 +
.../ROOT/pages/pipes/plugins/filesystem.adoc | 4 +
.../tika/pipes/emitter/fs/FileSystemEmitter.java | 96 ++++++++++++---
.../pipes/emitter/fs/FileSystemEmitterConfig.java | 8 +-
.../pipes/emitter/fs/FileSystemEmitterTest.java | 134 ++++++++++++++++++++-
5 files changed, 228 insertions(+), 19 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 42be53d9c8..19ee2c9cc9 100644
--- a/CHANGES.txt
+++ b/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
diff --git a/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
b/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
index 266186421f..f25773c311 100644
--- a/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
+++ b/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]
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..5e12739827 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,9 +131,28 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
}
}
+ 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 class FileSystemEmitter extends
AbstractStreamEmitter {
}
}
+ 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 class FileSystemEmitter extends
AbstractStreamEmitter {
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 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
// 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);
}
}
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java
index 7fc17e2d84..5365c0e62c 100644
---
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java
@@ -19,17 +19,21 @@ package org.apache.tika.pipes.emitter.fs;
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)
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..21500b40ce 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,16 @@ 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());
@@ -51,9 +67,12 @@ public class FileSystemEmitterTest {
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 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);
+ }
+
+ @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));
+ }
}
