This is an automated email from the ASF dual-hosted git repository.
ppkarwasz pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/logging-flume.git
The following commit(s) were added to refs/heads/trunk by this push:
new ae6c7db2 Fix the file channel and the test suite on Windows (#478)
ae6c7db2 is described below
commit ae6c7db22f7cab4d4dc496ae4cdb1d13130a8343
Author: Piotr P. Karwasz <[email protected]>
AuthorDate: Fri Jul 31 21:58:37 2026 +0200
Fix the file channel and the test suite on Windows (#478)
* Skip Taildir tests on non-POSIX file systems
TaildirSource tracks files by inode, so its tests cannot pass when the
default file system lacks a `unix` attribute view (e.g. on Windows).
Document that requirement in the user guide.
Assisted-By: Claude Fable 5 <[email protected]>
* Fix TestSyslogTcpSource.testSSLMessages on Windows
The test wrote to the TLS socket and immediately closed it. The client
never reads the TLS 1.3 session tickets sent by the server, so close()
aborts the connection with a TCP RST, which on Windows discards the
syslog record before the server reads it. Keep the socket open, waiting
with Awaitility, until the server commits the event to the channel.
Assisted-By: Claude Fable 5 <[email protected]>
* Add Windows to the CI matrix
Assisted-By: Claude Fable 5 <[email protected]>
* 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]>
* 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]>
* Close file channel queues and stores in tests
Several tests leaked FlumeEventQueue instances or reused one temporary
directory across test methods, which fails on Windows where the leaked
MapDB files cannot be deleted. Also stop channels whose start() failed,
now that FileChannel releases the log in that case.
Assisted-By: Claude Fable 5 <[email protected]>
* Make the remaining file channel tests Windows-compatible
TestFileChannelRestart never closed the stream it corrupted the
checkpoint metadata with, so the recovery could not delete the file on
Windows. TestFileChannelEncryption asserted the POSIX rendering of a
FileNotFoundException message. TestFileChannelErrorMetrics injects I/O
errors by deleting files the running channel uses, which Windows does
not allow at all, so those tests are skipped there.
Assisted-By: Claude Fable 5 <[email protected]>
* Improve `checkNotClosed` comment
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
* Clean up the bad-checkpoint failure paths
Attach close failures to the original exception instead of logging
them separately, so they show up with their cause in the recovery
warnings. Fold the V3 constructor helper back into the constructor,
which lets metaDataFile become final.
Assisted-By: Claude Fable 5 <[email protected]>
* Release the MapDB before re-creating a queue in tests
The FlumeEventQueue constructor deletes the queueset directory, which
fails on Windows while the previous queue instance still holds the
MapDB open. Production code always calls replayComplete() before a
queue is abandoned, so do the same in the tests.
Assisted-By: Claude Fable 5 <[email protected]>
* Let the GC release the checkpoint mapping
Forcibly unmapping the checkpoint MappedByteBuffer via
sun.misc.Unsafe.invokeCleaner is unsafe: any access to an unmapped
buffer crashes the JVM. Drop the buffer references when the backing
store closes and let the garbage collector release the mapping instead.
The corrupt-checkpoint recovery path deletes checkpoint files moments
after the discarded store is closed, so retry the deletion with GC
cycles in between (a no-op when the first attempt succeeds). Tests that
delete a recently mapped checkpoint wait the same way via
TestUtils.awaitDelete. The V2 to V3 upgrade now reads the queue state,
closes the V2 store and backs up the file before rewriting it in place,
which needs no unmapping: writing to and re-mapping a mapped file is
allowed even on Windows, only deletion is blocked.
Also ratchet the spotbugs and pmd ceilings to the current counts.
Assisted-By: Claude Fable 5 <[email protected]>
* Shorten comments and use semantic line breaks
Assisted-By: Claude Fable 5 <[email protected]>
* Fix TestLogFile and TestLog on Windows
TestLogFile deleted the data file while the writer from setup() still
had it open, which Windows forbids. TestLog filled the disk to exactly
the minimum required space, so any temp files freed by other processes
on a shared CI runner pushed the usable space back above the limit
before the put checked it.
Assisted-By: Claude Fable 5 <[email protected]>
* Open java.lang to the EnvironmentVariables rule
On Windows the rule also patches
ProcessEnvironment.theCaseInsensitiveEnvironment, which lives in
java.lang, while on Linux opening java.util was enough.
Assisted-By: Claude Fable 5 <[email protected]>
* Retry directory deletion in TestFileChannelIntegrityTool
On Windows the checkpoint file of a stopped channel cannot be deleted
until the garbage collector releases its mapping. The failed teardown
leaked files into the shared directories and broke the following tests.
Assisted-By: Claude Fable 5 <[email protected]>
* Open java.lang to the EnvironmentVariables rule in config filter tests
Same fix as in flume-ng-node: on Windows the rule also patches
ProcessEnvironment.theCaseInsensitiveEnvironment, which lives in
java.lang.
Assisted-By: Claude Fable 5 <[email protected]>
* Skip TestExternalProcessConfigFilter on Windows
The external commands under test are shell scripts, which Windows
cannot execute.
Assisted-By: Claude Fable 5 <[email protected]>
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.github/workflows/build.yml | 2 +-
flume-ng-channels/flume-file-channel/pom.xml | 4 +-
.../file/EventQueueBackingStoreFactory.java | 20 ++-
.../channel/file/EventQueueBackingStoreFile.java | 101 +++++++++-----
.../channel/file/EventQueueBackingStoreFileV2.java | 25 ++--
.../channel/file/EventQueueBackingStoreFileV3.java | 146 +++++++++++----------
.../org/apache/flume/channel/file/FileChannel.java | 8 +-
.../apache/flume/channel/file/FlumeEventQueue.java | 90 ++++++++-----
.../java/org/apache/flume/channel/file/Log.java | 23 +++-
.../apache/flume/channel/file/Serialization.java | 36 ++++-
.../apache/flume/channel/file/TestCheckpoint.java | 9 ++
.../channel/file/TestCheckpointRebuilder.java | 13 +-
.../file/TestEventQueueBackingStoreFactory.java | 24 ++--
.../flume/channel/file/TestFileChannelBase.java | 3 +-
.../channel/file/TestFileChannelErrorMetrics.java | 17 +++
.../flume/channel/file/TestFileChannelRestart.java | 20 ++-
.../flume/channel/file/TestFlumeEventQueue.java | 36 ++++-
.../org/apache/flume/channel/file/TestLog.java | 8 +-
.../org/apache/flume/channel/file/TestLogFile.java | 4 +
.../org/apache/flume/channel/file/TestUtils.java | 39 ++++++
.../file/encryption/TestFileChannelEncryption.java | 8 +-
.../pom.xml | 5 +-
.../TestExternalProcessConfigFilter.java | 5 +
flume-ng-node/pom.xml | 5 +-
flume-ng-sources/flume-syslog-source/pom.xml | 6 +
.../flume/source/syslog/TestSyslogTcpSource.java | 22 +++-
.../source/taildir/TestTaildirEventReader.java | 11 ++
.../flume/source/taildir/TestTaildirSource.java | 11 ++
flume-parent/pom.xml | 8 ++
.../flume/tools/TestFileChannelIntegrityTool.java | 26 +++-
.../antora/modules/ROOT/pages/FlumeUserGuide.adoc | 5 +-
31 files changed, 549 insertions(+), 191 deletions(-)
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index afbb2120..49f910d0 100644
--- a/.github/workflows/build.yml
+++ b/.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`.
diff --git a/flume-ng-channels/flume-file-channel/pom.xml
b/flume-ng-channels/flume-file-channel/pom.xml
index 9e7e219c..70c26cd0 100644
--- a/flume-ng-channels/flume-file-channel/pom.xml
+++ b/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>
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..39a7fb1c 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
@@ -20,6 +20,8 @@ import com.google.common.io.Files;
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 @@ class EventQueueBackingStoreFactory {
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);
}
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..43e835c4 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
@@ -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 @@ 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, 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 @@ abstract class EventQueueBackingStoreFile extends
EventQueueBackingStore {
}
protected long getCheckpointLogWriteOrderID() {
+ checkNotClosed();
return elementsBuffer.get(INDEX_WRITE_ORDER_ID);
}
@@ -223,8 +241,22 @@ abstract class EventQueueBackingStoreFile extends
EventQueueBackingStore {
}
}
+ /**
+ * 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 @@ 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 +342,14 @@ abstract class EventQueueBackingStoreFile extends
EventQueueBackingStore {
@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 @@ abstract class EventQueueBackingStoreFile extends
EventQueueBackingStore {
@Override
long get(int index) {
+ checkNotClosed();
int realIndex = getPhysicalIndex(index);
long result = EMPTY;
if (overwriteMap.containsKey(realIndex)) {
@@ -346,6 +386,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..49e96ed6 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
@@ -49,75 +49,83 @@ final class EventQueueBackingStoreFileV3 extends
EventQueueBackingStoreFile {
boolean compressBackup)
throws IOException, BadCheckpointException {
super(capacity, name, counter, checkpointFile, checkpointBackupDir,
backupCheckpoint, compressBackup);
- Preconditions.checkArgument(capacity > 0, "capacity must be greater
than 0 " + capacity);
- metaDataFile = Serialization.getMetaDataFile(checkpointFile);
- logger.info("Starting up with " + checkpointFile + " and " +
metaDataFile);
- if (metaDataFile.exists()) {
- FileInputStream inputStream = new FileInputStream(metaDataFile);
- try {
- logger.info("Reading checkpoint metadata from " +
metaDataFile);
- ProtosFactory.Checkpoint checkpoint =
ProtosFactory.Checkpoint.parseDelimitedFrom(inputStream);
- if (checkpoint == null) {
+ try {
+ Preconditions.checkArgument(capacity > 0, "capacity must be
greater than 0 " + capacity);
+ metaDataFile = Serialization.getMetaDataFile(checkpointFile);
+ logger.info("Starting up with " + checkpointFile + " and " +
metaDataFile);
+ if (metaDataFile.exists()) {
+ FileInputStream inputStream = new
FileInputStream(metaDataFile);
+ try {
+ logger.info("Reading checkpoint metadata from " +
metaDataFile);
+ ProtosFactory.Checkpoint checkpoint =
ProtosFactory.Checkpoint.parseDelimitedFrom(inputStream);
+ if (checkpoint == null) {
+ throw new BadCheckpointException(
+ "The checkpoint metadata file does " + "not
exist or has zero length");
+ }
+ int version = checkpoint.getVersion();
+ if (version != getVersion()) {
+ throw new BadCheckpointException(
+ "Invalid version: " + version + " " + name +
", expected " + getVersion());
+ }
+ long logWriteOrderID = checkpoint.getWriteOrderID();
+ if (logWriteOrderID != getCheckpointLogWriteOrderID()) {
+ String msg = "Checkpoint and Meta files have differing
" + "logWriteOrderIDs "
+ + getCheckpointLogWriteOrderID() + ", and "
+ + logWriteOrderID;
+ logger.warn(msg);
+ throw new BadCheckpointException(msg);
+ }
+ WriteOrderOracle.setSeed(logWriteOrderID);
+ setLogWriteOrderID(logWriteOrderID);
+ setSize(checkpoint.getQueueSize());
+ setHead(checkpoint.getQueueHead());
+ for (ProtosFactory.ActiveLog activeLog :
checkpoint.getActiveLogsList()) {
+ Integer logFileID = activeLog.getLogFileID();
+ Integer count = activeLog.getCount();
+ logFileIDReferenceCounts.put(logFileID, new
AtomicInteger(count));
+ }
+ } catch (InvalidProtocolBufferException ex) {
throw new BadCheckpointException(
- "The checkpoint metadata file does " + "not exist
or has zero length");
+ "Checkpoint metadata file is invalid. "
+ + "The agent might have been stopped while
it was being "
+ + "written",
+ ex);
+ } finally {
+ try {
+ inputStream.close();
+ } catch (IOException e) {
+ logger.warn("Unable to close " + metaDataFile, e);
+ }
}
- int version = checkpoint.getVersion();
- if (version != getVersion()) {
+ } else {
+ if (backupExists(checkpointBackupDir) && shouldBackup) {
+ // If a backup exists, then throw an exception to recover
checkpoint
throw new BadCheckpointException(
- "Invalid version: " + version + " " + name + ",
expected " + getVersion());
- }
- long logWriteOrderID = checkpoint.getWriteOrderID();
- if (logWriteOrderID != getCheckpointLogWriteOrderID()) {
- String msg = "Checkpoint and Meta files have differing " +
"logWriteOrderIDs "
- + getCheckpointLogWriteOrderID() + ", and "
- + logWriteOrderID;
- logger.warn(msg);
- throw new BadCheckpointException(msg);
+ "The checkpoint metadata file does " + "not exist,
but a backup exists");
}
- WriteOrderOracle.setSeed(logWriteOrderID);
- setLogWriteOrderID(logWriteOrderID);
- setSize(checkpoint.getQueueSize());
- setHead(checkpoint.getQueueHead());
- for (ProtosFactory.ActiveLog activeLog :
checkpoint.getActiveLogsList()) {
- Integer logFileID = activeLog.getLogFileID();
- Integer count = activeLog.getCount();
- logFileIDReferenceCounts.put(logFileID, new
AtomicInteger(count));
- }
- } catch (InvalidProtocolBufferException ex) {
- throw new BadCheckpointException(
- "Checkpoint metadata file is invalid. "
- + "The agent might have been stopped while it
was being "
- + "written",
- ex);
- } finally {
+ ProtosFactory.Checkpoint.Builder checkpointBuilder =
ProtosFactory.Checkpoint.newBuilder();
+ checkpointBuilder.setVersion(getVersion());
+ checkpointBuilder.setQueueHead(getHead());
+ checkpointBuilder.setQueueSize(getSize());
+ checkpointBuilder.setWriteOrderID(getLogWriteOrderID());
+ FileOutputStream outputStream = new
FileOutputStream(metaDataFile);
try {
- inputStream.close();
- } catch (IOException e) {
- logger.warn("Unable to close " + metaDataFile, e);
- }
- }
- } else {
- if (backupExists(checkpointBackupDir) && shouldBackup) {
- // If a backup exists, then throw an exception to recover
checkpoint
- throw new BadCheckpointException(
- "The checkpoint metadata file does " + "not exist, but
a backup exists");
- }
- ProtosFactory.Checkpoint.Builder checkpointBuilder =
ProtosFactory.Checkpoint.newBuilder();
- checkpointBuilder.setVersion(getVersion());
- checkpointBuilder.setQueueHead(getHead());
- checkpointBuilder.setQueueSize(getSize());
- checkpointBuilder.setWriteOrderID(getLogWriteOrderID());
- FileOutputStream outputStream = new FileOutputStream(metaDataFile);
- try {
- checkpointBuilder.build().writeDelimitedTo(outputStream);
- outputStream.getChannel().force(true);
- } finally {
- try {
- outputStream.close();
- } catch (IOException e) {
- logger.warn("Unable to close " + metaDataFile, e);
+ checkpointBuilder.build().writeDelimitedTo(outputStream);
+ outputStream.getChannel().force(true);
+ } finally {
+ try {
+ outputStream.close();
+ } catch (IOException e) {
+ logger.warn("Unable to close " + metaDataFile, e);
+ }
}
}
+ } catch (IOException | RuntimeException e) {
+ // Close the store opened by the superclass:
+ // dropping the buffer references lets the garbage collector
release the mapping
+ // before the recovery path deletes the checkpoint file.
+ close();
+ throw e;
}
}
@@ -159,14 +167,14 @@ final class EventQueueBackingStoreFileV3 extends
EventQueueBackingStoreFile {
}
}
- static void upgrade(EventQueueBackingStoreFileV2 backingStoreV2, File
checkpointFile, File metaDataFile)
+ static void upgrade(
+ File checkpointFile,
+ File metaDataFile,
+ int head,
+ int size,
+ long writeOrderID,
+ Map<Integer, AtomicInteger> referenceCounts)
throws IOException {
-
- int head = backingStoreV2.getHead();
- int size = backingStoreV2.getSize();
- long writeOrderID = backingStoreV2.getLogWriteOrderID();
- Map<Integer, AtomicInteger> referenceCounts =
backingStoreV2.logFileIDReferenceCounts;
-
ProtosFactory.Checkpoint.Builder checkpointBuilder =
ProtosFactory.Checkpoint.newBuilder();
checkpointBuilder.setVersion(Serialization.VERSION_3);
checkpointBuilder.setQueueHead(head);
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..2c7d52d2 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..ef67f87c 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,63 @@ 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 +409,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..c7da2159 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 and drop all
references to them:
+ // the checkpoint mapping is only released when the buffer is
garbage collected,
+ // and the deletion below cannot succeed on Windows until then.
+ if (queue != null) {
+ try {
+ queue.close();
+ } catch (Exception closeEx) {
+ ex.addSuppressed(closeEx);
+ }
+ queue = null;
+ } else if (backingStore != null) {
+ try {
+ backingStore.close();
+ } catch (Exception closeEx) {
+ ex.addSuppressed(closeEx);
+ }
+ }
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) {
diff --git
a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Serialization.java
b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Serialization.java
index 309f1ac6..b58d2c44 100644
---
a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Serialization.java
+++
b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/Serialization.java
@@ -26,10 +26,12 @@ import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.Collections;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.conf.internal.SuppressFBWarnings;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.xerial.snappy.SnappyInputStream;
@@ -53,6 +55,10 @@ public class Serialization {
// 64 K buffer to copy and compress files.
private static final int FILE_BUFFER_SIZE = 64 * 1024;
+ // Retries for deleting files that a garbage-collectable mapping still
pins.
+ private static final int DELETE_ATTEMPTS = 10;
+ private static final long DELETE_RETRY_DELAY_MILLIS = 100;
+
private static final Logger logger = LogManager.getLogger();
static File getMetaDataTempFile(File metaDataFile) {
@@ -102,7 +108,7 @@ public class Serialization {
logger.info("Skipping " + file.getName() + " because it is in
excludes " + "set");
continue;
}
- if (!FileUtils.deleteQuietly(file)) {
+ if (!deleteWithRetries(file)) {
logger.info(builder.toString());
logger.error("Error while attempting to delete: " +
file.getAbsolutePath());
return false;
@@ -114,6 +120,34 @@ public class Serialization {
return true;
}
+ /**
+ * Deletes a file, retrying with garbage collection cycles in between.
+ *
+ * <p>On Windows the checkpoint file cannot be deleted while a discarded
backing store still maps it.
+ * Nothing unmaps the buffer explicitly,
+ * so ask for a collection and retry a bounded number of times.
+ * The first attempt is free, so healthy platforms and unmapped files pay
nothing.
+ */
+ @SuppressFBWarnings(value = "DM_GC")
+ private static boolean deleteWithRetries(File file) {
+ if (FileUtils.deleteQuietly(file)) {
+ return true;
+ }
+ for (int attempt = 0; attempt < DELETE_ATTEMPTS; attempt++) {
+ System.gc();
+ try {
+ TimeUnit.MILLISECONDS.sleep(DELETE_RETRY_DELAY_MILLIS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ if (FileUtils.deleteQuietly(file)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Copy a file using a 64K size buffer. This method will copy the file and
* then fsync to disk
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestCheckpoint.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestCheckpoint.java
index 54b51222..4fffa2dd 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestCheckpoint.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestCheckpoint.java
@@ -19,6 +19,7 @@ package org.apache.flume.channel.file;
import java.io.File;
import java.io.IOException;
import junit.framework.Assert;
+import org.apache.commons.io.FileUtils;
import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
import org.junit.After;
import org.junit.Before;
@@ -43,6 +44,9 @@ public class TestCheckpoint {
@After
public void cleanup() {
file.delete();
+ inflightPuts.delete();
+ inflightTakes.delete();
+ FileUtils.deleteQuietly(queueSet);
}
@Test
@@ -52,12 +56,17 @@ public class TestCheckpoint {
FlumeEventPointer ptrIn = new FlumeEventPointer(10, 20);
FlumeEventQueue queueIn = new FlumeEventQueue(backingStore,
inflightTakes, inflightPuts, queueSet);
queueIn.addHead(ptrIn);
+ // The three queues share one backing store and one queue set
directory,
+ // so each queue must release the queue set database before the next
queue is created.
+ queueIn.replayComplete();
FlumeEventQueue queueOut = new FlumeEventQueue(backingStore,
inflightTakes, inflightPuts, queueSet);
Assert.assertEquals(0, queueOut.getLogWriteOrderID());
+ queueOut.replayComplete();
queueIn.checkpoint(false);
FlumeEventQueue queueOut2 = new FlumeEventQueue(backingStore,
inflightTakes, inflightPuts, queueSet);
FlumeEventPointer ptrOut = queueOut2.removeHead(0L);
Assert.assertEquals(ptrIn, ptrOut);
Assert.assertTrue(queueOut2.getLogWriteOrderID() > 0);
+ queueOut2.close();
}
}
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestCheckpointRebuilder.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestCheckpointRebuilder.java
index 7ebebc7e..2e40edcf 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestCheckpointRebuilder.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestCheckpointRebuilder.java
@@ -53,20 +53,27 @@ public class TestCheckpointRebuilder extends
TestFileChannelBase {
Assert.assertTrue(channel.isOpen());
Set<String> in = fillChannel(channel, "checkpointBulder");
channel.stop();
+ // Drop the channel reference so the GC can release the checkpoint
mapping.
+ channel = null;
File checkpointFile = new File(checkpointDir, "checkpoint");
File metaDataFile = Serialization.getMetaDataFile(checkpointFile);
File inflightTakesFile = new File(checkpointDir, "inflighttakes");
File inflightPutsFile = new File(checkpointDir, "inflightputs");
File queueSetDir = new File(checkpointDir, "queueset");
- Assert.assertTrue(checkpointFile.delete());
+ TestUtils.awaitDelete(checkpointFile);
Assert.assertTrue(metaDataFile.delete());
Assert.assertTrue(inflightTakesFile.delete());
Assert.assertTrue(inflightPutsFile.delete());
EventQueueBackingStore backingStore =
EventQueueBackingStoreFactory.get(checkpointFile, 50, "test",
new FileChannelCounter("test"));
FlumeEventQueue queue = new FlumeEventQueue(backingStore,
inflightTakesFile, inflightPutsFile, queueSetDir);
- CheckpointRebuilder checkpointRebuilder = new
CheckpointRebuilder(getAllLogs(dataDirs), queue, true);
- Assert.assertTrue(checkpointRebuilder.rebuild());
+ try {
+ CheckpointRebuilder checkpointRebuilder = new
CheckpointRebuilder(getAllLogs(dataDirs), queue, true);
+ Assert.assertTrue(checkpointRebuilder.rebuild());
+ } finally {
+ // Release the checkpoint files before the channel below replays
them.
+ queue.close();
+ }
channel = createFileChannel(overrides);
channel.start();
Assert.assertTrue(channel.isOpen());
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestEventQueueBackingStoreFactory.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestEventQueueBackingStoreFactory.java
index fb26f382..1c8c8f12 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestEventQueueBackingStoreFactory.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestEventQueueBackingStoreFactory.java
@@ -272,16 +272,20 @@ public class TestEventQueueBackingStoreFactory {
private void verify(EventQueueBackingStore backingStore, long
expectedVersion, List<Long> expectedPointers)
throws Exception {
FlumeEventQueue queue = new FlumeEventQueue(backingStore,
inflightTakes, inflightPuts, queueSetDir);
- List<Long> actualPointers = Lists.newArrayList();
- FlumeEventPointer ptr;
- while ((ptr = queue.removeHead(0L)) != null) {
- actualPointers.add(ptr.toLong());
+ try {
+ List<Long> actualPointers = Lists.newArrayList();
+ FlumeEventPointer ptr;
+ while ((ptr = queue.removeHead(0L)) != null) {
+ actualPointers.add(ptr.toLong());
+ }
+ Assert.assertEquals(expectedPointers, actualPointers);
+ Assert.assertEquals(10, backingStore.getCapacity());
+ DataInputStream in = new DataInputStream(new
FileInputStream(checkpoint));
+ long actualVersion = in.readLong();
+ Assert.assertEquals(expectedVersion, actualVersion);
+ in.close();
+ } finally {
+ queue.close();
}
- Assert.assertEquals(expectedPointers, actualPointers);
- Assert.assertEquals(10, backingStore.getCapacity());
- DataInputStream in = new DataInputStream(new
FileInputStream(checkpoint));
- long actualVersion = in.readLong();
- Assert.assertEquals(expectedVersion, actualVersion);
- in.close();
}
}
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelBase.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelBase.java
index 46257d14..bc821af2 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelBase.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelBase.java
@@ -70,7 +70,8 @@ public class TestFileChannelBase {
@After
public void teardown() {
- if (channel != null && channel.isOpen()) {
+ // Stop the channel even when it failed to start: it still holds open
files.
+ if (channel != null) {
channel.stop();
}
FileUtils.deleteQuietly(baseDir);
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelErrorMetrics.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelErrorMetrics.java
index 4b435ec0..22bfebee 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelErrorMetrics.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelErrorMetrics.java
@@ -33,6 +33,7 @@ import org.apache.flume.Transaction;
import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.exception.ChannelException;
+import org.junit.Assume;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
@@ -46,12 +47,26 @@ public class TestFileChannelErrorMetrics extends
TestFileChannelBase {
super(1);
}
+ /**
+ * Skips tests that delete in-use files on Windows.
+ *
+ * <p>These tests inject I/O errors by deleting the channel's files while
the channel is using them.
+ * Windows does not allow deleting files that are in use,
+ * so the deletion fails in the test instead of causing the intended
errors in the channel.
+ */
+ private static void requireOpenFileDeletion() {
+ Assume.assumeFalse(
+ "Windows cannot delete files that are in use",
+ System.getProperty("os.name").startsWith("Windows"));
+ }
+
/**
* This tests multiple successful and failed put and take operations
* and checks the values of the channel's counters.
*/
@Test
public void testEventTakePutErrorCount() throws Exception {
+ requireOpenFileDeletion();
final long usableSpaceRefreshInterval = 1;
FileChannel channel = Mockito.spy(createFileChannel());
Mockito.when(channel.createLogBuilder()).then(new
Answer<Log.Builder>() {
@@ -161,6 +176,7 @@ public class TestFileChannelErrorMetrics extends
TestFileChannelBase {
@Test
public void testCheckpointWriteErrorCount() throws Exception {
+ requireOpenFileDeletion();
int checkpointInterval = 1500;
final FileChannel channel = createFileChannel(Collections.singletonMap(
FileChannelConfiguration.CHECKPOINT_INTERVAL,
String.valueOf(checkpointInterval)));
@@ -262,6 +278,7 @@ public class TestFileChannelErrorMetrics extends
TestFileChannelBase {
@Test
public void testCheckpointBackupWriteErrorShouldIncreaseCounter2() throws
Exception {
+ requireOpenFileDeletion();
int checkpointInterval = 1500;
Map config = new HashMap();
config.put(FileChannelConfiguration.CHECKPOINT_INTERVAL,
String.valueOf(checkpointInterval));
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelRestart.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelRestart.java
index 69bb9a4b..f18b04b3 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelRestart.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFileChannelRestart.java
@@ -126,8 +126,10 @@ public class TestFileChannelRestart extends
TestFileChannelBase {
}
channel.stop();
if (deleteCheckpoint) {
+ // Drop the channel reference so the GC can release the checkpoint
mapping.
+ channel = null;
File checkpoint = new File(checkpointDir, "checkpoint");
- Assert.assertTrue(checkpoint.delete());
+ TestUtils.awaitDelete(checkpoint);
File checkpointMetaData =
Serialization.getMetaDataFile(checkpoint);
Assert.assertTrue(checkpointMetaData.delete());
}
@@ -161,8 +163,10 @@ public class TestFileChannelRestart extends
TestFileChannelBase {
Thread.sleep(2000);
}
channel.stop();
+ // Drop the channel reference so the GC can release the checkpoint
mapping.
+ channel = null;
File checkpoint = new File(checkpointDir, "checkpoint");
- Assert.assertTrue(checkpoint.delete());
+ TestUtils.awaitDelete(checkpoint);
File checkpointMetaData = Serialization.getMetaDataFile(checkpoint);
Assert.assertTrue(checkpointMetaData.exists());
channel = createFileChannel(overrides);
@@ -235,10 +239,12 @@ public class TestFileChannelRestart extends
TestFileChannelBase {
Thread.sleep(2000);
}
channel.stop();
+ // Drop the channel reference so the GC can release the checkpoint
mapping.
+ channel = null;
File checkpoint = new File(checkpointDir, "checkpoint");
File checkpointMetaData = Serialization.getMetaDataFile(checkpoint);
Assert.assertTrue(checkpointMetaData.delete());
- Assert.assertTrue(checkpoint.delete());
+ TestUtils.awaitDelete(checkpoint);
channel = createFileChannel(overrides);
channel.start();
Assert.assertTrue(channel.isOpen());
@@ -317,6 +323,8 @@ public class TestFileChannelRestart extends
TestFileChannelBase {
FileOutputStream os = new
FileOutputStream(Serialization.getMetaDataFile(checkpoint));
meta.toBuilder().setVersion(2).build().writeDelimitedTo(os);
os.flush();
+ // Close the stream, or the recovery below cannot delete the file on
Windows.
+ os.close();
channel = createFileChannel(overrides);
channel.start();
Assert.assertTrue(channel.isOpen());
@@ -356,6 +364,8 @@ public class TestFileChannelRestart extends
TestFileChannelBase {
FileOutputStream os = new
FileOutputStream(Serialization.getMetaDataFile(checkpoint));
meta.toBuilder().setWriteOrderID(12).build().writeDelimitedTo(os);
os.flush();
+ // Close the stream, or the recovery below cannot delete the file on
Windows.
+ os.close();
channel = createFileChannel(overrides);
channel.start();
Assert.assertTrue(channel.isOpen());
@@ -869,8 +879,10 @@ public class TestFileChannelRestart extends
TestFileChannelBase {
Assert.assertEquals(compressedBackupCheckpoint.exists(),
originalCheckpointCompressed);
Assert.assertEquals(uncompressedBackupCheckpoint.exists(),
!originalCheckpointCompressed);
channel.stop();
+ // Drop the channel reference so the GC can release the checkpoint
mapping.
+ channel = null;
File checkpoint = new File(checkpointDir, "checkpoint");
- Assert.assertTrue(checkpoint.delete());
+ TestUtils.awaitDelete(checkpoint);
File checkpointMetaData = Serialization.getMetaDataFile(checkpoint);
Assert.assertTrue(checkpointMetaData.delete());
overrides.put(
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFlumeEventQueue.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFlumeEventQueue.java
index c8c3c2a9..f5f148ce 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFlumeEventQueue.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestFlumeEventQueue.java
@@ -54,6 +54,17 @@ public class TestFlumeEventQueue {
File queueSetDir;
EventQueueBackingStoreSupplier() {
+ reset();
+ }
+
+ /**
+ * Creates a fresh directory for each test.
+ *
+ * <p>The supplier instances are shared by all test methods of a
parameter.
+ * A file leaked by one test, e.g. still open on Windows,
+ * must not break the cleanup of the following tests.
+ */
+ void reset() {
baseDir = Files.createTempDir();
checkpoint = new File(baseDir, "checkpoint");
inflightTakes = new File(baseDir, "inflightputs");
@@ -117,11 +128,16 @@ public class TestFlumeEventQueue {
@Before
public void setup() throws Exception {
+ backingStoreSupplier.reset();
backingStore = backingStoreSupplier.get();
}
@After
public void cleanup() throws IOException {
+ if (queue != null) {
+ queue.close();
+ queue = null;
+ }
if (backingStore != null) {
backingStore.close();
}
@@ -131,8 +147,10 @@ public class TestFlumeEventQueue {
@Test
public void testCapacity() throws Exception {
backingStore.close();
+ // Drop the reference so the GC can release the mapping and the file
can be deleted.
+ backingStore = null;
File checkpoint = backingStoreSupplier.getCheckpoint();
- Assert.assertTrue(checkpoint.delete());
+ TestUtils.awaitDelete(checkpoint);
backingStore = new EventQueueBackingStoreFileV2(checkpoint, 1, "test",
new FileChannelCounter("test"));
queue = new FlumeEventQueue(
backingStore,
@@ -146,8 +164,10 @@ public class TestFlumeEventQueue {
@Test(expected = IllegalArgumentException.class)
public void testInvalidCapacityZero() throws Exception {
backingStore.close();
+ // Drop the reference so the GC can release the mapping and the file
can be deleted.
+ backingStore = null;
File checkpoint = backingStoreSupplier.getCheckpoint();
- Assert.assertTrue(checkpoint.delete());
+ TestUtils.awaitDelete(checkpoint);
backingStore = new EventQueueBackingStoreFileV2(checkpoint, 0, "test",
new FileChannelCounter("test"));
queue = new FlumeEventQueue(
backingStore,
@@ -159,8 +179,10 @@ public class TestFlumeEventQueue {
@Test(expected = IllegalArgumentException.class)
public void testInvalidCapacityNegative() throws Exception {
backingStore.close();
+ // Drop the reference so the GC can release the mapping and the file
can be deleted.
+ backingStore = null;
File checkpoint = backingStoreSupplier.getCheckpoint();
- Assert.assertTrue(checkpoint.delete());
+ TestUtils.awaitDelete(checkpoint);
backingStore = new EventQueueBackingStoreFileV2(checkpoint, -1,
"test", new FileChannelCounter("test"));
queue = new FlumeEventQueue(
backingStore,
@@ -400,6 +422,8 @@ public class TestFlumeEventQueue {
queue.addWithoutCommit(new FlumeEventPointer(2, 2), txnID2);
queue.checkpoint(true);
TimeUnit.SECONDS.sleep(3L);
+ // Close the MapDB, so that the new queue can delete the queueset
directory on Windows.
+ queue.replayComplete();
queue = new FlumeEventQueue(
backingStore,
backingStoreSupplier.getInflightTakes(),
@@ -428,6 +452,8 @@ public class TestFlumeEventQueue {
queue.removeHead(txnID2);
queue.checkpoint(true);
TimeUnit.SECONDS.sleep(3L);
+ // Close the MapDB, so that the new queue can delete the queueset
directory on Windows.
+ queue.replayComplete();
queue = new FlumeEventQueue(
backingStore,
backingStoreSupplier.getInflightTakes(),
@@ -458,6 +484,8 @@ public class TestFlumeEventQueue {
inflight = new
RandomAccessFile(backingStoreSupplier.getInflightPuts(), "rw");
inflight.seek(0);
inflight.writeInt(new Random().nextInt());
+ // Close the MapDB, so that the new queue can delete the queueset
directory on Windows.
+ queue.replayComplete();
queue = new FlumeEventQueue(
backingStore,
backingStoreSupplier.getInflightTakes(),
@@ -491,6 +519,8 @@ public class TestFlumeEventQueue {
inflight = new
RandomAccessFile(backingStoreSupplier.getInflightTakes(), "rw");
inflight.seek(0);
inflight.writeInt(new Random().nextInt());
+ // Close the MapDB, so that the new queue can delete the queueset
directory on Windows.
+ queue.replayComplete();
queue = new FlumeEventQueue(
backingStore,
backingStoreSupplier.getInflightTakes(),
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java
index 879949be..9e0ed60d 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java
@@ -224,8 +224,12 @@ public class TestLog {
log.replay();
File filler = new File(checkpointDir, "filler");
byte[] buffer = new byte[64 * 1024];
+ // Fill well below the limit:
+ // other processes on a shared CI runner free temp space all the time,
+ // and a thin margin lets the usable space bounce back above the limit
before the put checks it.
+ long margin = 2L * 1024L * 1024L;
FileOutputStream out = new FileOutputStream(filler);
- while (checkpointDir.getUsableSpace() > minimumRequiredSpace) {
+ while (checkpointDir.getUsableSpace() > minimumRequiredSpace - margin)
{
out.write(buffer);
}
out.close();
@@ -419,7 +423,7 @@ public class TestLog {
log.commitPut(transactionID);
log.close();
if (useFastReplay) {
- FileUtils.deleteQuietly(checkpointDir);
+ TestUtils.awaitDeleteDirectory(checkpointDir);
Assert.assertTrue(checkpointDir.mkdir());
}
List<File> logFiles = Lists.newArrayList();
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java
index d2e003be..79bca4a7 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLogFile.java
@@ -85,6 +85,8 @@ public class TestLogFile {
@Test
public void testWriterFailsWithDirectory() throws IOException {
+ // Close the writer opened by setup(), so that the file can be deleted
on Windows.
+ logFileWriter.close();
FileUtils.deleteQuietly(dataFile);
Assert.assertFalse(dataFile.exists());
Assert.assertTrue(dataFile.mkdirs());
@@ -245,6 +247,8 @@ public class TestLogFile {
@Test
public void testWriteDelimitedTo() throws IOException {
+ // Close the writer opened by setup(), so that the file can be deleted
on Windows.
+ logFileWriter.close();
if (dataFile.isFile()) {
Assert.assertTrue(dataFile.delete());
}
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
index 51d62848..2fc4aa93 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestUtils.java
@@ -41,6 +41,7 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.zip.GZIPInputStream;
+import org.apache.commons.io.FileUtils;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
@@ -52,6 +53,44 @@ import org.junit.Assert;
public class TestUtils {
+ private static final int DELETE_ATTEMPTS = 50;
+ private static final long DELETE_RETRY_DELAY_MILLIS = 100;
+
+ /**
+ * Deletes a file that a discarded backing store may still map.
+ *
+ * <p>On Windows a mapped file cannot be deleted,
+ * and the mapping is only released when the buffer is garbage collected,
+ * so this method retries with garbage collection cycles in between.
+ * The caller must first drop all references to the store and to anything
that references it,
+ * such as a stopped channel,
+ * otherwise the mapping stays reachable and this method fails the test.
+ */
+ public static void awaitDelete(File file) throws InterruptedException {
+ for (int attempt = 0; attempt < DELETE_ATTEMPTS; attempt++) {
+ if (!file.exists() || file.delete()) {
+ return;
+ }
+ System.gc();
+ Thread.sleep(DELETE_RETRY_DELAY_MILLIS);
+ }
+ Assert.fail("Could not delete " + file + ", a mapping is probably
still reachable");
+ }
+
+ /**
+ * Recursively deletes a directory, retrying like {@link
#awaitDelete(File)}.
+ */
+ public static void awaitDeleteDirectory(File dir) throws
InterruptedException {
+ for (int attempt = 0; attempt < DELETE_ATTEMPTS; attempt++) {
+ if (FileUtils.deleteQuietly(dir) && !dir.exists()) {
+ return;
+ }
+ System.gc();
+ Thread.sleep(DELETE_RETRY_DELAY_MILLIS);
+ }
+ Assert.fail("Could not delete " + dir + ", a mapping is probably still
reachable");
+ }
+
public static FlumeEvent newPersistableEvent() {
Map<String, String> headers = Maps.newHashMap();
String timestamp = String.valueOf(System.currentTimeMillis());
diff --git
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestFileChannelEncryption.java
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestFileChannelEncryption.java
index 5c8499d6..d88e7a8f 100644
---
a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestFileChannelEncryption.java
+++
b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestFileChannelEncryption.java
@@ -314,9 +314,11 @@ public class TestFileChannelEncryption extends
TestFileChannelBase {
channel = createFileChannel(overrides);
Assert.fail();
} catch (RuntimeException ex) {
+ // The path separator and the reason text are platform-dependent.
Assert.assertTrue(
"Exception message is incorrect: " + ex.getMessage(),
- ex.getMessage().startsWith("java.io.FileNotFoundException:
/path/does/not/exist "));
+ ex.getMessage()
+ .startsWith("java.io.FileNotFoundException: " +
new File("/path/does/not/exist") + " "));
}
}
@@ -334,9 +336,11 @@ public class TestFileChannelEncryption extends
TestFileChannelBase {
channel = createFileChannel(overrides);
Assert.fail();
} catch (RuntimeException ex) {
+ // The path separator and the reason text are platform-dependent.
Assert.assertTrue(
"Exception message is incorrect: " + ex.getMessage(),
- ex.getMessage().startsWith("java.io.FileNotFoundException:
/path/does/not/exist "));
+ ex.getMessage()
+ .startsWith("java.io.FileNotFoundException: " +
new File("/path/does/not/exist") + " "));
}
}
diff --git
a/flume-ng-configfilters/flume-ng-environment-variable-config-filter/pom.xml
b/flume-ng-configfilters/flume-ng-environment-variable-config-filter/pom.xml
index 9126933b..78c30b0c 100644
--- a/flume-ng-configfilters/flume-ng-environment-variable-config-filter/pom.xml
+++ b/flume-ng-configfilters/flume-ng-environment-variable-config-filter/pom.xml
@@ -52,7 +52,10 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
- <argLine>--add-opens java.base/java.util=ALL-UNNAMED</argLine>
+ <!-- `java.lang` is opened for the `EnvironmentVariables` rule,
+ which needs `ProcessEnvironment.theCaseInsensitiveEnvironment`
on Windows. -->
+ <argLine>--add-opens java.base/java.util=ALL-UNNAMED
+ --add-opens java.base/java.lang=ALL-UNNAMED</argLine>
</configuration>
</plugin>
</plugins>
diff --git
a/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/java/org/apache/flume/configfilter/TestExternalProcessConfigFilter.java
b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/java/org/apache/flume/configfilter/TestExternalProcessConfigFilter.java
index ed59341a..2f3e6253 100644
---
a/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/java/org/apache/flume/configfilter/TestExternalProcessConfigFilter.java
+++
b/flume-ng-configfilters/flume-ng-external-process-config-filter/src/test/java/org/apache/flume/configfilter/TestExternalProcessConfigFilter.java
@@ -21,6 +21,7 @@ import static org.junit.Assert.assertNull;
import java.io.File;
import java.util.HashMap;
+import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
@@ -34,6 +35,10 @@ public class TestExternalProcessConfigFilter {
@Before
public void setUp() {
+ // The external commands are shell scripts, which Windows cannot
execute.
+ Assume.assumeFalse(
+ "Windows cannot execute the test shell scripts",
+ System.getProperty("os.name").startsWith("Windows"));
configFilter = new ExternalProcessConfigFilter();
}
diff --git a/flume-ng-node/pom.xml b/flume-ng-node/pom.xml
index 0542d117..3f96fb84 100644
--- a/flume-ng-node/pom.xml
+++ b/flume-ng-node/pom.xml
@@ -186,8 +186,11 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
+ <!-- `java.lang` is opened for the `EnvironmentVariables` rule,
+ which needs `ProcessEnvironment.theCaseInsensitiveEnvironment`
on Windows. -->
<argLine>-Djava.net.preferIPv4Stack=true
- --add-opens java.base/java.util=ALL-UNNAMED</argLine>
+ --add-opens java.base/java.util=ALL-UNNAMED
+ --add-opens java.base/java.lang=ALL-UNNAMED</argLine>
</configuration>
</plugin>
diff --git a/flume-ng-sources/flume-syslog-source/pom.xml
b/flume-ng-sources/flume-syslog-source/pom.xml
index f4bf49c1..d169d164 100644
--- a/flume-ng-sources/flume-syslog-source/pom.xml
+++ b/flume-ng-sources/flume-syslog-source/pom.xml
@@ -74,6 +74,12 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
diff --git
a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogTcpSource.java
b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogTcpSource.java
index cbc6f404..d6cf09a4 100644
---
a/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogTcpSource.java
+++
b/flume-ng-sources/flume-syslog-source/src/test/java/org/apache/flume/source/syslog/TestSyslogTcpSource.java
@@ -16,7 +16,9 @@
*/
package org.apache.flume.source.syslog;
+import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
@@ -269,17 +271,23 @@ public class TestSyslogTcpSource {
},
null);
SocketFactory socketFactory = sslContext.getSocketFactory();
- Socket socket = socketFactory.createSocket();
- socket.connect(address);
- OutputStream outputStream = socket.getOutputStream();
- outputStream.write(bodyWithTandH.getBytes());
- socket.close();
- // Thread.sleep(100);
+ try (Socket socket = socketFactory.createSocket()) {
+ socket.connect(address);
+ OutputStream outputStream = socket.getOutputStream();
+ outputStream.write(bodyWithTandH.getBytes());
+ outputStream.flush();
+ // Close the socket only after the server has committed the event
to the channel.
+ // The client never reads the TLS 1.3 session tickets sent by the
server,
+ // so an earlier close() aborts the connection with a TCP RST,
+ // which on Windows can discard the data before the server reads
it.
+ await().until(() ->
source.getSourceCounter().getEventAcceptedCount() >= 1);
+ }
Transaction transaction = channel.getTransaction();
transaction.begin();
Event event = channel.take();
- assertEquals(new String(event.getBody()), data1);
+ assertNotNull("The source accepted an event, but the channel did not
deliver one.", event);
+ assertEquals(data1, new String(event.getBody()));
transaction.commit();
transaction.close();
}
diff --git
a/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirEventReader.java
b/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirEventReader.java
index bb198640..54d57380 100644
---
a/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirEventReader.java
+++
b/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirEventReader.java
@@ -31,18 +31,29 @@ import com.google.common.collect.Table;
import com.google.common.io.Files;
import java.io.File;
import java.io.IOException;
+import java.nio.file.FileSystems;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.flume.Event;
import org.junit.After;
+import org.junit.Assume;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Test;
public class TestTaildirEventReader {
private File tmpDir;
private String posFilePath;
+ @BeforeClass
+ public static void requireUnixFileAttributeView() {
+ // ReliableTaildirEventReader tracks files by inode via the `unix:ino`
file attribute
+ Assume.assumeTrue(
+ "The default file system does not provide a `unix` file
attribute view",
+
FileSystems.getDefault().supportedFileAttributeViews().contains("unix"));
+ }
+
public static String bodyAsString(Event event) {
return new String(event.getBody());
}
diff --git
a/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirSource.java
b/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirSource.java
index 2991bfb7..4c8ec6a6 100644
---
a/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirSource.java
+++
b/flume-ng-sources/flume-taildir-source/src/test/java/org/apache/flume/source/taildir/TestTaildirSource.java
@@ -39,6 +39,7 @@ import com.google.common.collect.Lists;
import com.google.common.io.Files;
import java.io.File;
import java.io.IOException;
+import java.nio.file.FileSystems;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -58,7 +59,9 @@ import org.apache.flume.lifecycle.LifecycleController;
import org.apache.flume.lifecycle.LifecycleState;
import org.apache.flume.util.Whitebox;
import org.junit.After;
+import org.junit.Assume;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
@@ -68,6 +71,14 @@ public class TestTaildirSource {
private File tmpDir;
private String posFilePath;
+ @BeforeClass
+ public static void requireUnixFileAttributeView() {
+ // TaildirSource tracks files by inode via the `unix:ino` file
attribute
+ Assume.assumeTrue(
+ "The default file system does not provide a `unix` file
attribute view",
+
FileSystems.getDefault().supportedFileAttributeViews().contains("unix"));
+ }
+
@Before
public void setUp() {
source = new TaildirSource();
diff --git a/flume-parent/pom.xml b/flume-parent/pom.xml
index 56a3f80b..00c1969a 100644
--- a/flume-parent/pom.xml
+++ b/flume-parent/pom.xml
@@ -95,6 +95,7 @@
<test.include.pattern>**/Test*.java</test.include.pattern>
<redirectTestOutput>true</redirectTestOutput>
+ <awaitility.version>4.3.0</awaitility.version>
<curator.version>5.9.0</curator.version>
<fest-reflect.version>1.4.1</fest-reflect.version>
<hadoop.version>3.5.0</hadoop.version>
@@ -118,6 +119,13 @@
<scope>import</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
diff --git
a/flume-tools/src/test/java/org/apache/flume/tools/TestFileChannelIntegrityTool.java
b/flume-tools/src/test/java/org/apache/flume/tools/TestFileChannelIntegrityTool.java
index b829a55d..91da39c0 100644
---
a/flume-tools/src/test/java/org/apache/flume/tools/TestFileChannelIntegrityTool.java
+++
b/flume-tools/src/test/java/org/apache/flume/tools/TestFileChannelIntegrityTool.java
@@ -85,14 +85,32 @@ public class TestFileChannelIntegrityTool {
@After
public void tearDown() throws Exception {
- FileUtils.deleteDirectory(checkpointDir);
- FileUtils.deleteDirectory(dataDir);
+ deleteDirectoryWithRetries(checkpointDir);
+ deleteDirectoryWithRetries(dataDir);
}
@AfterClass
public static void tearDownClass() throws Exception {
- FileUtils.deleteDirectory(origCheckpointDir);
- FileUtils.deleteDirectory(origDataDir);
+ deleteDirectoryWithRetries(origCheckpointDir);
+ deleteDirectoryWithRetries(origDataDir);
+ }
+
+ /**
+ * Deletes a directory, retrying with garbage collection cycles in between.
+ *
+ * <p>On Windows the checkpoint file of a stopped channel cannot be deleted
+ * until the garbage collector releases its mapping.
+ * A failed deletion would leak files into the next test's directories.
+ */
+ private static void deleteDirectoryWithRetries(File dir) throws
InterruptedException {
+ for (int attempt = 0; attempt < 50; attempt++) {
+ if (FileUtils.deleteQuietly(dir) && !dir.exists()) {
+ return;
+ }
+ System.gc();
+ Thread.sleep(100);
+ }
+ Assert.fail("Could not delete " + dir + ", a mapping is probably still
reachable");
}
@Test
diff --git a/src/site/antora/modules/ROOT/pages/FlumeUserGuide.adoc
b/src/site/antora/modules/ROOT/pages/FlumeUserGuide.adoc
index 72d2428c..397c8027 100644
--- a/src/site/antora/modules/ROOT/pages/FlumeUserGuide.adoc
+++ b/src/site/antora/modules/ROOT/pages/FlumeUserGuide.adoc
@@ -1680,7 +1680,10 @@ This deserializer reads a Binary Large Object (BLOB) per
event, typically one BL
[NOTE]
====
-*This source is provided as a preview feature. It does not work on Windows.*
+*This source is provided as a preview feature.*
+
+This source tracks files by their inode number and therefore requires a file
system that provides the `unix` file attribute view.
+In particular, it does not work on Windows.
====
Watch the specified files, and tail them in nearly real-time once detected new
lines appended to the each files.
