|
16 | 16 | */ |
17 | 17 | package org.apache.tika.plugins; |
18 | 18 |
|
19 | | -import java.io.File; |
20 | 19 | import java.io.IOException; |
21 | | -import java.io.RandomAccessFile; |
22 | | -import java.nio.channels.FileChannel; |
23 | | -import java.nio.channels.FileLock; |
| 20 | +import java.nio.file.AtomicMoveNotSupportedException; |
| 21 | +import java.nio.file.DirectoryNotEmptyException; |
| 22 | +import java.nio.file.FileAlreadyExistsException; |
24 | 23 | import java.nio.file.Files; |
25 | 24 | import java.nio.file.Path; |
26 | | -import java.util.ArrayList; |
27 | | -import java.util.List; |
| 25 | +import java.nio.file.StandardCopyOption; |
| 26 | +import java.util.Comparator; |
| 27 | +import java.util.UUID; |
| 28 | +import java.util.stream.Stream; |
28 | 29 |
|
29 | 30 | import org.pf4j.util.Unzip; |
30 | 31 | import org.slf4j.Logger; |
31 | 32 | import org.slf4j.LoggerFactory; |
32 | 33 |
|
| 34 | +/** |
| 35 | + * Thread-safe and process-safe plugin unzipper using atomic rename. |
| 36 | + * <p> |
| 37 | + * This avoids file locking issues on Windows by using a simple strategy: |
| 38 | + * <ol> |
| 39 | + * <li>Check if destination directory exists with completion marker - if yes, already extracted</li> |
| 40 | + * <li>Extract to a temporary directory with a unique name</li> |
| 41 | + * <li>Create a completion marker file in the temp directory</li> |
| 42 | + * <li>Atomically rename temp dir to final destination</li> |
| 43 | + * <li>If rename fails (another process won), clean up temp dir</li> |
| 44 | + * </ol> |
| 45 | + * <p> |
| 46 | + * The completion marker ensures that even if atomic move is not supported, |
| 47 | + * other processes won't attempt to load a partially-moved directory. |
| 48 | + */ |
33 | 49 | public class ThreadSafeUnzipper { |
34 | 50 | private static final Logger LOG = LoggerFactory.getLogger(TikaPluginManager.class); |
| 51 | + private static final String COMPLETE_MARKER = ".tika-extraction-complete"; |
35 | 52 |
|
36 | | - private static final long MAX_WAIT_MS = 60000; |
37 | | - |
38 | | - public static synchronized void unzipPlugin(Path source) throws IOException { |
39 | | - if (! source.getFileName().toString().endsWith(".zip")) { |
| 53 | + /** |
| 54 | + * Unzips a plugin zip file to a directory with the same name (minus .zip extension). |
| 55 | + * Safe for concurrent calls from multiple threads or processes. See |
| 56 | + * documentation at the head of this class for how it works. |
| 57 | + * |
| 58 | + * @param source path to the .zip file |
| 59 | + * @throws IOException if extraction fails |
| 60 | + */ |
| 61 | + public static void unzipPlugin(Path source) throws IOException { |
| 62 | + if (!source.getFileName().toString().endsWith(".zip")) { |
40 | 63 | throw new IllegalArgumentException("source file name must end in '.zip'"); |
41 | 64 | } |
42 | | - File lockFile = new File(source.toAbsolutePath() + ".lock"); |
43 | | - FileChannel fileChannel = null; |
44 | | - FileLock fileLock = null; |
45 | | - List<IOException> exceptions = new ArrayList<>(); |
| 65 | + |
| 66 | + Path destination = getDestination(source); |
| 67 | + |
| 68 | + // Already extracted - check for both directory AND completion marker |
| 69 | + if (isExtractionComplete(destination)) { |
| 70 | + LOG.debug("{} is already extracted", source); |
| 71 | + return; |
| 72 | + } |
| 73 | + |
| 74 | + // Extract to a unique temp directory |
| 75 | + Path tempDir = destination.resolveSibling( |
| 76 | + destination.getFileName() + ".tmp." + UUID.randomUUID()); |
| 77 | + |
46 | 78 | try { |
47 | | - fileChannel = new RandomAccessFile(lockFile, "rw").getChannel(); |
48 | | - LOG.debug("acquiring lock"); |
49 | | - fileLock = fileChannel.lock(); |
50 | | - LOG.debug("acquired lock"); |
51 | | - if (isExtracted(source)) { |
52 | | - LOG.debug("{} is already extracted", source); |
53 | | - return; |
54 | | - } |
55 | | - extract(source); |
56 | | - } finally { |
57 | | - if (fileLock != null && fileLock.isValid()) { |
58 | | - try { |
59 | | - fileLock.release(); |
60 | | - } catch (IOException e) { |
61 | | - LOG.warn("failed to release the lock"); |
62 | | - exceptions.add(e); |
63 | | - } |
64 | | - } |
65 | | - if (fileChannel != null) { |
| 79 | + LOG.debug("extracting {} to temp dir {}", source, tempDir); |
| 80 | + new Unzip(source.toFile(), tempDir.toFile()).extract(); |
| 81 | + |
| 82 | + // Create completion marker in temp dir before moving |
| 83 | + Files.createFile(tempDir.resolve(COMPLETE_MARKER)); |
| 84 | + |
| 85 | + // Atomically rename to final destination |
| 86 | + try { |
| 87 | + Files.move(tempDir, destination, StandardCopyOption.ATOMIC_MOVE); |
| 88 | + LOG.debug("successfully extracted {}", destination); |
| 89 | + } catch (FileAlreadyExistsException | DirectoryNotEmptyException e) { |
| 90 | + // Another process extracted it first - wait for completion marker |
| 91 | + LOG.debug("plugin already extracted by another process: {}", destination); |
| 92 | + waitForExtractionComplete(destination); |
| 93 | + } catch (AtomicMoveNotSupportedException e) { |
| 94 | + // Filesystem doesn't support atomic move, try regular move |
66 | 95 | try { |
67 | | - fileChannel.close(); |
68 | | - } catch (IOException e) { |
69 | | - LOG.warn("failed to close the file channel"); |
70 | | - exceptions.add(e); |
| 96 | + Files.move(tempDir, destination); |
| 97 | + LOG.debug("successfully extracted {} (non-atomic)", destination); |
| 98 | + } catch (FileAlreadyExistsException | DirectoryNotEmptyException e2) { |
| 99 | + // Another process extracted it first - wait for completion marker |
| 100 | + LOG.debug("plugin already extracted by another process: {}", destination); |
| 101 | + waitForExtractionComplete(destination); |
71 | 102 | } |
72 | 103 | } |
73 | | - boolean isDeleted = lockFile.delete(); |
74 | | - if (! isDeleted) { |
75 | | - LOG.warn("failed to delete the lock file"); |
76 | | - exceptions.add(new IOException("failed to delete lock file: " + lockFile)); |
| 104 | + } finally { |
| 105 | + // Clean up temp dir if it still exists (we lost the race or there was an error) |
| 106 | + if (Files.exists(tempDir)) { |
| 107 | + deleteRecursively(tempDir); |
77 | 108 | } |
78 | 109 | } |
79 | | - if (! exceptions.isEmpty()) { |
80 | | - throw exceptions.get(0); |
81 | | - } |
82 | 110 | } |
83 | 111 |
|
84 | | - private static void extract(Path source) throws IOException { |
85 | | - Path destination = getDestination(source); |
86 | | - Unzip unzip = new Unzip(source.toFile(), destination.toFile()); |
87 | | - unzip.extract(); |
| 112 | + /** |
| 113 | + * Checks if extraction is complete by verifying both directory exists and completion marker is present. |
| 114 | + */ |
| 115 | + private static boolean isExtractionComplete(Path destination) { |
| 116 | + return Files.isDirectory(destination) && Files.exists(destination.resolve(COMPLETE_MARKER)); |
88 | 117 | } |
89 | 118 |
|
90 | | - private static boolean isExtracted(Path source) { |
91 | | - Path destination = getDestination(source); |
92 | | - return Files.isDirectory(destination); |
| 119 | + /** |
| 120 | + * Waits for extraction to complete by polling for the completion marker. |
| 121 | + * This is called when we detect another process is extracting. |
| 122 | + */ |
| 123 | + private static void waitForExtractionComplete(Path destination) throws IOException { |
| 124 | + long maxWaitMs = 60000; // 1 minute max wait |
| 125 | + long pollIntervalMs = 100; |
| 126 | + long waited = 0; |
| 127 | + |
| 128 | + while (waited < maxWaitMs) { |
| 129 | + if (isExtractionComplete(destination)) { |
| 130 | + LOG.debug("extraction completed by another process: {}", destination); |
| 131 | + return; |
| 132 | + } |
| 133 | + try { |
| 134 | + Thread.sleep(pollIntervalMs); |
| 135 | + } catch (InterruptedException e) { |
| 136 | + Thread.currentThread().interrupt(); |
| 137 | + throw new IOException("interrupted while waiting for extraction to complete", e); |
| 138 | + } |
| 139 | + waited += pollIntervalMs; |
| 140 | + } |
| 141 | + |
| 142 | + throw new IOException("timed out waiting for extraction to complete: " + destination); |
93 | 143 | } |
94 | 144 |
|
95 | 145 | private static Path getDestination(Path source) { |
96 | 146 | String fName = source.getFileName().toString(); |
97 | 147 | fName = fName.substring(0, fName.length() - 4); |
98 | 148 | return source.toAbsolutePath().getParent().resolve(fName); |
99 | 149 | } |
| 150 | + |
| 151 | + private static void deleteRecursively(Path path) { |
| 152 | + try (Stream<Path> walk = Files.walk(path)) { |
| 153 | + walk.sorted(Comparator.reverseOrder()) |
| 154 | + .forEach(p -> { |
| 155 | + try { |
| 156 | + Files.delete(p); |
| 157 | + } catch (IOException e) { |
| 158 | + LOG.warn("failed to delete temp file: {}", p, e); |
| 159 | + } |
| 160 | + }); |
| 161 | + } catch (IOException e) { |
| 162 | + LOG.warn("failed to clean up temp directory: {}", path, e); |
| 163 | + } |
| 164 | + } |
100 | 165 | } |
0 commit comments