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 3c12e418a598481b7a362d6dd513759b8b59bb68 Author: Piotr P. Karwasz <[email protected]> AuthorDate: Thu Jul 30 16:52:21 2026 +0200 Close the file channel resources on every failure path MapDB 0.9.x cannot unmap its buffers on Java 17, so disable memory-mapped files for the transient queue set. Close the queue and backing store discarded by the bad-checkpoint recovery in Log.replay, close the log of a channel whose start() failed, and release the inflight files when the queue constructor throws. Windows keeps open and mapped files locked, so any of these leaks made the channel unable to restart over the same directories. Assisted-By: Claude Fable 5 <[email protected]> --- .../org/apache/flume/channel/file/FileChannel.java | 8 +- .../apache/flume/channel/file/FlumeEventQueue.java | 89 ++++++++++++++-------- .../java/org/apache/flume/channel/file/Log.java | 23 +++++- 3 files changed, 82 insertions(+), 38 deletions(-) diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java index bd9b9c3a..2338400b 100644 --- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java +++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java @@ -317,7 +317,7 @@ public class FileChannel extends BasicChannelSemantics implements TransactionCap public synchronized void stop() { logger.info("Stopping {}...", this); startupError = null; - int size = getDepth(); + int size = open ? getDepth() : 0; close(); if (!open) { channelCounter.setChannelSize(size); @@ -370,8 +370,10 @@ public class FileChannel extends BasicChannelSemantics implements TransactionCap } void close() { - if (open) { - setOpen(false); + setOpen(false); + // Close the log even when the channel never opened: a failed start() leaves + // a partially built log whose open files must still be released. + if (log != null) { try { log.close(); } catch (Exception e) { diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEventQueue.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEventQueue.java index accabf23..0c70a8cd 100644 --- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEventQueue.java +++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FlumeEventQueue.java @@ -84,31 +84,62 @@ final class FlumeEventQueue { logger.error("Could not read checkpoint.", e); throw e; } - if (queueSetDBDir.isDirectory()) { - FileUtils.deleteDirectory(queueSetDBDir); - } else if (queueSetDBDir.isFile() && !queueSetDBDir.delete()) { - throw new IOException("QueueSetDir " + queueSetDBDir + " is a file and" + " could not be deleted"); - } - if (!queueSetDBDir.mkdirs()) { - throw new IllegalStateException("Could not create QueueSet Dir " + queueSetDBDir); - } - File dbFile = new File(queueSetDBDir, "db"); - db = DBMaker.newFileDB(dbFile) - .closeOnJvmShutdown() - .transactionDisable() - .syncOnCommitDisable() - .deleteFilesAfterClose() - .cacheDisable() - .mmapFileEnableIfSupported() - .make(); - queueSet = - db.createHashSet("QueueSet " + " - " + backingStore.getName()).make(); - long start = System.currentTimeMillis(); - for (int i = 0; i < backingStore.getSize(); i++) { - queueSet.add(get(i)); + try { + if (queueSetDBDir.isDirectory()) { + FileUtils.deleteDirectory(queueSetDBDir); + } else if (queueSetDBDir.isFile() && !queueSetDBDir.delete()) { + throw new IOException("QueueSetDir " + queueSetDBDir + " is a file and" + " could not be deleted"); + } + if (!queueSetDBDir.mkdirs()) { + throw new IllegalStateException("Could not create QueueSet Dir " + queueSetDBDir); + } + File dbFile = new File(queueSetDBDir, "db"); + // Don't enable memory-mapped files: MapDB 0.9.x cannot unmap its buffers on + // Java 17, so `deleteFilesAfterClose()` would silently fail on Windows and + // the next construction over the same directory could not delete the files. + db = DBMaker.newFileDB(dbFile) + .closeOnJvmShutdown() + .transactionDisable() + .syncOnCommitDisable() + .deleteFilesAfterClose() + .cacheDisable() + .make(); + queueSet = db.createHashSet("QueueSet " + " - " + backingStore.getName()) + .make(); + long start = System.currentTimeMillis(); + for (int i = 0; i < backingStore.getSize(); i++) { + queueSet.add(get(i)); + } + logger.info("QueueSet population inserting " + backingStore.getSize() + " took " + + (System.currentTimeMillis() - start)); + } catch (Exception e) { + closeQuietly(); + throw e; + } + } + + /** Releases the resources this constructor opened, without closing the backing store. */ + private void closeQuietly() { + try { + if (db != null) { + db.close(); + } + } catch (Exception ex) { + logger.warn("Error closing db", ex); + } finally { + db = null; + queueSet = null; + } + try { + inflightPuts.close(); + } catch (IOException ex) { + logger.warn("Error closing inflight puts", ex); + } + try { + inflightTakes.close(); + } catch (IOException ex) { + logger.warn("Error closing inflight takes", ex); } - logger.info("QueueSet population inserting " + backingStore.getSize() + " took " - + (System.currentTimeMillis() - start)); } SetMultimap<Long, Long> deserializeInflightPuts() throws IOException, BadCheckpointException { @@ -377,20 +408,12 @@ final class FlumeEventQueue { } synchronized void close() throws IOException { - try { - if (db != null) { - db.close(); - } - } catch (Exception ex) { - logger.warn("Error closing db", ex); - } try { backingStore.close(); - inflightPuts.close(); - inflightTakes.close(); } catch (IOException e) { logger.warn("Error closing backing store", e); } + closeQuietly(); } /** diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java index 68d37913..f5f2e6b7 100644 --- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java +++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Log.java @@ -485,6 +485,23 @@ public class Log { */ doReplay(queue, dataFiles, encryptionKeyProvider, shouldFastReplay); } catch (BadCheckpointException ex) { + // Close the discarded queue and backing store first: their open handles and + // buffers would prevent the deletion of the checkpoint files on Windows. + if (queue != null) { + try { + queue.close(); + } catch (Exception closeEx) { + logger.warn("Error closing the queue built from a bad checkpoint", closeEx); + } + queue = null; + } else if (backingStore != null) { + try { + backingStore.close(); + } catch (Exception closeEx) { + logger.warn("Error closing the backing store of a bad checkpoint", closeEx); + } + } + backingStore = null; backupRestored = false; if (useDualCheckpoints) { logger.warn( @@ -839,7 +856,7 @@ public class Log { try { open = false; try { - if (checkpointOnClose) { + if (checkpointOnClose && queue != null) { writeCheckpoint(true); // do this before acquiring exclusive lock } } catch (Exception err) { @@ -865,7 +882,9 @@ public class Log { } } } - queue.close(); + if (queue != null) { + queue.close(); + } try { unlock(checkpointDir); } catch (IOException ex) {
