This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new db732669a52 NIFI-16238: Checkpoint write-ahead logs when the journal
reaches a co… (#11575)
db732669a52 is described below
commit db732669a521b3df05b4ee1f2ef1cb1d37c78846
Author: Mark Payne <[email protected]>
AuthorDate: Fri Aug 21 05:48:07 2026 -0400
NIFI-16238: Checkpoint write-ahead logs when the journal reaches a co…
(#11575)
Journals could grow without bound between interval-based checkpoints. A
maximum journal size now triggers a checkpoint so storage is reclaimed even
when updates keep arriving.
---
.../java/org/apache/nifi/util/NiFiProperties.java | 16 +++
.../apache/nifi/wali/LengthDelimitedJournal.java | 17 +++
.../nifi/wali/SequentialAccessWriteAheadLog.java | 86 ++++++++++-
.../org/apache/nifi/wali/WriteAheadJournal.java | 9 ++
.../nifi/wali/TestLengthDelimitedJournal.java | 69 +++++++++
.../wali/TestSequentialAccessWriteAheadLog.java | 121 +++++++++++++++-
.../src/main/asciidoc/administration-guide.adoc | 11 +-
.../repository/WriteAheadFlowFileRepository.java | 7 +-
.../local/WriteAheadLocalStateProvider.java | 16 ++-
.../tasks/NiFiPropertiesDiagnosticTask.java | 1 +
.../TestWriteAheadFlowFileRepository.java | 67 +++++++++
.../local/TestWriteAheadLocalStateProvider.java | 51 +++++++
.../nifi-framework/nifi-resources/pom.xml | 1 +
.../src/main/resources/conf/nifi.properties | 1 +
.../src/main/resources/conf/state-management.xml | 5 +
.../processors/tests/system/AppendLocalState.java | 93 ++++++++++++
.../services/org.apache.nifi.processor.Processor | 1 +
.../repositories/WriteAheadJournalSizeIT.java | 157 +++++++++++++++++++++
.../default/state-management-bounded-journal.xml | 38 +++++
19 files changed, 753 insertions(+), 14 deletions(-)
diff --git
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index 2c42ca01859..ad89e7dd6b9 100644
---
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -102,6 +102,7 @@ public class NiFiProperties extends ApplicationProperties {
public static final String FLOWFILE_REPOSITORY_ALWAYS_SYNC =
"nifi.flowfile.repository.always.sync";
public static final String FLOWFILE_REPOSITORY_DIRECTORY =
"nifi.flowfile.repository.directory";
public static final String FLOWFILE_REPOSITORY_CHECKPOINT_INTERVAL =
"nifi.flowfile.repository.checkpoint.interval";
+ public static final String FLOWFILE_REPOSITORY_CHECKPOINT_MAX_JOURNAL_SIZE
= "nifi.flowfile.repository.checkpoint.max.journal.size";
public static final String FLOWFILE_SWAP_MANAGER_IMPLEMENTATION =
"nifi.swap.manager.implementation";
public static final String QUEUE_SWAP_THRESHOLD =
"nifi.queue.swap.threshold";
@@ -629,6 +630,21 @@ public class NiFiProperties extends ApplicationProperties {
return getProperty(FLOWFILE_REPOSITORY_CHECKPOINT_INTERVAL,
DEFAULT_FLOWFILE_CHECKPOINT_INTERVAL);
}
+ /**
+ * Returns the amount of storage that the FlowFile Repository's journal is
allowed to consume before the repository is checkpointed,
+ * expressed as a data size such as <code>100 MB</code>.
+ *
+ * @return the configured maximum journal size, or <code>null</code> if
the repository is to be checkpointed only on the configured interval
+ */
+ public String getFlowFileRepositoryMaxJournalSize() {
+ final String maxJournalSize =
getProperty(FLOWFILE_REPOSITORY_CHECKPOINT_MAX_JOURNAL_SIZE);
+ if (maxJournalSize == null || maxJournalSize.isBlank()) {
+ return null;
+ }
+
+ return maxJournalSize.trim();
+ }
+
/**
* @return the restore directory or null if not configured
*/
diff --git
a/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/LengthDelimitedJournal.java
b/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/LengthDelimitedJournal.java
index 71b66d3d228..7d955b974ed 100644
---
a/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/LengthDelimitedJournal.java
+++
b/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/LengthDelimitedJournal.java
@@ -48,6 +48,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
public class LengthDelimitedJournal<T> implements WriteAheadJournal<T> {
private static final Logger logger =
LoggerFactory.getLogger(LengthDelimitedJournal.class);
@@ -74,6 +75,8 @@ public class LengthDelimitedJournal<T> implements
WriteAheadJournal<T> {
private int transactionCount;
private boolean headerWritten = false;
+ private final AtomicLong bytesWritten = new AtomicLong(0L);
+
private volatile Throwable poisonCause = null;
private volatile boolean closed = false;
private final ByteBuffer transactionPreamble = ByteBuffer.allocate(12); //
guarded by synchronized block
@@ -172,6 +175,7 @@ public class LengthDelimitedJournal<T> implements
WriteAheadJournal<T> {
}
outStream.flush();
+ bytesWritten.addAndGet(outStream.size());
} catch (final Throwable t) {
poison(t);
@@ -299,6 +303,9 @@ public class LengthDelimitedJournal<T> implements
WriteAheadJournal<T> {
}
}
+ // The overflow file is fully written and synced at this point, so
its length is stable.
+ final long overflowFileLength = (overflowFile == null) ? 0L :
overflowFile.length();
+
final ByteArrayOutputStream baos =
bados.getByteArrayOutputStream();
final OutputStream out = getOutputStream();
@@ -318,6 +325,11 @@ public class LengthDelimitedJournal<T> implements
WriteAheadJournal<T> {
out.write(transactionPreamble.array());
baos.writeTo(out);
out.flush();
+
+ // A transaction consumes a single-byte marker, the
fixed-length preamble, and the serialized records. Storage for an
+ // overflow file is counted only now that the transaction
referencing it has been written, so that a failed update,
+ // whose overflow file is deleted, does not inflate the
count.
+ bytesWritten.addAndGet(1L + transactionPreamble.capacity()
+ baos.size() + overflowFileLength);
} catch (final Throwable t) {
// While the outer Throwable that wraps this "catch" will
call Poison, it is imperative that we call poison()
// before the synchronized block is excited. Otherwise,
another thread could potentially corrupt the journal before
@@ -612,6 +624,11 @@ public class LengthDelimitedJournal<T> implements
WriteAheadJournal<T> {
return true;
}
+ @Override
+ public long getBytesWritten() {
+ return bytesWritten.get();
+ }
+
@Override
public synchronized JournalSummary getSummary() {
if (transactionCount < 1) {
diff --git
a/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/SequentialAccessWriteAheadLog.java
b/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/SequentialAccessWriteAheadLog.java
index 2f4ff152051..dacce156e3f 100644
---
a/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/SequentialAccessWriteAheadLog.java
+++
b/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/SequentialAccessWriteAheadLog.java
@@ -36,6 +36,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BooleanSupplier;
import java.util.regex.Pattern;
/**
@@ -53,6 +54,14 @@ import java.util.regex.Pattern;
* that records are recovered correctly if two threads simultaneously update
the write-ahead log
* with updates for the same record.
* </p>
+ *
+ * <p>
+ * A maximum journal size may be provided. When it is, an update that causes
the amount of storage consumed by the
+ * journal to reach that size triggers a checkpoint, which rolls over to a new
journal and reclaims the storage that
+ * was consumed by the previous one. The size is a trigger rather than a
strict cap, because the update that crosses
+ * the threshold is always written in full before the journal is rolled over.
When no maximum journal size is provided,
+ * the journal grows until the owner of this repository chooses to checkpoint.
+ * </p>
*/
public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T> {
private static final int PARTITION_INDEX = 0;
@@ -65,6 +74,7 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
private final File journalsDirectory;
protected final SerDeFactory<T> serdeFactory;
private final SyncListener syncListener;
+ private final Long maxJournalBytes;
private final Set<String> recoveredSwapLocations = new HashSet<>();
private final ReadWriteLock journalRWLock = new ReentrantReadWriteLock();
@@ -88,6 +98,25 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
}
public SequentialAccessWriteAheadLog(final File storageDirectory, final
SerDeFactory<T> serdeFactory, final SyncListener syncListener) throws
IOException {
+ this(storageDirectory, serdeFactory, syncListener, null);
+ }
+
+ /**
+ * Creates a Write-Ahead Log that optionally checkpoints whenever its
journal reaches a given size.
+ *
+ * @param storageDirectory the directory in which the snapshot and
journals are stored
+ * @param serdeFactory the factory for creating the
serializer/deserializer to use for records
+ * @param syncListener the listener to notify when data is synchronized to
disk
+ * @param maxJournalBytes the amount of storage that the journal may
consume before an update triggers a checkpoint, or
+ * <code>null</code> if the journal is to grow until the caller
explicitly checkpoints. When provided, the value must be
+ * greater than 0.
+ * @throws IOException if unable to create or access the storage directory
+ */
+ public SequentialAccessWriteAheadLog(final File storageDirectory, final
SerDeFactory<T> serdeFactory, final SyncListener syncListener, final Long
maxJournalBytes) throws IOException {
+ if (maxJournalBytes != null && maxJournalBytes <= 0) {
+ throw new IllegalArgumentException("Maximum Journal Size must be
greater than 0 bytes but was " + maxJournalBytes);
+ }
+
if (!storageDirectory.exists() && !storageDirectory.mkdirs()) {
throw new IOException("Directory " + storageDirectory + " does not
exist and cannot be created");
}
@@ -109,6 +138,7 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
this.serdeFactory = serdeFactory;
this.syncListener = (syncListener == null) ?
SyncListener.NOP_SYNC_LISTENER : syncListener;
+ this.maxJournalBytes = maxJournalBytes;
}
@Override
@@ -117,6 +147,7 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
throw new IllegalStateException("Cannot update repository until
record recovery has been performed");
}
+ final boolean maxJournalSizeReached;
journalReadLock.lock();
try {
journal.update(records, recordLookup);
@@ -127,10 +158,25 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
}
snapshot.update(records);
+
+ maxJournalSizeReached = maxJournalBytes != null &&
journal.getBytesWritten() >= maxJournalBytes;
} finally {
journalReadLock.unlock();
}
+ // The checkpoint requires the write lock, so it must be performed
only after the read lock has been released. Because the
+ // journal may be rolled over by another thread in the meantime, the
checkpoint verifies that the journal is still large
+ // enough to warrant rolling over.
+ if (maxJournalSizeReached) {
+ logger.debug("Checkpointing Write-Ahead Log at {} because its
journal has reached the maximum size of {} bytes", storageDirectory,
maxJournalBytes);
+ try {
+ checkpointIfJournalExceedsLimit();
+ } catch (final Exception e) {
+ logger.error("Failed to checkpoint Write-Ahead Log at {} after
its journal reached the maximum size of {} bytes; the update was already
applied",
+ storageDirectory, maxJournalBytes, e);
+ }
+ }
+
return PARTITION_INDEX;
}
@@ -252,12 +298,41 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
}
private int checkpoint(final Set<String> swapLocations) throws IOException
{
+ return checkpoint(swapLocations, () -> true);
+ }
+
+ /**
+ * Checkpoints only if the journal still consumes at least the configured
maximum number of bytes. Because the update that triggered
+ * this call must release the journal read lock before a checkpoint can
acquire the write lock, another thread may have already
+ * rolled the journal over. In that case, no checkpoint is performed.
+ *
+ * @throws IOException if unable to write the snapshot or create the new
journal
+ */
+ private void checkpointIfJournalExceedsLimit() throws IOException {
+ checkpoint(null, () -> journal != null && journal.getBytesWritten() >=
maxJournalBytes);
+ }
+
+ /**
+ * Writes a snapshot of the current state of the repository and rolls over
to a new journal.
+ *
+ * @param swapLocations the swap locations to include in the snapshot, or
<code>null</code> to use the swap locations that are already known
+ * @param checkpointRequired evaluated while the journal write lock is
held; if it returns <code>false</code>, no checkpoint is performed. This allows
a checkpoint
+ * that was triggered by the size of a journal to be abandoned
if that journal has already been rolled over by another thread
+ * @return the number of records that are stored in the snapshot
+ * @throws IOException if unable to write the snapshot or create the new
journal
+ */
+ private int checkpoint(final Set<String> swapLocations, final
BooleanSupplier checkpointRequired) throws IOException {
final SnapshotCapture<T> snapshotCapture;
final long startNanos = System.nanoTime();
final File[] existingJournals;
+ long checkpointedJournalBytes = 0L;
journalWriteLock.lock();
try {
+ if (!checkpointRequired.getAsBoolean()) {
+ return snapshot.getRecordCount();
+ }
+
if (journal != null) {
final JournalSummary journalSummary = journal.getSummary();
if (journalSummary.getTransactionCount() == 0 &&
journal.isHealthy()) {
@@ -266,6 +341,8 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
return snapshot.getRecordCount();
}
+ checkpointedJournalBytes = journal.getBytesWritten();
+
try {
journal.fsync();
} catch (final Exception e) {
@@ -314,14 +391,14 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
snapshot.writeSnapshot(snapshotCapture);
for (final File existingJournal : existingJournals) {
- final WriteAheadJournal journal = new
LengthDelimitedJournal<>(existingJournal, serdeFactory, streamPool,
nextTransactionId);
- journal.dispose();
+ final WriteAheadJournal<T> existingWriteAheadJournal = new
LengthDelimitedJournal<>(existingJournal, serdeFactory, streamPool,
nextTransactionId);
+ existingWriteAheadJournal.dispose();
}
final long totalNanos = System.nanoTime() - startNanos;
final long millis = TimeUnit.NANOSECONDS.toMillis(totalNanos);
- logger.info("Checkpointed Write-Ahead Log with {} Records and {} Swap
Files in {} milliseconds (Stop-the-world time = {} milliseconds), max
Transaction ID {}",
- snapshotCapture.getRecords().size(),
snapshotCapture.getSwapLocations().size(), millis, stopTheWorldMillis,
snapshotCapture.getMaxTransactionId());
+ logger.info("Checkpointed Write-Ahead Log with {} Records and {} Swap
Files in {} milliseconds (Stop-the-world time = {} milliseconds), max
Transaction ID {}, reclaiming {} bytes of storage",
+ snapshotCapture.getRecords().size(),
snapshotCapture.getSwapLocations().size(), millis, stopTheWorldMillis,
snapshotCapture.getMaxTransactionId(), checkpointedJournalBytes);
return snapshotCapture.getRecords().size();
}
@@ -337,4 +414,5 @@ public class SequentialAccessWriteAheadLog<T> implements
WriteAheadRepository<T>
journalWriteLock.unlock();
}
}
+
}
diff --git
a/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/WriteAheadJournal.java
b/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/WriteAheadJournal.java
index a9c339ec984..7baaaba81d1 100644
---
a/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/WriteAheadJournal.java
+++
b/nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/WriteAheadJournal.java
@@ -48,6 +48,15 @@ public interface WriteAheadJournal<T> extends Closeable {
*/
JournalSummary getSummary();
+ /**
+ * Returns the amount of storage that has been consumed by this journal.
This includes the journal file itself as well as any
+ * overflow files that the journal references, so that the value reflects
the amount of disk space that must be reclaimed by
+ * checkpointing the journal.
+ *
+ * @return the number of bytes that have been written to storage for this
journal
+ */
+ long getBytesWritten();
+
/**
* @return <code>true</code> if the journal is healthy and can be written
to, <code>false</code> if either the journal has been closed or is poisoned
*/
diff --git
a/nifi-commons/nifi-write-ahead-log/src/test/java/org/apache/nifi/wali/TestLengthDelimitedJournal.java
b/nifi-commons/nifi-write-ahead-log/src/test/java/org/apache/nifi/wali/TestLengthDelimitedJournal.java
index 66c5713166a..9610210dc2b 100644
---
a/nifi-commons/nifi-write-ahead-log/src/test/java/org/apache/nifi/wali/TestLengthDelimitedJournal.java
+++
b/nifi-commons/nifi-write-ahead-log/src/test/java/org/apache/nifi/wali/TestLengthDelimitedJournal.java
@@ -61,6 +61,10 @@ public class TestLengthDelimitedJournal {
private ObjectPool<ByteArrayDataOutputStream> streamPool;
private static final int BUFFER_SIZE = 4096;
+ // Large enough that the reference written for an overflow file, which
contains the absolute path of that file, does not itself
+ // exceed the threshold and cause a second overflow file to be created.
+ private static final int OVERFLOW_THRESHOLD_BYTES = 1024;
+
@BeforeEach
public void setupJournal() throws IOException {
Files.deleteIfExists(journalFile.toPath());
@@ -77,6 +81,71 @@ public class TestLengthDelimitedJournal {
stream -> stream.getByteArrayOutputStream().reset());
}
+ @Test
+ public void testBytesWrittenMatchesJournalFileForInlineUpdates() throws
IOException {
+ try (final LengthDelimitedJournal<DummyRecord> journal = new
LengthDelimitedJournal<>(journalFile, serdeFactory, streamPool, 0L)) {
+ assertEquals(0L, journal.getBytesWritten());
+
+ journal.writeHeader();
+ assertEquals(journalFile.length(), journal.getBytesWritten());
+
+ for (int i = 0; i < 5; i++) {
+ final long bytesBeforeUpdate = journal.getBytesWritten();
+ journal.update(Collections.singleton(new
DummyRecord(String.valueOf(i), UpdateType.CREATE)), key -> null);
+
+ assertTrue(journal.getBytesWritten() > bytesBeforeUpdate);
+ assertEquals(journalFile.length(), journal.getBytesWritten());
+ }
+ }
+ }
+
+ @Test
+ public void testBytesWrittenIncludesOverflowFiles() throws IOException {
+ // A small in-heap serialization threshold causes the journal to write
the bulk of the update to an overflow file and to
+ // write only a reference to that file into the journal itself.
+ try (final LengthDelimitedJournal<DummyRecord> journal = new
LengthDelimitedJournal<>(journalFile, serdeFactory, streamPool, 0L,
OVERFLOW_THRESHOLD_BYTES)) {
+ journal.writeHeader();
+
+ final List<DummyRecord> records = new ArrayList<>();
+ for (int i = 0; i < 1_000; i++) {
+ records.add(new DummyRecord(String.valueOf(i),
UpdateType.CREATE));
+ }
+
+ journal.update(records, key -> null);
+
+ final Set<File> overflowFiles = serde.getExternalFileReferences();
+ assertEquals(1, overflowFiles.size());
+
+ long overflowBytes = 0L;
+ for (final File overflowFile : overflowFiles) {
+ overflowBytes += overflowFile.length();
+ }
+
+ assertTrue(overflowBytes > 0L);
+ assertEquals(journalFile.length() + overflowBytes,
journal.getBytesWritten());
+ }
+ }
+
+ @Test
+ public void testBytesWrittenUnchangedWhenUpdateFails() throws IOException {
+ try (final LengthDelimitedJournal<DummyRecord> journal = new
LengthDelimitedJournal<>(journalFile, serdeFactory, streamPool, 0L,
OVERFLOW_THRESHOLD_BYTES)) {
+ journal.writeHeader();
+ final long headerBytes = journal.getBytesWritten();
+
+ final List<DummyRecord> records = new ArrayList<>();
+ for (int i = 0; i < 1_000; i++) {
+ records.add(new DummyRecord(String.valueOf(i),
UpdateType.CREATE));
+ }
+
+ // Fail the update after enough records have been serialized that
an overflow file has been created. The overflow file is
+ // removed when the update fails, so its contents must not be
counted against the journal.
+ serde.setThrowIOEAfterNSerializeEdits(500);
+ assertThrows(IOException.class, () -> journal.update(records, key
-> null));
+
+ assertEquals(headerBytes, journal.getBytesWritten());
+ }
+ }
+
@Test
public void testHandlingOfTrailingNulBytes() throws IOException {
try (final LengthDelimitedJournal<DummyRecord> journal = new
LengthDelimitedJournal<>(journalFile, serdeFactory, streamPool, 0L)) {
diff --git
a/nifi-commons/nifi-write-ahead-log/src/test/java/org/apache/nifi/wali/TestSequentialAccessWriteAheadLog.java
b/nifi-commons/nifi-write-ahead-log/src/test/java/org/apache/nifi/wali/TestSequentialAccessWriteAheadLog.java
index 11317010019..3c65e806039 100644
---
a/nifi-commons/nifi-write-ahead-log/src/test/java/org/apache/nifi/wali/TestSequentialAccessWriteAheadLog.java
+++
b/nifi-commons/nifi-write-ahead-log/src/test/java/org/apache/nifi/wali/TestSequentialAccessWriteAheadLog.java
@@ -26,6 +26,7 @@ import org.wali.DummyRecord;
import org.wali.DummyRecordSerde;
import org.wali.SerDeFactory;
import org.wali.SingletonSerDeFactory;
+import org.wali.SyncListener;
import org.wali.UpdateType;
import org.wali.WriteAheadRepository;
@@ -35,6 +36,7 @@ import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.NumberFormat;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
@@ -42,11 +44,13 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -188,12 +192,9 @@ public class TestSequentialAccessWriteAheadLog {
}
private SequentialAccessWriteAheadLog<DummyRecord>
createRecoveryRepo(TestInfo testInfo) throws IOException {
- final File targetDir = new File("target");
- final File storageDir = new File(targetDir,
testInfo.getTestMethod().get().getName());
-
final DummyRecordSerde serde = new DummyRecordSerde();
final SerDeFactory<DummyRecord> serdeFactory = new
SingletonSerDeFactory<>(serde);
- final SequentialAccessWriteAheadLog<DummyRecord> repo = new
SequentialAccessWriteAheadLog<>(storageDir, serdeFactory);
+ final SequentialAccessWriteAheadLog<DummyRecord> repo = new
SequentialAccessWriteAheadLog<>(getStorageDirectory(testInfo), serdeFactory);
return repo;
}
@@ -203,13 +204,21 @@ public class TestSequentialAccessWriteAheadLog {
}
private SequentialAccessWriteAheadLog<DummyRecord> createWriteRepo(final
TestInfo testInfo, final DummyRecordSerde serde) throws IOException {
- final File targetDir = new File("target");
- final File storageDir = new File(targetDir,
testInfo.getTestMethod().get().getName());
+ return createWriteRepo(testInfo, serde, null);
+ }
+
+ private SequentialAccessWriteAheadLog<DummyRecord> createWriteRepo(final
TestInfo testInfo, final DummyRecordSerde serde, final Long maxJournalBytes)
throws IOException {
+ return createWriteRepo(testInfo, serde, maxJournalBytes,
SyncListener.NOP_SYNC_LISTENER);
+ }
+
+ private SequentialAccessWriteAheadLog<DummyRecord> createWriteRepo(final
TestInfo testInfo, final DummyRecordSerde serde, final Long maxJournalBytes,
+ final SyncListener syncListener) throws IOException {
+ final File storageDir = getStorageDirectory(testInfo);
deleteRecursively(storageDir);
assertTrue(storageDir.mkdirs());
final SerDeFactory<DummyRecord> serdeFactory = new
SingletonSerDeFactory<>(serde);
- final SequentialAccessWriteAheadLog<DummyRecord> repo = new
SequentialAccessWriteAheadLog<>(storageDir, serdeFactory);
+ final SequentialAccessWriteAheadLog<DummyRecord> repo = new
SequentialAccessWriteAheadLog<>(storageDir, serdeFactory, syncListener,
maxJournalBytes);
final Collection<DummyRecord> recovered = repo.recoverRecords();
assertNotNull(recovered);
@@ -218,6 +227,104 @@ public class TestSequentialAccessWriteAheadLog {
return repo;
}
+ private File getStorageDirectory(final TestInfo testInfo) {
+ return new File("target", testInfo.getTestMethod().get().getName());
+ }
+
+ private List<File> getJournalFiles(final TestInfo testInfo) {
+ final File[] journalFiles = new File(getStorageDirectory(testInfo),
"journals").listFiles(file -> file.getName().endsWith(".journal"));
+ assertNotNull(journalFiles);
+
+ return Arrays.asList(journalFiles);
+ }
+
+ @Test
+ public void testMaximumJournalSizeMustBePositive() {
+ final SerDeFactory<DummyRecord> serdeFactory = new
SingletonSerDeFactory<>(new DummyRecordSerde());
+ final File storageDir = new File("target",
"testMaximumJournalSizeMustBePositive");
+
+ assertThrows(IllegalArgumentException.class, () -> new
SequentialAccessWriteAheadLog<>(storageDir, serdeFactory,
SyncListener.NOP_SYNC_LISTENER, 0L));
+ assertThrows(IllegalArgumentException.class, () -> new
SequentialAccessWriteAheadLog<>(storageDir, serdeFactory,
SyncListener.NOP_SYNC_LISTENER, -1L));
+ }
+
+ @Test
+ public void testJournalRolledOverWhenMaximumJournalSizeReached(final
TestInfo testInfo) throws IOException {
+ // Determine how much storage the journal consumes for its header and
for a single-record update, so that the maximum journal
+ // size can be set to a value that is reached by a known number of
updates.
+ final SequentialAccessWriteAheadLog<DummyRecord> sizingRepo =
createWriteRepo(testInfo);
+ final File sizingJournalFile = getJournalFiles(testInfo).getFirst();
+ final long headerBytes = sizingJournalFile.length();
+ sizingRepo.update(Collections.singleton(new DummyRecord("0",
UpdateType.CREATE)), false);
+ final long updateBytes = sizingJournalFile.length() - headerBytes;
+ sizingRepo.shutdown();
+
+ final int updatesPerJournal = 3;
+ final SequentialAccessWriteAheadLog<DummyRecord> repo =
createWriteRepo(testInfo, new DummyRecordSerde(), headerBytes +
updatesPerJournal * updateBytes);
+ final File firstJournalFile = getJournalFiles(testInfo).getFirst();
+
+ for (int i = 0; i < updatesPerJournal - 1; i++) {
+ repo.update(Collections.singleton(new
DummyRecord(String.valueOf(i), UpdateType.CREATE)), false);
+ assertEquals(List.of(firstJournalFile), getJournalFiles(testInfo));
+ }
+
+ repo.update(Collections.singleton(new
DummyRecord(String.valueOf(updatesPerJournal - 1), UpdateType.CREATE)), false);
+
+ final List<File> rolledOverJournalFiles = getJournalFiles(testInfo);
+ assertEquals(1, rolledOverJournalFiles.size());
+
+ final File secondJournalFile = rolledOverJournalFiles.getFirst();
+ assertNotEquals(firstJournalFile, secondJournalFile);
+ assertFalse(firstJournalFile.exists());
+ assertEquals(headerBytes, secondJournalFile.length());
+
+ repo.shutdown();
+
+ final SequentialAccessWriteAheadLog<DummyRecord> recoveryRepo =
createRecoveryRepo(testInfo);
+ final Collection<DummyRecord> recovered =
recoveryRepo.recoverRecords();
+ assertEquals(updatesPerJournal, recovered.size());
+ recoveryRepo.shutdown();
+ }
+
+ @Test
+ public void testUpdateSucceedsWhenAutomaticCheckpointFails(final TestInfo
testInfo) throws IOException {
+ final SequentialAccessWriteAheadLog<DummyRecord> sizingRepo =
createWriteRepo(testInfo);
+ final File sizingJournalFile = getJournalFiles(testInfo).getFirst();
+ final long headerBytes = sizingJournalFile.length();
+ sizingRepo.update(Collections.singleton(new DummyRecord("0",
UpdateType.CREATE)), false);
+ final long updateBytes = sizingJournalFile.length() - headerBytes;
+ sizingRepo.shutdown();
+
+ final AtomicBoolean failCheckpoint = new AtomicBoolean(false);
+ final SyncListener failingCheckpointListener = new SyncListener() {
+ @Override
+ public void onSync(final int partitionIndex) {
+ }
+
+ @Override
+ public void onGlobalSync() {
+ if (failCheckpoint.get()) {
+ throw new RuntimeException("Injected checkpoint failure");
+ }
+ }
+ };
+
+ final int updatesPerJournal = 3;
+ final SequentialAccessWriteAheadLog<DummyRecord> repo =
createWriteRepo(testInfo, new DummyRecordSerde(), headerBytes +
updatesPerJournal * updateBytes, failingCheckpointListener);
+ failCheckpoint.set(true);
+
+ for (int i = 0; i < updatesPerJournal; i++) {
+ repo.update(Collections.singleton(new
DummyRecord(String.valueOf(i), UpdateType.CREATE)), false);
+ }
+
+ repo.shutdown();
+
+ failCheckpoint.set(false);
+ final SequentialAccessWriteAheadLog<DummyRecord> recoveryRepo =
createRecoveryRepo(testInfo);
+ final Collection<DummyRecord> recovered =
recoveryRepo.recoverRecords();
+ assertEquals(updatesPerJournal, recovered.size());
+ recoveryRepo.shutdown();
+ }
+
/**
* This test is designed to update the repository in several different
wants, testing CREATE, UPDATE, SWAP IN, SWAP OUT, and DELETE
* update types, as well as testing updates with single records and with
multiple records in a transaction. It also verifies that we
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc
b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index 5ba8082df22..0ffda8071ea 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -2287,7 +2287,15 @@ Otherwise, NiFi will fail to startup.
==== Local State Provider
By default, the Local State Provider is configured to be a
`WriteAheadLocalStateProvider` that persists the data to the
-`$NIFI_HOME/state/local` directory.
+`$NIFI_HOME/state/local` directory. It writes each state update to a journal
and periodically checkpoints, which writes a snapshot
+of all state and removes the journal that the snapshot now covers. The
`Checkpoint Interval` property controls how much time passes
+between checkpoints.
+
+Because the journal grows with every state update, a long checkpoint interval
combined with frequent state updates can consume a
+significant amount of disk space. The optional `Maximum Journal Size` property
bounds this growth by triggering a checkpoint as soon
+as the journal reaches the configured size, such as `100 MB`. The value is a
threshold that triggers a checkpoint rather than a strict
+limit on the size of the journal, because the update that reaches the
threshold is written in full before the checkpoint is performed.
+If the property is not specified, checkpoints occur only on the `Checkpoint
Interval`.
==== ZooKeeper Cluster State Provider
@@ -2983,6 +2991,7 @@ NOTE: Switching repository implementations should only be
done on an instance wi
|*Property*|*Description*
|`nifi.flowfile.repository.directory`*|The location of the FlowFile
Repository. The default value is `./flowfile_repository`.
|`nifi.flowfile.repository.checkpoint.interval`| The FlowFile Repository
checkpoint interval. The default value is `20 secs`.
+|`nifi.flowfile.repository.checkpoint.max.journal.size`| The amount of storage
that the repository's journal is allowed to consume before a checkpoint is
performed, such as `100 MB`. This bounds the amount of disk space that the
repository consumes between checkpoints, regardless of how many FlowFiles are
updated. The value is a threshold that triggers a checkpoint rather than a
strict limit on the size of the journal, because the update that reaches the
threshold is written in full befo [...]
|`nifi.flowfile.repository.always.sync`|If set to `true`, any change to the
repository will be synchronized to the disk, meaning that NiFi will ask the
operating system not to cache the information. This is very expensive and can
significantly reduce NiFi performance. However, if it is `false`, there could
be the potential for data loss if either there is a sudden power loss or the
operating system crashes. The default value is `false`.
|====
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java
index 96345a89cd5..f18d5b89159 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java
@@ -102,6 +102,7 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
private final int maxCharactersToCache;
private final long truncationThreshold;
private final boolean truncationEnabled;
+ private final Long maximumJournalBytes;
private volatile Collection<SerializedRepositoryRecord> recoveredRecords =
null;
private final Set<ResourceClaim> orphanedResourceClaims =
Collections.synchronizedSet(new HashSet<>());
@@ -150,6 +151,7 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
maxCharactersToCache = 0;
truncationThreshold = Long.MAX_VALUE;
truncationEnabled = false;
+ maximumJournalBytes = null;
}
public WriteAheadFlowFileRepository(final NiFiProperties nifiProperties) {
@@ -171,6 +173,9 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
checkpointDelayMillis =
FormatUtils.getTimeDuration(nifiProperties.getFlowFileRepositoryCheckpointInterval(),
TimeUnit.MILLISECONDS);
+ final String maximumJournalSize =
nifiProperties.getFlowFileRepositoryMaxJournalSize();
+ maximumJournalBytes = (maximumJournalSize == null) ? null :
DataUnit.parseDataSize(maximumJournalSize, DataUnit.B).longValue();
+
checkpointExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
final Thread t = Executors.defaultThreadFactory().newThread(r);
t.setName("Checkpoint FlowFile Repository");
@@ -202,7 +207,7 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
// delete backup. On restore, if no files exist in partition's
directory, would have to check backup directory
this.serdeFactory = serdeFactory;
- wal = new
SequentialAccessWriteAheadLog<>(flowFileRepositoryPaths.get(0), serdeFactory,
this);
+ wal = new
SequentialAccessWriteAheadLog<>(flowFileRepositoryPaths.get(0), serdeFactory,
this, maximumJournalBytes);
logger.info("Initialized FlowFile Repository");
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/local/WriteAheadLocalStateProvider.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/local/WriteAheadLocalStateProvider.java
index 1ba08c83c5a..1eb68ac2595 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/local/WriteAheadLocalStateProvider.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/local/WriteAheadLocalStateProvider.java
@@ -25,12 +25,14 @@ import org.apache.nifi.controller.state.StandardStateMap;
import org.apache.nifi.controller.state.StateMapSerDe;
import org.apache.nifi.controller.state.StateMapUpdate;
import org.apache.nifi.controller.state.providers.AbstractStateProvider;
+import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.wali.SequentialAccessWriteAheadLog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wali.SerDe;
import org.wali.SerDeFactory;
+import org.wali.SyncListener;
import org.wali.UpdateType;
import org.wali.WriteAheadRepository;
@@ -99,6 +101,13 @@ public class WriteAheadLocalStateProvider extends
AbstractStateProvider {
.required(true)
.build();
+ static final PropertyDescriptor MAXIMUM_JOURNAL_SIZE = new
PropertyDescriptor.Builder()
+ .name("Maximum Journal Size")
+ .description("The amount of data that may be written to the
write-ahead log's journal before a checkpoint is performed. If not specified,
checkpoints occur only on the Checkpoint Interval.")
+ .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+ .required(false)
+ .build();
+
private WriteAheadRepository<StateMapUpdate> writeAheadLog;
private AtomicLong versionGenerator;
@@ -130,7 +139,11 @@ public class WriteAheadLocalStateProvider extends
AbstractStateProvider {
}
versionGenerator = new AtomicLong(-1L);
- writeAheadLog = new SequentialAccessWriteAheadLog<>(basePath, new
SerdeFactory(serde));
+
+ final Double maximumJournalSize =
context.getProperty(MAXIMUM_JOURNAL_SIZE).asDataSize(DataUnit.B);
+ final Long maximumJournalBytes = (maximumJournalSize == null) ? null :
maximumJournalSize.longValue();
+
+ writeAheadLog = new SequentialAccessWriteAheadLog<>(basePath, new
SerdeFactory(serde), SyncListener.NOP_SYNC_LISTENER, maximumJournalBytes);
final Collection<StateMapUpdate> updates =
writeAheadLog.recoverRecords();
long maxRecordVersion = EMPTY_VERSION;
@@ -165,6 +178,7 @@ public class WriteAheadLocalStateProvider extends
AbstractStateProvider {
properties.add(ALWAYS_SYNC);
properties.add(CHECKPOINT_INTERVAL);
properties.add(NUM_PARTITIONS);
+ properties.add(MAXIMUM_JOURNAL_SIZE);
return properties;
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/diagnostics/bootstrap/tasks/NiFiPropertiesDiagnosticTask.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/diagnostics/bootstrap/tasks/NiFiPropertiesDiagnosticTask.java
index 27b0cb63805..6cfb081dc59 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/diagnostics/bootstrap/tasks/NiFiPropertiesDiagnosticTask.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/diagnostics/bootstrap/tasks/NiFiPropertiesDiagnosticTask.java
@@ -44,6 +44,7 @@ public class NiFiPropertiesDiagnosticTask implements
DiagnosticTask {
"nifi.content.repository.archive.max.retention.period",
"nifi.content.repository.archive.max.usage.percentage",
"nifi.flowfile.repository.checkpoint.interval",
+ "nifi.flowfile.repository.checkpoint.max.journal.size",
"nifi.flowfile.repository.always.sync",
"nifi.components.status.snapshot.frequency",
"nifi.bored.yield.duration",
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
index 25062eb8583..7033e93dc52 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
@@ -64,6 +64,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -76,6 +77,7 @@ import java.util.function.Predicate;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -627,6 +629,71 @@ public class TestWriteAheadFlowFileRepository {
assertEquals(claim2.getResourceClaim(), swappedOutClaims.get(0));
}
+ /**
+ * Verifies that configuring a maximum journal size causes the repository
to checkpoint as soon as its journal reaches that size, even
+ * though the checkpoint interval has not elapsed, and that the FlowFiles
written before the checkpoint remain recoverable.
+ */
+ @Test
+ public void testJournalRolledOverWhenMaximumJournalSizeReached() throws
IOException {
+ final NiFiProperties boundedJournalProperties =
NiFiProperties.createBasicNiFiProperties(
+
TestWriteAheadFlowFileRepository.class.getResource("/conf/nifi.properties").getFile(),
+
Map.of(NiFiProperties.FLOWFILE_REPOSITORY_CHECKPOINT_MAX_JOURNAL_SIZE, "8 KB"));
+
+ final TestQueueProvider queueProvider = new TestQueueProvider();
+ final FlowFileQueue queue = createMockQueue(queueProvider);
+ final int flowFileCount = 100;
+
+ try (final WriteAheadFlowFileRepository repo = new
WriteAheadFlowFileRepository(boundedJournalProperties)) {
+ repo.initialize(new StandardResourceClaimManager());
+ repo.loadFlowFiles(queueProvider);
+
+ createFlowFiles(repo, queue, flowFileCount);
+
+ final List<File> journalFiles = getJournalFiles();
+ assertEquals(1, journalFiles.size());
+ assertNotEquals("0.journal", journalFiles.getFirst().getName());
+ }
+
+ try (final WriteAheadFlowFileRepository recoveredRepo = new
WriteAheadFlowFileRepository(boundedJournalProperties)) {
+ recoveredRepo.initialize(new StandardResourceClaimManager());
+ assertEquals(flowFileCount,
recoveredRepo.loadFlowFiles(queueProvider));
+ }
+ }
+
+ private FlowFileQueue createMockQueue(final TestQueueProvider
queueProvider) {
+ final Connection connection = Mockito.mock(Connection.class);
+ when(connection.getIdentifier()).thenReturn("1234");
+
+ final FlowFileQueue queue = Mockito.mock(FlowFileQueue.class);
+ when(queue.getIdentifier()).thenReturn("1234");
+ when(connection.getFlowFileQueue()).thenReturn(queue);
+
+ queueProvider.addConnection(connection);
+
+ return queue;
+ }
+
+ private void createFlowFiles(final WriteAheadFlowFileRepository
repository, final FlowFileQueue queue, final int flowFileCount) throws
IOException {
+ for (int i = 0; i < flowFileCount; i++) {
+ final FlowFileRecord flowFile = new
StandardFlowFileRecord.Builder()
+ .id(i + 1L)
+ .addAttribute("uuid", UUID.randomUUID().toString())
+ .build();
+
+ final StandardRepositoryRecord record = new
StandardRepositoryRecord(queue);
+ record.setWorking(flowFile, false);
+ record.setDestination(queue);
+ repository.updateRepository(List.of(record));
+ }
+ }
+
+ private List<File> getJournalFiles() {
+ final File[] journalFiles = new
File("target/test-repo/journals").listFiles(file ->
file.getName().endsWith(".journal"));
+ assertNotNull(journalFiles);
+
+ return Arrays.asList(journalFiles);
+ }
+
@Test
public void testRestartWithOneRecord() throws IOException {
final Path path = Paths.get("target/test-repo");
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java
index af3c32e7933..fb11a2eef0f 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/state/providers/local/TestWriteAheadLocalStateProvider.java
@@ -32,15 +32,20 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.wali.WriteAheadRepository;
+import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import javax.net.ssl.SSLContext;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
public class TestWriteAheadLocalStateProvider extends
AbstractTestStateProvider {
@TempDir
@@ -104,13 +109,59 @@ public class TestWriteAheadLocalStateProvider extends
AbstractTestStateProvider
}
}
+ /**
+ * Verifies that the Maximum Journal Size property causes the Provider to
checkpoint as soon as its journal reaches the configured size,
+ * even though the Checkpoint Interval has not elapsed, and that the state
written before the checkpoint remains recoverable.
+ */
+ @Test
+ public void testJournalRolledOverWhenMaximumJournalSizeReached() throws
IOException {
+ final Path storageDirectory =
temporaryDirectory.resolve("bounded-journal");
+ final String testComponentId = "test-bounded-journal-component";
+
+ final WriteAheadLocalStateProvider boundedProvider =
initializeProvider(storageDirectory.toString(), "4 KB");
+ try {
+ updateStateRepeatedly(boundedProvider, testComponentId, 100);
+
+ final List<File> journalFiles = getJournalFiles(storageDirectory);
+ assertEquals(1, journalFiles.size());
+ assertNotEquals("0.journal", journalFiles.getFirst().getName());
+ } finally {
+ boundedProvider.shutdown();
+ }
+
+ final WriteAheadLocalStateProvider recoveredProvider =
initializeProvider(storageDirectory.toString(), "4 KB");
+ try {
+ assertEquals(Collections.singletonMap("iteration", "99"),
recoveredProvider.getState(testComponentId).toMap());
+ } finally {
+ recoveredProvider.shutdown();
+ }
+ }
+
+ private void updateStateRepeatedly(final StateProvider stateProvider,
final String stateComponentId, final int iterations) throws IOException {
+ for (int i = 0; i < iterations; i++) {
+ stateProvider.setState(Collections.singletonMap("iteration",
String.valueOf(i)), stateComponentId);
+ }
+ }
+
+ private List<File> getJournalFiles(final Path storageDirectory) {
+ final File[] journalFiles =
storageDirectory.resolve("journals").toFile().listFiles(file ->
file.getName().endsWith(".journal"));
+ assertNotNull(journalFiles);
+
+ return Arrays.asList(journalFiles);
+ }
+
private WriteAheadLocalStateProvider initializeProvider(final String
storageDirectory) throws IOException {
+ return initializeProvider(storageDirectory, null);
+ }
+
+ private WriteAheadLocalStateProvider initializeProvider(final String
storageDirectory, final String maximumJournalSize) throws IOException {
final WriteAheadLocalStateProvider newProvider = new
WriteAheadLocalStateProvider();
final Map<PropertyDescriptor, PropertyValue> properties = new
HashMap<>();
properties.put(WriteAheadLocalStateProvider.PATH, new
StandardPropertyValue(storageDirectory, null, ParameterLookup.EMPTY));
properties.put(WriteAheadLocalStateProvider.ALWAYS_SYNC, new
StandardPropertyValue("false", null, ParameterLookup.EMPTY));
properties.put(WriteAheadLocalStateProvider.CHECKPOINT_INTERVAL, new
StandardPropertyValue("2 mins", null, ParameterLookup.EMPTY));
properties.put(WriteAheadLocalStateProvider.NUM_PARTITIONS, new
StandardPropertyValue("16", null, ParameterLookup.EMPTY));
+ properties.put(WriteAheadLocalStateProvider.MAXIMUM_JOURNAL_SIZE, new
StandardPropertyValue(maximumJournalSize, null, ParameterLookup.EMPTY));
newProvider.initialize(new StateProviderInitializationContext() {
@Override
diff --git a/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
b/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
index fe8268e7500..819cd4ef672 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
+++ b/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
@@ -58,6 +58,7 @@
<nifi.flowfile.repository.wal.implementation>org.apache.nifi.wali.SequentialAccessWriteAheadLog</nifi.flowfile.repository.wal.implementation>
<nifi.flowfile.repository.directory>./flowfile_repository</nifi.flowfile.repository.directory>
<nifi.flowfile.repository.checkpoint.interval>20
secs</nifi.flowfile.repository.checkpoint.interval>
+ <nifi.flowfile.repository.checkpoint.max.journal.size />
<nifi.flowfile.repository.always.sync>false</nifi.flowfile.repository.always.sync>
<nifi.flowfile.repository.retain.orphaned.flowfiles>true</nifi.flowfile.repository.retain.orphaned.flowfiles>
<nifi.swap.manager.implementation>org.apache.nifi.controller.FileSystemSwapManager</nifi.swap.manager.implementation>
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
index fba5d1be311..772e4792c1e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
+++
b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
@@ -73,6 +73,7 @@
nifi.flowfile.repository.implementation=${nifi.flowfile.repository.implementatio
nifi.flowfile.repository.wal.implementation=${nifi.flowfile.repository.wal.implementation}
nifi.flowfile.repository.directory=${nifi.flowfile.repository.directory}
nifi.flowfile.repository.checkpoint.interval=${nifi.flowfile.repository.checkpoint.interval}
+nifi.flowfile.repository.checkpoint.max.journal.size=${nifi.flowfile.repository.checkpoint.max.journal.size}
nifi.flowfile.repository.always.sync=${nifi.flowfile.repository.always.sync}
nifi.flowfile.repository.retain.orphaned.flowfiles=${nifi.flowfile.repository.retain.orphaned.flowfiles}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml
b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml
index 1fd40ea6a3d..74c7b413ad3 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml
+++
b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/state-management.xml
@@ -30,6 +30,11 @@
operating system crashes. The default value is false.
Partitions - The number of partitions.
Checkpoint Interval - The amount of time between checkpoints.
+
+ This Provider also supports the following optional property:
+
+ Maximum Journal Size - The amount of data that may be written to the
write-ahead log's journal, such as "100 MB", before a checkpoint is performed.
If not specified, checkpoints occur
+ only on the Checkpoint Interval.
-->
<local-provider>
<id>local-provider</id>
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/AppendLocalState.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/AppendLocalState.java
new file mode 100644
index 00000000000..7265eea1e3f
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/AppendLocalState.java
@@ -0,0 +1,93 @@
+/*
+ * 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.nifi.processors.tests.system;
+
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.configuration.DefaultSchedule;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.DataUnit;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@DefaultSchedule(period = "10 mins")
+@Stateful(scopes = Scope.LOCAL, description = "Stores the number of the most
recent state update along with a fixed-size value")
+public class AppendLocalState extends AbstractProcessor {
+ static final PropertyDescriptor UPDATE_COUNT = new
PropertyDescriptor.Builder()
+ .name("Update Count")
+ .description("The number of state updates to perform each time the
Processor is triggered.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("500")
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor VALUE_SIZE = new
PropertyDescriptor.Builder()
+ .name("Value Size")
+ .description("The size of the value that is stored in state alongside
the update number.")
+ .addValidator(StandardValidators.createDataSizeBoundsValidator(1, 1024
* 1024))
+ .defaultValue("64 B")
+ .required(true)
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .build();
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return List.of(UPDATE_COUNT, VALUE_SIZE);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return Set.of(REL_SUCCESS);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ flowFile = session.create();
+ }
+
+ final int updateCount = context.getProperty(UPDATE_COUNT).asInteger();
+ final int valueSize =
context.getProperty(VALUE_SIZE).asDataSize(DataUnit.B).intValue();
+ final String value = "A".repeat(valueSize);
+
+ final StateManager stateManager = context.getStateManager();
+ for (int i = 0; i < updateCount; i++) {
+ try {
+ stateManager.setState(Map.of("update", String.valueOf(i),
"value", value), Scope.LOCAL);
+ } catch (final IOException e) {
+ throw new ProcessException("Failed to update local state", e);
+ }
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index b3167115e44..2d0e1afa8ab 100644
---
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -13,6 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+org.apache.nifi.processors.tests.system.AppendLocalState
org.apache.nifi.processors.tests.system.AssetReadingProcessor
org.apache.nifi.processors.tests.system.BacklogReportingTestProcessor
org.apache.nifi.processors.tests.system.ClassloaderIsolationWithServiceProperty
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/WriteAheadJournalSizeIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/WriteAheadJournalSizeIT.java
new file mode 100644
index 00000000000..556e7cdc9ef
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/repositories/WriteAheadJournalSizeIT.java
@@ -0,0 +1,157 @@
+/*
+ * 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.nifi.tests.system.repositories;
+
+import org.apache.nifi.tests.system.NiFiInstance;
+import org.apache.nifi.tests.system.NiFiSystemIT;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.StateMapDTO;
+import org.apache.nifi.web.api.entity.ConnectionEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies that the FlowFile Repository and the local State Provider are
checkpointed when their write-ahead log's journal reaches the
+ * configured maximum size. Both are configured with a checkpoint interval
that cannot elapse while the test is running, so a checkpoint
+ * that the test observes can only have been triggered by the size of the
journal.
+ */
+public class WriteAheadJournalSizeIT extends NiFiSystemIT {
+ private static final String MAXIMUM_JOURNAL_SIZE = "20 KB";
+ private static final long MAXIMUM_JOURNAL_BYTES = 20 * 1024L;
+
+ private static final int FLOWFILE_COUNT = 1000;
+ private static final int STATE_UPDATE_COUNT = 500;
+ private static final int STATE_VALUE_SIZE = 64;
+
+ @Override
+ protected Map<String, String> getNifiPropertiesOverrides() {
+ return Map.of("nifi.flowfile.repository.checkpoint.interval", "1 hour",
+ "nifi.flowfile.repository.checkpoint.max.journal.size",
MAXIMUM_JOURNAL_SIZE,
+ "nifi.state.management.configuration.file",
"conf/state-management-bounded-journal.xml");
+ }
+
+ @Test
+ public void
testFlowFileRepositoryCheckpointedWhenMaximumJournalSizeReached() throws
NiFiClientException, IOException, InterruptedException {
+ final File flowFileRepositoryDirectory = new
File(getNiFiInstance().getInstanceDirectory(), "flowfile_repository");
+ final long initialTransactionId =
getMaximumJournalTransactionId(flowFileRepositoryDirectory);
+
+ final ProcessorEntity generate =
getClientUtil().createProcessor("GenerateFlowFile");
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile");
+ getClientUtil().updateProcessorProperties(generate, Map.of("Batch
Size", "1", "File Size", "0 B", "Max FlowFiles",
String.valueOf(FLOWFILE_COUNT)));
+ getClientUtil().updateProcessorSchedulingPeriod(generate, "0 sec");
+
+ final ConnectionEntity connection =
getClientUtil().createConnection(generate, terminate, "success");
+
+ // Each invocation of the Processor creates a single FlowFile and
commits its session, so each invocation results in its own small
+ // update to the FlowFile Repository.
+ getClientUtil().startProcessor(generate);
+ waitForQueueCount(connection.getId(), FLOWFILE_COUNT);
+ getClientUtil().stopProcessor(generate);
+ getClientUtil().waitForStoppedProcessor(generate.getId());
+
+ assertJournalCheckpointed(flowFileRepositoryDirectory,
initialTransactionId);
+
+ // The FlowFiles that were written before and after the checkpoint
must all still be accounted for.
+ assertEquals(FLOWFILE_COUNT,
getConnectionQueueSize(connection.getId()));
+ }
+
+ @Test
+ public void
testLocalStateProviderCheckpointedWhenMaximumJournalSizeReached() throws
NiFiClientException, IOException, InterruptedException {
+ final File localStateDirectory = new
File(getNiFiInstance().getInstanceDirectory(), "state/local");
+ final long initialTransactionId =
getMaximumJournalTransactionId(localStateDirectory);
+
+ final ProcessorEntity appendState =
getClientUtil().createProcessor("AppendLocalState");
+ getClientUtil().updateProcessorProperties(appendState, Map.of("Update
Count", String.valueOf(STATE_UPDATE_COUNT), "Value Size", STATE_VALUE_SIZE + "
B"));
+ getClientUtil().setAutoTerminatedRelationships(appendState, "success");
+
+ // Each of the Processor's state updates is written to the State
Provider individually, so the Processor produces many small updates.
+ getClientUtil().runProcessorOnce(appendState);
+ getClientUtil().waitForStoppedProcessor(appendState.getId());
+
+ assertJournalCheckpointed(localStateDirectory, initialTransactionId);
+
+ final Map<String, String> expectedState = Map.of("update",
String.valueOf(STATE_UPDATE_COUNT - 1), "value", "A".repeat(STATE_VALUE_SIZE));
+ assertEquals(expectedState, getLocalState(appendState.getId()));
+
+ // Restarting proves that the state that was checkpointed and the
state that was written to the new journal afterward are both recovered.
+ final NiFiInstance nifiInstance = getNiFiInstance();
+ nifiInstance.stop();
+ nifiInstance.start(true);
+ setupClient();
+
+ assertEquals(expectedState, getLocalState(appendState.getId()));
+ }
+
+ /**
+ * Waits until the write-ahead log in the given directory has rolled over
to a new journal and then asserts that the journals that were
+ * checkpointed have been removed and that the storage consumed by the
remaining journal is within the configured maximum. A journal is
+ * named for the identifier of the first transaction that it holds, so a
larger transaction identifier means a newer journal.
+ *
+ * @param storageDirectory the storage directory of the write-ahead log
+ * @param initialTransactionId the largest journal transaction identifier
that existed before the test performed any updates
+ */
+ private void assertJournalCheckpointed(final File storageDirectory, final
long initialTransactionId) throws InterruptedException {
+ waitFor(() -> getMaximumJournalTransactionId(storageDirectory) >
initialTransactionId);
+
+ final List<File> journalFiles = listJournalFiles(storageDirectory);
+ assertEquals(1, journalFiles.size(), "Expected the journals that were
checkpointed to be removed but found " + journalFiles);
+
+ long journalBytes = 0L;
+ for (final File journalFile : journalFiles) {
+ journalBytes += journalFile.length();
+ }
+
+ assertTrue(journalBytes <= MAXIMUM_JOURNAL_BYTES, "Expected the
journal storage to be bounded by " + MAXIMUM_JOURNAL_BYTES + " bytes but it
consumed " + journalBytes + " bytes");
+ }
+
+ private long getMaximumJournalTransactionId(final File storageDirectory) {
+ long maximumTransactionId = -1L;
+
+ for (final File journalFile : listJournalFiles(storageDirectory)) {
+ final String filename = journalFile.getName();
+ final long transactionId = Long.parseLong(filename.substring(0,
filename.indexOf(".")));
+ maximumTransactionId = Math.max(maximumTransactionId,
transactionId);
+ }
+
+ return maximumTransactionId;
+ }
+
+ private List<File> listJournalFiles(final File storageDirectory) {
+ final File[] journalFiles = new File(storageDirectory,
"journals").listFiles(file -> file.getName().endsWith(".journal"));
+ return (journalFiles == null) ? List.of() : List.of(journalFiles);
+ }
+
+ private Map<String, String> getLocalState(final String processorId) throws
NiFiClientException, IOException {
+ final StateMapDTO localState =
getNifiClient().getProcessorClient().getProcessorState(processorId).getComponentState().getLocalState();
+ assertNotNull(localState);
+
+ final Map<String, String> state = new HashMap<>();
+ localState.getState().forEach(entry -> state.put(entry.getKey(),
entry.getValue()));
+
+ return state;
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/state-management-bounded-journal.xml
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/state-management-bounded-journal.xml
new file mode 100644
index 00000000000..a332b22dc8e
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/state-management-bounded-journal.xml
@@ -0,0 +1,38 @@
+<!--
+ 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.
+-->
+<!--
+ Configures the local State Provider to checkpoint whenever its journal
reaches 20 KB. The Checkpoint Interval is long enough that it
+ cannot elapse during a test, so any checkpoint that a test observes must
have been triggered by the Maximum Journal Size. A test selects
+ this configuration by overriding the
nifi.state.management.configuration.file property.
+-->
+<stateManagement>
+ <local-provider>
+ <id>local-provider</id>
+
<class>org.apache.nifi.controller.state.providers.local.WriteAheadLocalStateProvider</class>
+ <property name="Directory">./state/local</property>
+ <property name="Always Sync">false</property>
+ <property name="Partitions">16</property>
+ <property name="Checkpoint Interval">1 hour</property>
+ <property name="Maximum Journal Size">20 KB</property>
+ </local-provider>
+ <cluster-provider>
+ <id>zk-provider</id>
+
<class>org.apache.nifi.controller.state.providers.zookeeper.ZooKeeperStateProvider</class>
+ <property name="Connect String">localhost:62181</property>
+ <property name="Root Node">/nifi-integration-test</property>
+ <property name="Session Timeout">30 seconds</property>
+ <property name="Access Control">Open</property>
+ </cluster-provider>
+</stateManagement>
