This is an automated email from the ASF dual-hosted git repository. ppkarwasz pushed a commit to branch feat/fix-file-channel-windows in repository https://gitbox.apache.org/repos/asf/logging-flume.git
commit 778140f4cba7c749552058b8db28c149837f61e8 Author: Piotr P. Karwasz <[email protected]> AuthorDate: Thu Jul 30 16:52:07 2026 +0200 Unmap the checkpoint buffer when the backing store closes On Java 17 nothing unmaps the checkpoint MappedByteBuffer, and on Windows a mapped file cannot be deleted, so checkpoint recovery and restarts fail. Unmap explicitly via sun.misc.Unsafe.invokeCleaner, make close() idempotent, guard the accessors against use after unmap (which would crash the JVM) and release the mapping when a constructor throws BadCheckpointException or when upgrading a V2 checkpoint. Assisted-By: Claude Fable 5 <[email protected]> --- .../file/EventQueueBackingStoreFactory.java | 13 ++- .../channel/file/EventQueueBackingStoreFile.java | 96 +++++++++++++++------- .../channel/file/EventQueueBackingStoreFileV2.java | 25 +++--- .../channel/file/EventQueueBackingStoreFileV3.java | 14 +++- .../org/apache/flume/channel/file/MappedFiles.java | 74 +++++++++++++++++ 5 files changed, 178 insertions(+), 44 deletions(-) diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFactory.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFactory.java index db5bc899..525f02c6 100644 --- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFactory.java +++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFactory.java @@ -116,10 +116,15 @@ class EventQueueBackingStoreFactory { logger.info("Attempting upgrade of " + checkpointFile + " for " + name); EventQueueBackingStoreFileV2 backingStoreV2 = new EventQueueBackingStoreFileV2(checkpointFile, capacity, name, counter); - 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); + try { + 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); + } finally { + // Release the V2 mapping before the V3 store maps the same file. + backingStoreV2.close(); + } return new EventQueueBackingStoreFileV3( checkpointFile, capacity, name, counter, backupCheckpointDir, shouldBackup, compressBackup); } diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFile.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFile.java index 11768082..a1162efe 100644 --- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFile.java +++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFile.java @@ -65,6 +65,7 @@ abstract class EventQueueBackingStoreFile extends EventQueueBackingStore { protected final boolean compressBackup; private final File backupDir; private final ExecutorService checkpointBackUpExecutor; + private volatile boolean closed; protected EventQueueBackingStoreFile( int capacity, String name, FileChannelCounter fileChannelCounter, File checkpointFile) @@ -87,36 +88,53 @@ abstract class EventQueueBackingStoreFile extends EventQueueBackingStore { 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 release the file handle and the mapping before rethrowing: + // a leaked mapping keeps the checkpoint file locked on Windows. + RandomAccessFile fileHandle = new RandomAccessFile(checkpointFile, "rw"); + MappedByteBuffer buffer = null; + try { + long totalBytes = (capacity + HEADER_SIZE) * Serialization.SIZE_OF_LONG; + if (fileHandle.length() == 0) { + allocate(checkpointFile, totalBytes); + fileHandle.seek(INDEX_VERSION * Serialization.SIZE_OF_LONG); + fileHandle.writeLong(getVersion()); + fileHandle.getChannel().force(true); + logger.info( + "Preallocated " + checkpointFile + " to " + fileHandle.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); + } + buffer = fileHandle.getChannel().map(MapMode.READ_WRITE, 0, checkpointFile.length()); + elementsBuffer = buffer.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; + MappedFiles.unmap(buffer); + try { + fileHandle.close(); + } catch (IOException closeEx) { + e.addSuppressed(closeEx); + } + throw e; } + checkpointFileHandle = fileHandle; + mappedBuffer = buffer; if (shouldBackup) { checkpointBackUpExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder() .setNameFormat(getName() + " - CheckpointBackUpThread") @@ -223,8 +241,19 @@ abstract class EventQueueBackingStoreFile extends EventQueueBackingStore { } } + /** + * Accessing the buffer after {@link #close()} unmapped it would crash the JVM, + * so every buffer access checks this first. + */ + private void checkNotClosed() { + if (closed) { + 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 +277,7 @@ abstract class EventQueueBackingStoreFile extends EventQueueBackingStore { @Override void checkpoint() throws IOException { - + checkNotClosed(); setLogWriteOrderID(WriteOrderOracle.next()); logger.info("Updating checkpoint metadata: logWriteOrderID: " + getLogWriteOrderID() + ", queueSize: " + getSize() + ", queueHead: " @@ -310,7 +339,14 @@ abstract class EventQueueBackingStoreFile extends EventQueueBackingStore { @Override void close() { + if (closed) { + return; + } + closed = true; mappedBuffer.force(); + // The buffer must not be touched after this point, see the guards in the accessors. + MappedFiles.unmap(mappedBuffer); + elementsBuffer = null; try { checkpointFileHandle.close(); } catch (IOException e) { @@ -329,6 +365,7 @@ abstract class EventQueueBackingStoreFile extends EventQueueBackingStore { @Override long get(int index) { + checkNotClosed(); int realIndex = getPhysicalIndex(index); long result = EMPTY; if (overwriteMap.containsKey(realIndex)) { @@ -346,6 +383,7 @@ abstract class EventQueueBackingStoreFile extends EventQueueBackingStore { @Override void put(int index, long value) { + checkNotClosed(); int realIndex = getPhysicalIndex(index); overwriteMap.put(realIndex, value); } diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFileV2.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFileV2.java index 7dc53e74..cd06d24a 100644 --- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFileV2.java +++ b/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; } } diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFileV3.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFileV3.java index 61ccbcd4..a05d3896 100644 --- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFileV3.java +++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFileV3.java @@ -32,7 +32,7 @@ import org.apache.logging.log4j.Logger; final class EventQueueBackingStoreFileV3 extends EventQueueBackingStoreFile { private static final Logger logger = LogManager.getLogger(); - private final File metaDataFile; + private File metaDataFile; EventQueueBackingStoreFileV3(File checkpointFile, int capacity, String name, FileChannelCounter counter) throws IOException, BadCheckpointException { @@ -49,6 +49,18 @@ final class EventQueueBackingStoreFileV3 extends EventQueueBackingStoreFile { boolean compressBackup) throws IOException, BadCheckpointException { super(capacity, name, counter, checkpointFile, checkpointBackupDir, backupCheckpoint, compressBackup); + try { + constructorBody(checkpointFile, capacity, name, checkpointBackupDir); + } catch (IOException | RuntimeException e) { + // Release the checkpoint mapping opened by the superclass, otherwise the + // recovery path cannot delete the checkpoint file on Windows. + close(); + throw e; + } + } + + private void constructorBody(File checkpointFile, int capacity, String name, File checkpointBackupDir) + throws IOException, BadCheckpointException { Preconditions.checkArgument(capacity > 0, "capacity must be greater than 0 " + capacity); metaDataFile = Serialization.getMetaDataFile(checkpointFile); logger.info("Starting up with " + checkpointFile + " and " + metaDataFile); diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/MappedFiles.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/MappedFiles.java new file mode 100644 index 00000000..a508537b --- /dev/null +++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/MappedFiles.java @@ -0,0 +1,74 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.flume.channel.file; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.nio.ByteBuffer; +import java.nio.MappedByteBuffer; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +/** + * Releases {@link MappedByteBuffer}s eagerly instead of waiting for garbage collection. + * On Windows a memory-mapped file cannot be deleted while a mapping is alive, so the + * channel must unmap its buffers before checkpoint files can be removed or replaced. + */ +final class MappedFiles { + private static final Logger logger = LogManager.getLogger(); + private static final Object UNSAFE; + private static final Method INVOKE_CLEANER; + + static { + // `jdk.unsupported` exports and opens `sun.misc`, so no JVM flags are needed. + Object unsafe = null; + Method invokeCleaner = null; + try { + Class<?> unsafeClass = Class.forName("sun.misc.Unsafe"); + Field theUnsafe = unsafeClass.getDeclaredField("theUnsafe"); + theUnsafe.setAccessible(true); + unsafe = theUnsafe.get(null); + invokeCleaner = unsafeClass.getMethod("invokeCleaner", ByteBuffer.class); + } catch (ReflectiveOperationException | RuntimeException e) { + logger.warn( + "Cannot unmap memory-mapped files explicitly. Deleting checkpoint" + + " files may fail until the mappings are garbage collected.", + e); + } + UNSAFE = unsafe; + INVOKE_CLEANER = invokeCleaner; + } + + private MappedFiles() {} + + /** + * Unmaps the given buffer. The buffer must never be accessed afterwards: + * reading or writing an unmapped buffer crashes the JVM. + * + * @param buffer the buffer returned by {@code FileChannel.map}, may be {@code null} + */ + static void unmap(MappedByteBuffer buffer) { + if (buffer == null || INVOKE_CLEANER == null) { + return; + } + try { + INVOKE_CLEANER.invoke(UNSAFE, buffer); + } catch (ReflectiveOperationException | RuntimeException e) { + logger.warn("Failed to unmap buffer", e); + } + } +}
