Handle Windows delete pending files (elastic#12335)

When deleting temporary files created by the DLQ writer to store data before moving to their
final location, Windows may leave these files in a "delete pending" state, where the files
are somewhat in a state of limbo, where they result of `Files.exist(filename)` is `false`,
but the result of `filename.toFile().exists()` is true. When files are in this state, a new
file with the same name cannot be created, which causes the DLQ test used to ensure that
closing and reopening the DLQ (in such events as a pipeline restart) to fail.

This commit moves the temporary file to an alternative location before deletion, ensuring that
the "pending delete" status does not interrupt with the DLQ startup

logstash-core/src/main/java/org/logstash/common/io/DeadLetterQueueWriter.java

@@ -295,8 +295,8 @@ private static Stream<Path> listFiles(Path path, String suffix) throws IOExcepti
     // check if there is a corresponding .log file - if yes delete the temp file, if no atomic move the
     // temp file to be a new segment file..
     private void cleanupTempFile(final Path tempFile) {
-        String tempFilename = tempFile.getFileName().toString().split("\\.")[0];
-        Path segmentFile = queuePath.resolve(String.format("%s.log", tempFilename));
+        String segmentName = tempFile.getFileName().toString().split("\\.")[0];
+        Path segmentFile = queuePath.resolve(String.format("%s.log", segmentName));
         try {
             if (Files.exists(segmentFile)) {
                 Files.delete(tempFile);
@@ -305,16 +305,15 @@ private void cleanupTempFile(final Path tempFile) {
                 SegmentStatus segmentStatus = RecordIOReader.getSegmentStatus(tempFile);
                 switch (segmentStatus){
                     case VALID:
-                        logger.debug("Moving temp file {} to segment file {}", tempFilename, segmentFile);
+                        logger.debug("Moving temp file {} to segment file {}", tempFile, segmentFile);
                         Files.move(tempFile, segmentFile, StandardCopyOption.ATOMIC_MOVE);
                         break;
                     case EMPTY:
-                        logger.debug("Removing unused temp file {}", tempFilename);
-                        Files.delete(tempFile);
+                        deleteTemporaryFile(tempFile, segmentName);
                         break;
                     case INVALID:
-                        Path errorFile = queuePath.resolve(String.format("%s.err", tempFilename));
-                        logger.warn("Segment file {} is in an error state, saving as {}", tempFilename, errorFile);
+                        Path errorFile = queuePath.resolve(String.format("%s.err", segmentName));
+                        logger.warn("Segment file {} is in an error state, saving as {}", segmentFile, errorFile);
                         Files.move(tempFile, errorFile, StandardCopyOption.ATOMIC_MOVE);
                         break;
                     default:
@@ -325,4 +324,25 @@ private void cleanupTempFile(final Path tempFile) {
             throw new IllegalStateException("Unable to clean up temp file: " + tempFile, e);
         }
     }
+
+    // Windows can leave files in a "Delete pending" state, where the file presents as existing to certain
+    // methods, and not to others, and actively prevents a new file being created with the same file name,
+    // throwing AccessDeniedException. This method moves the temporary file to a .del file before
+    // deletion, enabling a new temp file to be created in its place.
+    private void deleteTemporaryFile(Path tempFile, String segmentName) throws IOException {
+        Path deleteTarget;
+        if (isWindows()) {
+            Path deletedFile = queuePath.resolve(String.format("%s.del", segmentName));
+            logger.debug("Moving temp file {} to {}", tempFile, deletedFile);
+            deleteTarget = deletedFile;
+            Files.move(tempFile, deletedFile, StandardCopyOption.ATOMIC_MOVE);
+        } else {
+            deleteTarget = tempFile;
+        }
+        Files.delete(deleteTarget);
+    }
+
+    private static boolean isWindows(){
+        return System.getProperty("os.name").startsWith("Windows");
+    }
 }