Fix the file channel and the test suite on Windows

.github/workflows/build.yml

@@ -43,7 +43,7 @@ jobs:
       # Don't cancel the remaining OS builds when one fails
       fail-fast: false
       matrix:
-        os: [ ubuntu-latest ]
+        os: [ ubuntu-latest, windows-latest ]
         java-distribution: [ temurin ]
         #
         # There is no protobuf 2.x version for `aarch64`.

flume-ng-channels/flume-file-channel/pom.xml

@@ -30,8 +30,8 @@
 
   <properties>
     <!-- TODO fix spotbugs violations -->
-    <spotbugs.maxAllowedViolations>284</spotbugs.maxAllowedViolations>
-    <pmd.maxAllowedViolations>544</pmd.maxAllowedViolations>
+    <spotbugs.maxAllowedViolations>162</spotbugs.maxAllowedViolations>
+    <pmd.maxAllowedViolations>19</pmd.maxAllowedViolations>
     <module.name>org.apache.flume.channel.file</module.name>
   </properties>
 

flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFactory.java

@@ -20,6 +20,8 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -116,10 +118,26 @@ private static EventQueueBackingStore upgrade(
         logger.info("Attempting upgrade of " + checkpointFile + " for " + name);
         EventQueueBackingStoreFileV2 backingStoreV2 =
                 new EventQueueBackingStoreFileV2(checkpointFile, capacity, name, counter);
+        int queueHead;
+        int queueSize;
+        long writeOrderID;
+        Map<Integer, AtomicInteger> referenceCounts;
+        try {
+            queueHead = backingStoreV2.getHead();
+            queueSize = backingStoreV2.getSize();
+            writeOrderID = backingStoreV2.getLogWriteOrderID();
+            referenceCounts = backingStoreV2.logFileIDReferenceCounts;
+        } finally {
+            // Close the V2 store before the file is copied and rewritten.
+            // The stale mapping lingers until garbage collected, which is harmless:
+            // even on Windows a mapped file can be read, written and mapped again.
+            backingStoreV2.close();
+        }
         String backupName = checkpointFile.getName() + "-backup-" + System.currentTimeMillis();
         Files.copy(checkpointFile, new File(checkpointFile.getParentFile(), backupName));
         File metaDataFile = Serialization.getMetaDataFile(checkpointFile);
-        EventQueueBackingStoreFileV3.upgrade(backingStoreV2, checkpointFile, metaDataFile);
+        EventQueueBackingStoreFileV3.upgrade(
+                checkpointFile, metaDataFile, queueHead, queueSize, writeOrderID, referenceCounts);
         return new EventQueueBackingStoreFileV3(
                 checkpointFile, capacity, name, counter, backupCheckpointDir, shouldBackup, compressBackup);
     }

flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFile.java

@@ -56,7 +56,7 @@ abstract class EventQueueBackingStoreFile extends EventQueueBackingStore {
     protected LongBuffer elementsBuffer;
     protected final Map<Integer, Long> overwriteMap = new HashMap<Integer, Long>();
     protected final Map<Integer, AtomicInteger> logFileIDReferenceCounts = Maps.newHashMap();
-    protected final MappedByteBuffer mappedBuffer;
+    private MappedByteBuffer mappedBuffer;
     protected final RandomAccessFile checkpointFileHandle;
     private final FileChannelCounter fileChannelCounter;
     protected final File checkpointFile;
@@ -87,36 +87,53 @@ protected EventQueueBackingStoreFile(
         this.shouldBackup = backupCheckpoint;
         this.compressBackup = compressBackup;
         this.backupDir = checkpointBackupDir;
-        checkpointFileHandle = new RandomAccessFile(checkpointFile, "rw");
-        long totalBytes = (capacity + HEADER_SIZE) * Serialization.SIZE_OF_LONG;
-        if (checkpointFileHandle.length() == 0) {
-            allocate(checkpointFile, totalBytes);
-            checkpointFileHandle.seek(INDEX_VERSION * Serialization.SIZE_OF_LONG);
-            checkpointFileHandle.writeLong(getVersion());
-            checkpointFileHandle.getChannel().force(true);
-            logger.info("Preallocated " + checkpointFile + " to " + checkpointFileHandle.length() + " for capacity "
-                    + capacity);
-        }
-        if (checkpointFile.length() != totalBytes) {
-            String msg = "Configured capacity is " + capacity + " but the "
-                    + " checkpoint file capacity is "
-                    + ((checkpointFile.length() / Serialization.SIZE_OF_LONG) - HEADER_SIZE)
-                    + ". See FileChannel documentation on how to change a channels" + " capacity.";
-            throw new BadCheckpointException(msg);
-        }
-        mappedBuffer = checkpointFileHandle.getChannel().map(MapMode.READ_WRITE, 0, checkpointFile.length());
-        elementsBuffer = mappedBuffer.asLongBuffer();
+        // On failure, close the file handle and drop all references to the mapping before rethrowing:
+        // the mapping is only released when the buffer is garbage collected,
+        // and a reachable buffer keeps the checkpoint file undeletable on Windows.
+        RandomAccessFile checkpointFileHandle = new RandomAccessFile(checkpointFile, "rw");
+        MappedByteBuffer mappedBuffer;
+        try {
+            long totalBytes = (capacity + HEADER_SIZE) * Serialization.SIZE_OF_LONG;
+            if (checkpointFileHandle.length() == 0) {
+                allocate(checkpointFile, totalBytes);
+                checkpointFileHandle.seek(INDEX_VERSION * Serialization.SIZE_OF_LONG);
+                checkpointFileHandle.writeLong(getVersion());
+                checkpointFileHandle.getChannel().force(true);
+                logger.info("Preallocated " + checkpointFile + " to " + checkpointFileHandle.length() + " for capacity "
+                        + capacity);
+            }
+            if (checkpointFile.length() != totalBytes) {
+                String msg = "Configured capacity is " + capacity + " but the "
+                        + " checkpoint file capacity is "
+                        + ((checkpointFile.length() / Serialization.SIZE_OF_LONG) - HEADER_SIZE)
+                        + ". See FileChannel documentation on how to change a channels" + " capacity.";
+                throw new BadCheckpointException(msg);
+            }
+            mappedBuffer = checkpointFileHandle.getChannel().map(MapMode.READ_WRITE, 0, checkpointFile.length());
+            elementsBuffer = mappedBuffer.asLongBuffer();
 
-        long version = elementsBuffer.get(INDEX_VERSION);
-        if (version != (long) getVersion()) {
-            throw new BadCheckpointException("Invalid version: " + version + " " + name + ", expected " + getVersion());
-        }
-        long checkpointComplete = elementsBuffer.get(INDEX_CHECKPOINT_MARKER);
-        if (checkpointComplete != (long) CHECKPOINT_COMPLETE) {
-            throw new BadCheckpointException("Checkpoint was not completed correctly,"
-                    + " probably because the agent stopped while the channel was"
-                    + " checkpointing.");
+            long version = elementsBuffer.get(INDEX_VERSION);
+            if (version != (long) getVersion()) {
+                throw new BadCheckpointException(
+                        "Invalid version: " + version + " " + name + ", expected " + getVersion());
+            }
+            long checkpointComplete = elementsBuffer.get(INDEX_CHECKPOINT_MARKER);
+            if (checkpointComplete != (long) CHECKPOINT_COMPLETE) {
+                throw new BadCheckpointException("Checkpoint was not completed correctly,"
+                        + " probably because the agent stopped while the channel was"
+                        + " checkpointing.");
+            }
+        } catch (IOException | RuntimeException e) {
+            elementsBuffer = null;
+            try {
+                checkpointFileHandle.close();
+            } catch (IOException closeEx) {
+                e.addSuppressed(closeEx);
+            }
+            throw e;
         }
+        this.checkpointFileHandle = checkpointFileHandle;
+        this.mappedBuffer = mappedBuffer;
         if (shouldBackup) {
             checkpointBackUpExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
                     .setNameFormat(getName() + " - CheckpointBackUpThread")
@@ -127,6 +144,7 @@ protected EventQueueBackingStoreFile(
     }
 
     protected long getCheckpointLogWriteOrderID() {
+        checkNotClosed();
         return elementsBuffer.get(INDEX_WRITE_ORDER_ID);
     }
 
@@ -223,8 +241,22 @@ public static boolean restoreBackup(File checkpointDir, File backupDir) throws I
         }
     }
 
+    /**
+     * Throws {@link IllegalStateException} if this store is closed.
+     *
+     * <p>A null {@code mappedBuffer} marks the store as closed.
+     * Methods that dereference {@code elementsBuffer} or {@code mappedBuffer} should call this method first,
+     * to fail fast instead of throwing a NullPointerException.
+     */
+    private void checkNotClosed() {
+        if (mappedBuffer == null) {
+            throw new IllegalStateException("Backing store " + checkpointFile + " is closed");
+        }
+    }
+
     @Override
     void beginCheckpoint() throws IOException {
+        checkNotClosed();
         logger.info("Start checkpoint for " + checkpointFile + ", elements to sync = " + overwriteMap.size());
 
         if (shouldBackup) {
@@ -248,7 +280,7 @@ void beginCheckpoint() throws IOException {
 
     @Override
     void checkpoint() throws IOException {
-
+        checkNotClosed();
         setLogWriteOrderID(WriteOrderOracle.next());
         logger.info("Updating checkpoint metadata: logWriteOrderID: "
                 + getLogWriteOrderID() + ", queueSize: " + getSize() + ", queueHead: "
@@ -310,7 +342,14 @@ public void run() {
 
     @Override
     void close() {
+        if (mappedBuffer == null) {
+            return;
+        }
         mappedBuffer.force();
+        // Drop the buffer references, so the mapping can be garbage collected while the store is still reachable:
+        // the file cannot be deleted on Windows until the mapping is gone.
+        mappedBuffer = null;
+        elementsBuffer = null;
         try {
             checkpointFileHandle.close();
         } catch (IOException e) {
@@ -329,6 +368,7 @@ void close() {
 
     @Override
     long get(int index) {
+        checkNotClosed();
         int realIndex = getPhysicalIndex(index);
         long result = EMPTY;
         if (overwriteMap.containsKey(realIndex)) {
@@ -346,6 +386,7 @@ ImmutableSortedSet<Integer> getReferenceCounts() {
 
     @Override
     void put(int index, long value) {
+        checkNotClosed();
         int realIndex = getPhysicalIndex(index);
         overwriteMap.put(realIndex, value);
     }

flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFileV2.java

@@ -34,19 +34,24 @@ final class EventQueueBackingStoreFileV2 extends EventQueueBackingStoreFile {
     EventQueueBackingStoreFileV2(File checkpointFile, int capacity, String name, FileChannelCounter counter)
             throws IOException, BadCheckpointException {
         super(capacity, name, counter, checkpointFile);
-        Preconditions.checkArgument(capacity > 0, "capacity must be greater than 0 " + capacity);
+        try {
+            Preconditions.checkArgument(capacity > 0, "capacity must be greater than 0 " + capacity);
 
-        setLogWriteOrderID(elementsBuffer.get(INDEX_WRITE_ORDER_ID));
-        setSize((int) elementsBuffer.get(INDEX_SIZE));
-        setHead((int) elementsBuffer.get(INDEX_HEAD));
+            setLogWriteOrderID(elementsBuffer.get(INDEX_WRITE_ORDER_ID));
+            setSize((int) elementsBuffer.get(INDEX_SIZE));
+            setHead((int) elementsBuffer.get(INDEX_HEAD));
 
-        int indexMaxLog = INDEX_ACTIVE_LOG + MAX_ACTIVE_LOGS;
-        for (int i = INDEX_ACTIVE_LOG; i < indexMaxLog; i++) {
-            long nextFileCode = elementsBuffer.get(i);
-            if (nextFileCode != EMPTY) {
-                Pair<Integer, Integer> idAndCount = deocodeActiveLogCounter(nextFileCode);
-                logFileIDReferenceCounts.put(idAndCount.getLeft(), new AtomicInteger(idAndCount.getRight()));
+            int indexMaxLog = INDEX_ACTIVE_LOG + MAX_ACTIVE_LOGS;
+            for (int i = INDEX_ACTIVE_LOG; i < indexMaxLog; i++) {
+                long nextFileCode = elementsBuffer.get(i);
+                if (nextFileCode != EMPTY) {
+                    Pair<Integer, Integer> idAndCount = deocodeActiveLogCounter(nextFileCode);
+                    logFileIDReferenceCounts.put(idAndCount.getLeft(), new AtomicInteger(idAndCount.getRight()));
+                }
             }
+        } catch (RuntimeException e) {
+            close();
+            throw e;
         }
     }