This is an automated email from the ASF dual-hosted git repository.
otto 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 c06bb97e85 NIFI-11833 - remove deprecated classes in nifi-commons
(#7503)
c06bb97e85 is described below
commit c06bb97e85f1a592a85c529116739d9a86540dcd
Author: Pierre Villard <[email protected]>
AuthorDate: Thu Jul 20 17:58:51 2023 +0200
NIFI-11833 - remove deprecated classes in nifi-commons (#7503)
Signed-off-by: Otto Fowler <[email protected]>
---
.../apache/nifi/stream/io/BufferedInputStream.java | 34 -
.../nifi/stream/io/BufferedOutputStream.java | 34 -
.../nifi/stream/io/ByteArrayInputStream.java | 32 -
.../nifi/stream/io/ByteArrayOutputStream.java | 31 -
.../apache/nifi/stream/io/DataOutputStream.java | 30 -
.../java/org/wali/MinimalLockingWriteAheadLog.java | 1192 --------------------
.../org/wali/TestMinimalLockingWriteAheadLog.java | 977 ----------------
.../repository/WriteAheadFlowFileRepository.java | 154 +--
.../local/WriteAheadLocalStateProvider.java | 36 +-
.../TestWriteAheadFlowFileRepository.java | 4 +-
.../http/StandardHttpFlowFileServerProtocol.java | 4 +-
.../cache/server/map/PersistentMapCache.java | 36 +-
.../cache/server/set/PersistentSetCache.java | 35 +-
.../cache/server/map/TestPersistentMapCache.java | 41 -
14 files changed, 108 insertions(+), 2532 deletions(-)
diff --git
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/BufferedInputStream.java
b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/BufferedInputStream.java
deleted file mode 100644
index 8dba8999e1..0000000000
---
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/BufferedInputStream.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.stream.io;
-
-import java.io.InputStream;
-
-/**
- * @deprecated use java.io.BufferedInputStream instead
- */
-@Deprecated
-public class BufferedInputStream extends java.io.BufferedInputStream {
-
- public BufferedInputStream(InputStream in) {
- super(in);
- }
-
- public BufferedInputStream(InputStream in, int bufferSize) {
- super(in, bufferSize);
- }
-}
diff --git
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/BufferedOutputStream.java
b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/BufferedOutputStream.java
deleted file mode 100644
index 8c28eca682..0000000000
---
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/BufferedOutputStream.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.stream.io;
-
-import java.io.OutputStream;
-
-/**
- * @deprecated use java.io.BufferedOutputStream instead
- */
-@Deprecated
-public class BufferedOutputStream extends java.io.BufferedOutputStream {
-
- public BufferedOutputStream(OutputStream out) {
- super(out);
- }
-
- public BufferedOutputStream(OutputStream out, int bufferSize) {
- super(out, bufferSize);
- }
-}
diff --git
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/ByteArrayInputStream.java
b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/ByteArrayInputStream.java
deleted file mode 100644
index fca5894819..0000000000
---
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/ByteArrayInputStream.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.stream.io;
-
-/**
- * @deprecated use java.io.ByteArrayInputStream instead
- */
-@Deprecated
-public class ByteArrayInputStream extends java.io.ByteArrayInputStream {
-
- public ByteArrayInputStream(byte[] buffer) {
- super(buffer);
- }
-
- public ByteArrayInputStream(byte[] buffer, int offset, int length) {
- super(buffer, offset, length);
- }
-}
diff --git
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/ByteArrayOutputStream.java
b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/ByteArrayOutputStream.java
deleted file mode 100644
index bdc61aba9a..0000000000
---
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/ByteArrayOutputStream.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.stream.io;
-
-/**
- * @deprecated use java.io.ByteArrayOutputStream instead
- */
-@Deprecated
-public class ByteArrayOutputStream extends java.io.ByteArrayOutputStream {
- public ByteArrayOutputStream() {
- super();
- }
-
- public ByteArrayOutputStream(int initialBufferSize) {
- super(initialBufferSize);
- }
-}
diff --git
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/DataOutputStream.java
b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/DataOutputStream.java
deleted file mode 100644
index 02b8f1ebcf..0000000000
---
a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/DataOutputStream.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.stream.io;
-
-import java.io.OutputStream;
-
-/**
- * @deprecated use java.io.DataOutputStream instead
- */
-@Deprecated
-public class DataOutputStream extends java.io.DataOutputStream {
-
- public DataOutputStream(OutputStream out) {
- super(out);
- }
-}
diff --git
a/nifi-commons/nifi-write-ahead-log/src/main/java/org/wali/MinimalLockingWriteAheadLog.java
b/nifi-commons/nifi-write-ahead-log/src/main/java/org/wali/MinimalLockingWriteAheadLog.java
deleted file mode 100644
index 8db2d87885..0000000000
---
a/nifi-commons/nifi-write-ahead-log/src/main/java/org/wali/MinimalLockingWriteAheadLog.java
+++ /dev/null
@@ -1,1192 +0,0 @@
-/*
- * 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.wali;
-
-import static java.util.Objects.requireNonNull;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.channels.FileChannel;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.StandardOpenOption;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.SortedSet;
-import java.util.TreeMap;
-import java.util.TreeSet;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.regex.Pattern;
-
-import org.apache.nifi.wali.SequentialAccessWriteAheadLog;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * <p>
- * This implementation provides as little Locking as possible in order to
- * provide the highest throughput possible. However, this implementation is
ONLY
- * appropriate if it can be guaranteed that only a single thread will ever
issue
- * updates for a given Record at any one time.
- * </p>
- *
- * @param <T> type of record this WAL is for
- *
- * @deprecated This implementation is now deprecated in favor of {@link
SequentialAccessWriteAheadLog}.
- * This implementation, when given more than 1 partition, can have
issues recovering after a sudden loss
- * of power or an operating system crash.
- */
-@Deprecated
-public final class MinimalLockingWriteAheadLog<T> implements
WriteAheadRepository<T> {
-
- private final Path basePath;
- private final Path partialPath;
- private final Path snapshotPath;
-
- private final SerDeFactory<T> serdeFactory;
- private final SyncListener syncListener;
- private final FileChannel lockChannel;
- private final AtomicLong transactionIdGenerator = new AtomicLong(0L);
-
- private final Partition<T>[] partitions;
- private final AtomicLong partitionIndex = new AtomicLong(0L);
- private final ConcurrentMap<Object, T> recordMap = new
ConcurrentHashMap<>();
- private final Map<Object, T> unmodifiableRecordMap =
Collections.unmodifiableMap(recordMap);
- private final Set<String> externalLocations = new CopyOnWriteArraySet<>();
-
- private final Set<String> recoveredExternalLocations = new
CopyOnWriteArraySet<>();
-
- private final AtomicInteger numberBlackListedPartitions = new
AtomicInteger(0);
-
- private static final Logger logger =
LoggerFactory.getLogger(MinimalLockingWriteAheadLog.class);
-
- private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
- private final Lock readLock = rwLock.readLock(); // required to update a
partition
- private final Lock writeLock = rwLock.writeLock(); // required for
checkpoint
-
- private volatile boolean updated = false;
- private volatile boolean recovered = false;
-
- public MinimalLockingWriteAheadLog(final Path path, final int
partitionCount, final SerDe<T> serde, final SyncListener syncListener) throws
IOException {
- this(new TreeSet<>(Collections.singleton(path)), partitionCount, new
SingletonSerDeFactory<>(serde), syncListener);
- }
-
- public MinimalLockingWriteAheadLog(final Path path, final int
partitionCount, final SerDeFactory<T> serdeFactory, final SyncListener
syncListener) throws IOException {
- this(new TreeSet<>(Collections.singleton(path)), partitionCount,
serdeFactory, syncListener);
- }
-
- public MinimalLockingWriteAheadLog(final SortedSet<Path> paths, final int
partitionCount, final SerDe<T> serde, final SyncListener syncListener) throws
IOException {
- this(paths, partitionCount, new SingletonSerDeFactory<>(serde),
syncListener);
- }
-
- /**
- *
- * @param paths a sorted set of Paths to use for the partitions/journals
and
- * the snapshot. The snapshot will always be written to the first path
- * specified.
- * @param partitionCount the number of partitions/journals to use. For best
- * performance, this should be close to the number of threads that are
- * expected to update the repository simultaneously
- * @param serdeFactory the factory for the serializer/deserializer for
records
- * @param syncListener the listener
- * @throws IOException if unable to initialize due to IO issue
- */
- @SuppressWarnings("unchecked")
- public MinimalLockingWriteAheadLog(final SortedSet<Path> paths, final int
partitionCount, final SerDeFactory<T> serdeFactory, final SyncListener
syncListener) throws IOException {
- this.syncListener = syncListener;
-
- requireNonNull(paths);
- requireNonNull(serdeFactory);
-
- if (paths.isEmpty()) {
- throw new IllegalArgumentException("Paths must be non-empty");
- }
-
- int resolvedPartitionCount = partitionCount;
- int existingPartitions = 0;
- for (final Path path : paths) {
- if (!Files.exists(path)) {
- Files.createDirectories(path);
- }
-
- final File file = path.toFile();
- if (!file.isDirectory()) {
- throw new IOException("Path given [" + path + "] is not a
directory");
- }
- if (!file.canWrite()) {
- throw new IOException("Path given [" + path + "] is not
writable");
- }
- if (!file.canRead()) {
- throw new IOException("Path given [" + path + "] is not
readable");
- }
- if (!file.canExecute()) {
- throw new IOException("Path given [" + path + "] is not
executable");
- }
-
- final File[] children = file.listFiles();
- if (children != null) {
- for (final File child : children) {
- if (child.isDirectory() &&
child.getName().startsWith("partition-")) {
- existingPartitions++;
- }
- }
-
- if (existingPartitions != 0 && existingPartitions !=
partitionCount) {
- logger.warn("Constructing MinimalLockingWriteAheadLog with
partitionCount={}, but the repository currently has "
- + "{} partitions; ignoring argument and proceeding
with {} partitions",
- new Object[]{partitionCount, existingPartitions,
existingPartitions});
- resolvedPartitionCount = existingPartitions;
- }
- }
- }
-
- this.basePath = paths.iterator().next();
- this.partialPath = basePath.resolve("snapshot.partial");
- this.snapshotPath = basePath.resolve("snapshot");
- this.serdeFactory = serdeFactory;
-
- final Path lockPath = basePath.resolve("wali.lock");
- lockChannel = new FileOutputStream(lockPath.toFile()).getChannel();
- lockChannel.lock();
-
- partitions = new Partition[resolvedPartitionCount];
-
- Iterator<Path> pathIterator = paths.iterator();
- for (int i = 0; i < resolvedPartitionCount; i++) {
- // If we're out of paths, create a new iterator to start over.
- if (!pathIterator.hasNext()) {
- pathIterator = paths.iterator();
- }
-
- final Path partitionBasePath = pathIterator.next();
-
- partitions[i] = new
Partition<>(partitionBasePath.resolve("partition-" + i), serdeFactory, i,
getVersion());
- }
- }
-
- @Override
- public int update(final Collection<T> records, final boolean forceSync)
throws IOException {
- if (!recovered) {
- throw new IllegalStateException("Cannot update repository until
record recovery has been performed");
- }
-
- if (records.isEmpty()) {
- return -1;
- }
-
- updated = true;
- readLock.lock();
- try {
- while (true) {
- final int numBlackListed = numberBlackListedPartitions.get();
- if (numBlackListed >= partitions.length) {
- throw new IOException("All Partitions have been
blacklisted due to "
- + "failures when attempting to update. If the
Write-Ahead Log is able to perform a checkpoint, "
- + "this issue may resolve itself. Otherwise,
manual intervention will be required.");
- }
-
- final long partitionIdx = partitionIndex.getAndIncrement();
- final int resolvedIdx = (int) (partitionIdx %
partitions.length);
- final Partition<T> partition = partitions[resolvedIdx];
- if (partition.tryClaim()) {
- try {
- final long transactionId =
transactionIdGenerator.getAndIncrement();
- if (logger.isTraceEnabled()) {
- for (final T record : records) {
- logger.trace("Partition {} performing
Transaction {}: {}", new Object[] {partition, transactionId, record});
- }
- }
-
- try {
- partition.update(records, transactionId,
unmodifiableRecordMap, forceSync);
- } catch (final Throwable t) {
- partition.blackList();
- numberBlackListedPartitions.incrementAndGet();
- throw t;
- }
-
- if (forceSync && syncListener != null) {
- syncListener.onSync(resolvedIdx);
- }
- } finally {
- partition.releaseClaim();
- }
-
- for (final T record : records) {
- final UpdateType updateType =
serdeFactory.getUpdateType(record);
- final Object recordIdentifier =
serdeFactory.getRecordIdentifier(record);
-
- if (updateType == UpdateType.DELETE) {
- recordMap.remove(recordIdentifier);
- } else if (updateType == UpdateType.SWAP_OUT) {
- final String newLocation =
serdeFactory.getLocation(record);
- if (newLocation == null) {
- logger.error("Received Record (ID=" +
recordIdentifier + ") with UpdateType of SWAP_OUT but "
- + "no indicator of where the Record is
to be Swapped Out to; these records may be "
- + "lost when the repository is
restored!");
- } else {
- recordMap.remove(recordIdentifier);
- this.externalLocations.add(newLocation);
- }
- } else if (updateType == UpdateType.SWAP_IN) {
- final String newLocation =
serdeFactory.getLocation(record);
- if (newLocation == null) {
- logger.error("Received Record (ID=" +
recordIdentifier + ") with UpdateType of SWAP_IN but no "
- + "indicator of where the Record is to
be Swapped In from; these records may be duplicated "
- + "when the repository is restored!");
- } else {
- externalLocations.remove(newLocation);
- }
- recordMap.put(recordIdentifier, record);
- } else {
- recordMap.put(recordIdentifier, record);
- }
- }
-
- return resolvedIdx;
- }
- }
- } finally {
- readLock.unlock();
- }
- }
-
- @Override
- public Collection<T> recoverRecords() throws IOException {
- if (updated) {
- throw new IllegalStateException("Cannot recover records after
updating the repository; must call recoverRecords first");
- }
-
- final long recoverStart = System.nanoTime();
- writeLock.lock();
- try {
- Long maxTransactionId = recoverFromSnapshot(recordMap);
- recoverFromEdits(recordMap, maxTransactionId);
-
- for (final Partition<T> partition : partitions) {
- final long transId = partition.getMaxRecoveredTransactionId();
- if (maxTransactionId == null || transId > maxTransactionId) {
- maxTransactionId = transId;
- }
- }
-
- this.transactionIdGenerator.set(maxTransactionId + 1);
- this.externalLocations.addAll(recoveredExternalLocations);
- logger.info("{} finished recovering records. Performing Checkpoint
to ensure proper state of Partitions before updates", this);
- } finally {
- writeLock.unlock();
- }
- final long recoverNanos = System.nanoTime() - recoverStart;
- final long recoveryMillis =
TimeUnit.MILLISECONDS.convert(recoverNanos, TimeUnit.NANOSECONDS);
- logger.info("Successfully recovered {} records in {} milliseconds",
recordMap.size(), recoveryMillis);
- checkpoint();
-
- recovered = true;
- return recordMap.values();
- }
-
- @Override
- public Set<String> getRecoveredSwapLocations() throws IOException {
- return recoveredExternalLocations;
- }
-
- private Long recoverFromSnapshot(final Map<Object, T> recordMap) throws
IOException {
- final boolean partialExists = Files.exists(partialPath);
- final boolean snapshotExists = Files.exists(snapshotPath);
-
- if (!partialExists && !snapshotExists) {
- return null;
- }
-
- if (partialExists && snapshotExists) {
- // both files exist -- assume we failed while checkpointing. Delete
- // the partial file
- Files.delete(partialPath);
- } else if (partialExists) {
- // partial exists but snapshot does not -- we must have completed
- // creating the partial, deleted the snapshot
- // but crashed before renaming the partial to the snapshot. Just
- // rename partial to snapshot
- Files.move(partialPath, snapshotPath);
- }
-
- if (Files.size(snapshotPath) == 0) {
- logger.warn("{} Found 0-byte Snapshot file; skipping Snapshot file
in recovery", this);
- return null;
- }
-
- // at this point, we know the snapshotPath exists because if it
didn't, then we either returned null
- // or we renamed partialPath to snapshotPath. So just Recover from
snapshotPath.
- try (final DataInputStream dataIn = new DataInputStream(new
BufferedInputStream(Files.newInputStream(snapshotPath,
StandardOpenOption.READ)))) {
- final String waliImplementationClass = dataIn.readUTF();
- final int waliImplementationVersion = dataIn.readInt();
-
- if
(!waliImplementationClass.equals(MinimalLockingWriteAheadLog.class.getName())) {
- throw new IOException("Write-Ahead Log located at " +
snapshotPath + " was written using the "
- + waliImplementationClass + " class; cannot restore
using " + getClass().getName());
- }
-
- if (waliImplementationVersion > getVersion()) {
- throw new IOException("Write-Ahead Log located at " +
snapshotPath + " was written using version "
- + waliImplementationVersion + " of the " +
waliImplementationClass + " class; cannot restore using Version " +
getVersion());
- }
-
- final String serdeEncoding = dataIn.readUTF(); // ignore serde
class name for now
- final int serdeVersion = dataIn.readInt();
- final long maxTransactionId = dataIn.readLong();
- final int numRecords = dataIn.readInt();
-
- final SerDe<T> serde = serdeFactory.createSerDe(serdeEncoding);
- serde.readHeader(dataIn);
-
- for (int i = 0; i < numRecords; i++) {
- final T record = serde.deserializeRecord(dataIn, serdeVersion);
- if (record == null) {
- throw new EOFException();
- }
-
- final UpdateType updateType = serde.getUpdateType(record);
- if (updateType == UpdateType.DELETE) {
- logger.warn("While recovering from snapshot, found record
with type 'DELETE'; this record will not be restored");
- continue;
- }
-
- logger.trace("Recovered from snapshot: {}", record);
- recordMap.put(serde.getRecordIdentifier(record), record);
- }
-
- final int numSwapRecords = dataIn.readInt();
- final Set<String> swapLocations = new HashSet<>();
- for (int i = 0; i < numSwapRecords; i++) {
- swapLocations.add(dataIn.readUTF());
- }
- this.recoveredExternalLocations.addAll(swapLocations);
-
- logger.debug("{} restored {} Records and {} Swap Files from
Snapshot, ending with Transaction ID {}",
- new Object[]{this, numRecords,
recoveredExternalLocations.size(), maxTransactionId});
- return maxTransactionId;
- }
- }
-
- /**
- * Recovers records from the edit logs via the Partitions. Returns a
boolean
- * if recovery of a Partition requires the Write-Ahead Log be checkpointed
- * before modification.
- *
- * @param modifiableRecordMap map
- * @param maxTransactionIdRestored index of max restored transaction
- * @throws IOException if unable to recover from edits
- */
- private void recoverFromEdits(final Map<Object, T> modifiableRecordMap,
final Long maxTransactionIdRestored) throws IOException {
- final Map<Object, T> updateMap = new HashMap<>();
- final Map<Object, T> unmodifiableRecordMap =
Collections.unmodifiableMap(modifiableRecordMap);
- final Map<Object, T> ignorableMap = new HashMap<>();
- final Set<String> ignorableSwapLocations = new HashSet<>();
-
- // populate a map of the next transaction id for each partition to the
- // partition that has that next transaction id.
- final SortedMap<Long, Partition<T>> transactionMap = new TreeMap<>();
- for (final Partition<T> partition : partitions) {
- Long transactionId;
- boolean keepTransaction;
- do {
- transactionId = partition.getNextRecoverableTransactionId();
-
- keepTransaction = transactionId == null ||
maxTransactionIdRestored == null || transactionId > maxTransactionIdRestored;
- if (keepTransaction && transactionId != null) {
- // map this transaction id to its partition so that we can
- // start restoring transactions from this partition,
- // starting at 'transactionId'
- transactionMap.put(transactionId, partition);
- } else if (transactionId != null) {
- // skip the next transaction, because our snapshot already
- // contained this transaction.
- try {
- partition.recoverNextTransaction(ignorableMap,
updateMap, ignorableSwapLocations);
- } catch (final EOFException e) {
- logger.error("{} unexpectedly reached End of File
while reading from {} for Transaction {}; "
- + "assuming crash and ignoring this
transaction.",
- new Object[]{this, partition, transactionId});
- }
- }
- } while (!keepTransaction);
- }
-
- while (!transactionMap.isEmpty()) {
- final Map.Entry<Long, Partition<T>> firstEntry =
transactionMap.entrySet().iterator().next();
- final Long firstTransactionId = firstEntry.getKey();
- final Partition<T> nextPartition = firstEntry.getValue();
-
- try {
- updateMap.clear();
- final Set<Object> idsRemoved =
nextPartition.recoverNextTransaction(unmodifiableRecordMap, updateMap,
recoveredExternalLocations);
- modifiableRecordMap.putAll(updateMap);
- for (final Object id : idsRemoved) {
- modifiableRecordMap.remove(id);
- }
- } catch (final EOFException e) {
- logger.error("{} unexpectedly reached End-of-File when reading
from {} for Transaction ID {}; "
- + "assuming crash and ignoring this transaction",
- new Object[]{this, nextPartition, firstTransactionId});
- }
-
- transactionMap.remove(firstTransactionId);
-
- Long subsequentTransactionId = null;
- try {
- subsequentTransactionId =
nextPartition.getNextRecoverableTransactionId();
- } catch (final IOException e) {
- logger.error("{} unexpectedly found End-of-File when reading
from {} for Transaction ID {}; "
- + "assuming crash and ignoring this transaction",
- new Object[]{this, nextPartition, firstTransactionId});
- }
-
- if (subsequentTransactionId != null) {
- transactionMap.put(subsequentTransactionId, nextPartition);
- }
- }
-
- for (final Partition<T> partition : partitions) {
- partition.endRecovery();
- }
- }
-
- @Override
- public synchronized int checkpoint() throws IOException {
- final Set<T> records;
- final Set<String> swapLocations;
- final long maxTransactionId;
-
- final long startNanos = System.nanoTime();
-
- FileOutputStream fileOut = null;
- DataOutputStream dataOut = null;
-
- long stopTheWorldNanos = -1L;
- long stopTheWorldStart = -1L;
- try {
- final List<OutputStream> partitionStreams = new ArrayList<>();
-
- writeLock.lock();
- try {
- stopTheWorldStart = System.nanoTime();
- // stop the world while we make a copy of the records that must
- // be checkpointed and rollover the partitions.
- // We copy the records because serializing them is potentially
- // very expensive, especially when we have hundreds
- // of thousands or even millions of them. We don't want to
- // prevent WALI from being used during this time.
-
- // So the design is to copy all of the records, determine the
- // last transaction ID that the records represent,
- // and roll over the partitions to new write-ahead logs.
- // Then, outside of the write lock, we will serialize the data
- // to disk, and then remove the old Partition data.
- records = new HashSet<>(recordMap.values());
- maxTransactionId = transactionIdGenerator.get() - 1;
-
- swapLocations = new HashSet<>(externalLocations);
- for (final Partition<T> partition : partitions) {
- try {
- partitionStreams.add(partition.rollover());
- } catch (final Throwable t) {
- partition.blackList();
- numberBlackListedPartitions.getAndIncrement();
- throw t;
- }
- }
- } finally {
- writeLock.unlock();
- }
-
- stopTheWorldNanos = System.nanoTime() - stopTheWorldStart;
-
- // Close all of the Partitions' Output Streams. We do this here,
instead of in Partition.rollover()
- // because we want to do this outside of the write lock. Because
calling close() on FileOutputStream can
- // be very expensive, as it has to flush the data to disk, we
don't want to prevent other Process Sessions
- // from getting committed. Since rollover() transitions the
partition to write to a new file already, there
- // is no reason that we need to close this FileOutputStream before
releasing the write lock. Also, if any Exception
- // does get thrown when calling close(), we don't need to
blacklist the partition, as the stream that was getting
- // closed is not the stream being written to for the partition
anyway. We also catch any IOException and wait until
- // after we've attempted to close all streams before we throw an
Exception, to avoid resource leaks if one of them
- // is unable to be closed (due to out of storage space, for
instance).
- IOException failure = null;
- for (final OutputStream partitionStream : partitionStreams) {
- try {
- partitionStream.close();
- } catch (final IOException e) {
- failure = e;
- }
- }
- if (failure != null) {
- throw failure;
- }
-
- // notify global sync with the write lock held. We do this because
we don't want the repository to get updated
- // while the listener is performing its necessary tasks
- if (syncListener != null) {
- syncListener.onGlobalSync();
- }
-
- final SerDe<T> serde = serdeFactory.createSerDe(null);
-
- // perform checkpoint, writing to .partial file
- fileOut = new FileOutputStream(partialPath.toFile());
- dataOut = new DataOutputStream(new BufferedOutputStream(fileOut));
- dataOut.writeUTF(MinimalLockingWriteAheadLog.class.getName());
- dataOut.writeInt(getVersion());
- dataOut.writeUTF(serde.getClass().getName());
- dataOut.writeInt(serde.getVersion());
- dataOut.writeLong(maxTransactionId);
- dataOut.writeInt(records.size());
- serde.writeHeader(dataOut);
-
- for (final T record : records) {
- logger.trace("Checkpointing {}", record);
- serde.serializeRecord(record, dataOut);
- }
-
- dataOut.writeInt(swapLocations.size());
- for (final String swapLocation : swapLocations) {
- dataOut.writeUTF(swapLocation);
- }
- } finally {
- if (dataOut != null) {
- try {
- try {
- dataOut.flush();
- fileOut.getFD().sync();
- } finally {
- dataOut.close();
- }
- } catch (final IOException e) {
- logger.warn("Failed to close Data Stream due to {}",
e.toString(), e);
- }
- }
- }
-
- // delete the snapshot, if it exists, and rename the .partial to
- // snapshot
- Files.deleteIfExists(snapshotPath);
- Files.move(partialPath, snapshotPath);
-
- // clear all of the edit logs
- final long partitionStart = System.nanoTime();
- for (final Partition<T> partition : partitions) {
- // we can call clearOld without claiming the partition because it
- // does not change the partition's state
- // and the only member variable it touches cannot be modified,
other
- // than when #rollover() is called.
- // And since this method is the only one that calls #rollover() and
- // this method is synchronized,
- // the value of that member variable will not change. And it's
- // volatile, so we will get the correct value.
- partition.clearOld();
- }
- final long partitionEnd = System.nanoTime();
- numberBlackListedPartitions.set(0);
-
- final long endNanos = System.nanoTime();
- final long millis = TimeUnit.MILLISECONDS.convert(endNanos -
startNanos, TimeUnit.NANOSECONDS);
- final long partitionMillis =
TimeUnit.MILLISECONDS.convert(partitionEnd - partitionStart,
TimeUnit.NANOSECONDS);
- final long stopTheWorldMillis =
TimeUnit.NANOSECONDS.toMillis(stopTheWorldNanos);
-
- logger.info("{} checkpointed with {} Records and {} Swap Files in {}
milliseconds (Stop-the-world "
- + "time = {} milliseconds, Clear Edit Logs time = {} millis),
max Transaction ID {}",
- new Object[]{this, records.size(), swapLocations.size(),
millis, stopTheWorldMillis, partitionMillis, maxTransactionId});
-
- return records.size();
- }
-
- @Override
- public void shutdown() throws IOException {
- writeLock.lock();
- try {
- for (final Partition<T> partition : partitions) {
- partition.close();
- }
- } finally {
- writeLock.unlock();
- lockChannel.close();
-
- final File lockFile = new File(basePath.toFile(), "wali.lock");
- lockFile.delete();
- }
- }
-
- public int getVersion() {
- return 1;
- }
-
- /**
- * Represents a partition of this repository, which maps directly to a
- * .journal file.
- *
- * All methods with the exceptions of {@link #claim()}, {@link
#tryClaim()},
- * and {@link #releaseClaim()} in this Partition MUST be called while
- * holding the claim (via {@link #claim} or {@link #tryClaim()}).
- *
- * @param <S> type of record held in the partitions
- */
- private static class Partition<S> {
- public static final String JOURNAL_EXTENSION = ".journal";
- private static final int NUL_BYTE = 0;
- private static final Pattern JOURNAL_FILENAME_PATTERN =
Pattern.compile("\\d+\\.journal");
-
- private final SerDeFactory<S> serdeFactory;
- private SerDe<S> serde;
-
- private final Path editDirectory;
- private final int writeAheadLogVersion;
-
- private DataOutputStream dataOut = null;
- private FileOutputStream fileOut = null;
- private volatile boolean blackListed = false;
- private volatile boolean closed = false;
- private DataInputStream recoveryIn;
- private int recoveryVersion;
- private String currentJournalFilename = "";
-
- private static final byte TRANSACTION_CONTINUE = 1;
- private static final byte TRANSACTION_COMMIT = 2;
-
- private final String description;
- private final AtomicLong maxTransactionId = new AtomicLong(-1L);
- private final Logger logger =
LoggerFactory.getLogger(MinimalLockingWriteAheadLog.class);
-
- private final Queue<Path> recoveryFiles;
-
- public Partition(final Path path, final SerDeFactory<S> serdeFactory,
final int partitionIndex, final int writeAheadLogVersion) throws IOException {
- this.editDirectory = path;
- this.serdeFactory = serdeFactory;
-
- final File file = path.toFile();
- if (!file.exists() && !file.mkdirs()) {
- throw new IOException("Could not create directory " +
file.getAbsolutePath());
- }
-
- this.recoveryFiles = new LinkedBlockingQueue<>();
- for (final Path recoveryPath : getRecoveryPaths()) {
- recoveryFiles.add(recoveryPath);
- }
-
- this.description = "Partition-" + partitionIndex;
- this.writeAheadLogVersion = writeAheadLogVersion;
- }
-
- public boolean tryClaim() {
- return !blackListed;
- }
-
- public void releaseClaim() {
- }
-
- public void close() {
- this.closed = true;
-
- // Note that here we are closing fileOut and NOT dataOut.
- // This is very much intentional, not an oversight. This is done
because of
- // the way that the OutputStreams are structured. dataOut wraps a
BufferedOutputStream,
- // which then wraps the FileOutputStream. If we close 'dataOut',
then this will call
- // the flush() method of BufferedOutputStream. Under normal
conditions, this is fine.
- // However, there is a very important corner case to consider:
- //
- // If we are writing to the DataOutputStream in the update()
method and that
- // call to write() then results in the BufferedOutputStream
calling flushBuffer() -
- // or if we finish the call to update() and call flush()
ourselves - it is possible
- // that the internal buffer of the BufferedOutputStream can
get partially written to
- // to the FileOutputStream and then an IOException occurs. If
this occurs, we have
- // written a partial record to disk. This still is okay, as
we have logic to handle
- // the condition where we have a partial record and then an
unexpected End-of-File.
- // But if we then call close() on 'dataOut', this will call
the flush() method of the
- // underlying BufferedOutputStream. As a result, we will end
up again writing the internal
- // buffer of the BufferedOutputStream to the underlying file.
At this point, we are left
- // not with an unexpected/premature End-of-File but instead a
bunch of seemingly random
- // bytes that happened to be residing in that internal
buffer, and this will result in
- // a corrupt and unrecoverable Write-Ahead Log.
- //
- // Additionally, we are okay not ever calling close on the
wrapping BufferedOutputStream and
- // DataOutputStream because they don't actually hold any resources
that need to be reclaimed,
- // and after each update to the Write-Ahead Log, we call flush()
ourselves to ensure that we don't
- // leave arbitrary data in the BufferedOutputStream that hasn't
been flushed to the underlying
- // FileOutputStream.
- final OutputStream out = fileOut;
- if (out != null) {
- try {
- out.close();
- } catch (final Exception e) {
- }
- }
-
- this.dataOut = null;
- this.fileOut = null;
- }
-
- public void blackList() {
- blackListed = true;
- logger.debug("Blacklisted {}", this);
- }
-
- /**
- * Closes resources pointing to the current journal and begins writing
- * to a new one
- *
- * @throws IOException if failure to rollover
- */
- public OutputStream rollover() throws IOException {
- // Note that here we are closing fileOut and NOT dataOut. See the
note in the close()
- // method to understand the logic behind this.
- final OutputStream oldOutputStream = fileOut;
- dataOut = null;
- fileOut = null;
-
- this.serde = serdeFactory.createSerDe(null);
- final Path editPath = getNewEditPath();
- final FileOutputStream fos = new
FileOutputStream(editPath.toFile());
- try {
- final DataOutputStream outStream = new DataOutputStream(new
BufferedOutputStream(fos));
-
outStream.writeUTF(MinimalLockingWriteAheadLog.class.getName());
- outStream.writeInt(writeAheadLogVersion);
- outStream.writeUTF(serde.getClass().getName());
- outStream.writeInt(serde.getVersion());
- serde.writeHeader(outStream);
-
- outStream.flush();
- dataOut = outStream;
- fileOut = fos;
- } catch (final IOException ioe) {
- try {
- oldOutputStream.close();
- } catch (final IOException ioe2) {
- ioe.addSuppressed(ioe2);
- }
-
- logger.error("Failed to create new journal for {} due to {}",
new Object[] {this, ioe.toString()}, ioe);
- try {
- fos.close();
- } catch (final IOException innerIOE) {
- }
-
- dataOut = null;
- fileOut = null;
- blackList();
-
- throw ioe;
- }
-
- currentJournalFilename = editPath.toFile().getName();
-
- blackListed = false;
- return oldOutputStream;
- }
-
- private long getJournalIndex(final File file) {
- final String filename = file.getName();
- final int dotIndex = filename.indexOf(".");
- final String number = filename.substring(0, dotIndex);
- return Long.parseLong(number);
- }
-
- private Path getNewEditPath() {
- final List<Path> recoveryPaths = getRecoveryPaths();
- final long newIndex;
- if (recoveryPaths == null || recoveryPaths.isEmpty()) {
- newIndex = 1;
- } else {
- final long lastFileIndex =
getJournalIndex(recoveryPaths.get(recoveryPaths.size() - 1).toFile());
- newIndex = lastFileIndex + 1;
- }
-
- return editDirectory.resolve(newIndex + JOURNAL_EXTENSION);
- }
-
- private List<Path> getRecoveryPaths() {
- final List<Path> paths = new ArrayList<>();
-
- final File directory = editDirectory.toFile();
- final File[] partitionFiles = directory.listFiles();
- if (partitionFiles == null) {
- return paths;
- }
-
- for (final File file : partitionFiles) {
- // if file is a journal file but no data has yet been
persisted, it may
- // very well be a 0-byte file (the journal is not SYNC'ed to
disk after
- // a header is written out, so it may be lost). In this case,
the journal
- // is empty, so we can just skip it.
- if (file.isDirectory() || file.length() == 0L) {
- continue;
- }
-
- if
(!JOURNAL_FILENAME_PATTERN.matcher(file.getName()).matches()) {
- continue;
- }
-
- if (isJournalFile(file)) {
- paths.add(file.toPath());
- } else {
- logger.warn("Found file {}, but could not access it, or it
was not in the expected format; "
- + "will ignore this file", file.getAbsolutePath());
- }
- }
-
- // Sort journal files by the numeric portion of the filename
- Collections.sort(paths, new Comparator<Path>() {
- @Override
- public int compare(final Path o1, final Path o2) {
- if (o1 == null && o2 == null) {
- return 0;
- }
- if (o1 == null) {
- return 1;
- }
- if (o2 == null) {
- return -1;
- }
-
- final long index1 = getJournalIndex(o1.toFile());
- final long index2 = getJournalIndex(o2.toFile());
- return Long.compare(index1, index2);
- }
- });
-
- return paths;
- }
-
- void clearOld() {
- final List<Path> oldRecoveryFiles = getRecoveryPaths();
-
- for (final Path path : oldRecoveryFiles) {
- final File file = path.toFile();
- if (file.getName().equals(currentJournalFilename)) {
- continue;
- }
- if (file.exists()) {
- file.delete();
- }
- }
- }
-
- private boolean isJournalFile(final File file) {
- final String expectedStartsWith =
MinimalLockingWriteAheadLog.class.getName();
- try {
- try (final FileInputStream fis = new FileInputStream(file);
- final InputStream bufferedIn = new
BufferedInputStream(fis);
- final DataInputStream in = new
DataInputStream(bufferedIn)) {
- final String waliImplClassName = in.readUTF();
- if (!expectedStartsWith.equals(waliImplClassName)) {
- return false;
- }
- }
- } catch (final IOException e) {
- return false;
- }
-
- return true;
- }
-
- public void update(final Collection<S> records, final long
transactionId, final Map<Object, S> recordMap, final boolean forceSync) throws
IOException {
- try (final ByteArrayOutputStream baos = new
ByteArrayOutputStream(256);
- final DataOutputStream out = new DataOutputStream(baos)) {
-
- out.writeLong(transactionId);
- final int numEditsToSerialize = records.size();
- int editsSerialized = 0;
- for (final S record : records) {
- final Object recordId = serde.getRecordIdentifier(record);
- final S previousVersion = recordMap.get(recordId);
-
- serde.serializeEdit(previousVersion, record, out);
- if (++editsSerialized < numEditsToSerialize) {
- out.write(TRANSACTION_CONTINUE);
- } else {
- out.write(TRANSACTION_COMMIT);
- }
- }
-
- out.flush();
-
- if (this.closed) {
- throw new IllegalStateException("Partition is closed");
- }
-
- baos.writeTo(dataOut);
- dataOut.flush();
-
- if (forceSync) {
- synchronized (fileOut) {
- fileOut.getFD().sync();
- }
- }
- }
- }
-
- private DataInputStream createDataInputStream(final Path path) throws
IOException {
- return new DataInputStream(new
BufferedInputStream(Files.newInputStream(path)));
- }
-
- private DataInputStream getRecoveryStream() throws IOException {
- if (recoveryIn != null && hasMoreData(recoveryIn)) {
- return recoveryIn;
- }
-
- while (true) {
- final Path nextRecoveryPath = recoveryFiles.poll();
- if (nextRecoveryPath == null) {
- return null;
- }
-
- logger.debug("{} recovering from {}", this, nextRecoveryPath);
- recoveryIn = createDataInputStream(nextRecoveryPath);
- if (hasMoreData(recoveryIn)) {
- try {
- final String waliImplementationClass =
recoveryIn.readUTF();
- if
(!MinimalLockingWriteAheadLog.class.getName().equals(waliImplementationClass)) {
- continue;
- }
-
- final long waliVersion = recoveryIn.readInt();
- if (waliVersion > writeAheadLogVersion) {
- throw new IOException("Cannot recovery from file "
+ nextRecoveryPath + " because it was written using "
- + "WALI version " + waliVersion + ", but the
version used to restore it is only " + writeAheadLogVersion);
- }
-
- final String serdeEncoding = recoveryIn.readUTF();
- this.recoveryVersion = recoveryIn.readInt();
- serde = serdeFactory.createSerDe(serdeEncoding);
-
- serde.readHeader(recoveryIn);
- break;
- } catch (final Exception e) {
- logger.warn("Failed to recover data from Write-Ahead
Log for {} because the header information could not be read properly. "
- + "This often is the result of the file not being
fully written out before the application is restarted. This file will be
ignored.", nextRecoveryPath);
- }
- }
- }
-
- return recoveryIn;
- }
-
- public Long getNextRecoverableTransactionId() throws IOException {
- while (true) {
- DataInputStream recoveryStream = getRecoveryStream();
- if (recoveryStream == null) {
- return null;
- }
-
- final long transactionId;
- try {
- transactionId = recoveryIn.readLong();
- } catch (final EOFException e) {
- continue;
- } catch (final Exception e) {
- // If the stream consists solely of NUL bytes, then we
want to treat it
- // the same as an EOF because we see this happen when we
suddenly lose power
- // while writing to a file.
- if (remainingBytesAllNul(recoveryIn)) {
- logger.warn("Failed to recover data from Write-Ahead
Log Partition because encountered trailing NUL bytes. "
- + "This will sometimes happen after a sudden power
loss. The rest of this journal file will be skipped for recovery purposes.");
- continue;
- } else {
- throw e;
- }
- }
-
- this.maxTransactionId.set(transactionId);
- return transactionId;
- }
- }
-
- /**
- * In the case of a sudden power loss, it is common - at least in a
Linux journaling File System -
- * that the partition file that is being written to will have many
trailing "NUL bytes" (0's).
- * If this happens, then on restart we want to treat this as an
incomplete transaction, so we detect
- * this case explicitly.
- *
- * @param in the input stream to scan
- * @return <code>true</code> if the InputStream contains no data or
contains only NUL bytes
- * @throws IOException if unable to read from the given InputStream
- */
- private boolean remainingBytesAllNul(final InputStream in) throws
IOException {
- int nextByte;
- while ((nextByte = in.read()) != -1) {
- if (nextByte != NUL_BYTE) {
- return false;
- }
- }
-
- return true;
- }
-
- private boolean hasMoreData(final InputStream in) throws IOException {
- in.mark(1);
- final int nextByte = in.read();
- in.reset();
- return nextByte >= 0;
- }
-
- public void endRecovery() throws IOException {
- if (recoveryIn != null) {
- recoveryIn.close();
- }
-
- final Path nextRecoveryPath = this.recoveryFiles.poll();
- if (nextRecoveryPath != null) {
- throw new IllegalStateException("Signaled to end recovery, but
there are more recovery files for Partition "
- + "in directory " + editDirectory);
- }
-
- final Path newEditPath = getNewEditPath();
-
- this.serde = serdeFactory.createSerDe(null);
- final FileOutputStream fos = new
FileOutputStream(newEditPath.toFile());
- final DataOutputStream outStream = new DataOutputStream(new
BufferedOutputStream(fos));
- outStream.writeUTF(MinimalLockingWriteAheadLog.class.getName());
- outStream.writeInt(writeAheadLogVersion);
- outStream.writeUTF(serde.getClass().getName());
- outStream.writeInt(serde.getVersion());
- serde.writeHeader(outStream);
-
- outStream.flush();
- dataOut = outStream;
- fileOut = fos;
- }
-
- public Set<Object> recoverNextTransaction(final Map<Object, S>
currentRecordMap, final Map<Object, S> updatedRecordMap, final Set<String>
swapLocations) throws IOException {
- final Set<Object> idsRemoved = new HashSet<>();
-
- int transactionFlag;
- do {
- final S record;
- try {
- record = serde.deserializeEdit(recoveryIn,
currentRecordMap, recoveryVersion);
- if (record == null) {
- throw new EOFException();
- }
- } catch (final EOFException eof) {
- throw eof;
- } catch (final Exception e) {
- // If the stream consists solely of NUL bytes, then we
want to treat it
- // the same as an EOF because we see this happen when we
suddenly lose power
- // while writing to a file. We also have logic already in
the caller of this
- // method to properly handle EOFException's, so we will
simply throw an EOFException
- // ourselves. However, if that is not the case, then
something else has gone wrong.
- // In such a case, there is not much that we can do. If we
simply skip over the transaction,
- // then the transaction may be indicating that a new
attribute was added or changed. Or the
- // content of the FlowFile changed. A subsequent
transaction for the same FlowFile may then
- // update the connection that is holding the FlowFile. In
this case, if we simply skip over
- // the transaction, we end up with a FlowFile in a queue
that has the wrong attributes or
- // content, and that can result in some very bad behavior
- even security vulnerabilities if
- // a Route processor, for instance, routes incorrectly due
to a missing attribute or content
- // is pointing to a previous claim where sensitive values
have not been removed, etc. So
- // instead of attempting to skip the transaction and move
on, we instead just throw the Exception
- // indicating that the write-ahead log is corrupt and
allow the user to handle it as he/she sees
- // fit (likely this will result in deleting the repo, but
it's possible that it could be repaired
- // manually or through some sort of script).
- if (remainingBytesAllNul(recoveryIn)) {
- final EOFException eof = new EOFException("Failed to
recover data from Write-Ahead Log Partition because encountered trailing NUL
bytes. "
- + "This will sometimes happen after a sudden power
loss. The rest of this journal file will be skipped for recovery purposes.");
- eof.addSuppressed(e);
- throw eof;
- } else {
- throw e;
- }
- }
-
-
- if (logger.isDebugEnabled()) {
- logger.debug("{} Recovering Transaction {}: {}", new
Object[] { this, maxTransactionId.get(), record });
- }
-
- final Object recordId = serde.getRecordIdentifier(record);
- final UpdateType updateType = serde.getUpdateType(record);
- if (updateType == UpdateType.DELETE) {
- updatedRecordMap.remove(recordId);
- idsRemoved.add(recordId);
- } else if (updateType == UpdateType.SWAP_IN) {
- final String location = serde.getLocation(record);
- if (location == null) {
- logger.error("Recovered SWAP_IN record from edit log,
but it did not contain a Location; skipping record");
- } else {
- swapLocations.remove(location);
- updatedRecordMap.put(recordId, record);
- idsRemoved.remove(recordId);
- }
- } else if (updateType == UpdateType.SWAP_OUT) {
- final String location = serde.getLocation(record);
- if (location == null) {
- logger.error("Recovered SWAP_OUT record from edit log,
but it did not contain a Location; skipping record");
- } else {
- swapLocations.add(location);
- updatedRecordMap.remove(recordId);
- idsRemoved.add(recordId);
- }
- } else {
- updatedRecordMap.put(recordId, record);
- idsRemoved.remove(recordId);
- }
-
- transactionFlag = recoveryIn.read();
- } while (transactionFlag != TRANSACTION_COMMIT);
-
- return idsRemoved;
- }
-
- /**
- * Must be called after recovery has finished
- *
- * @return max recovered transaction id
- */
- public long getMaxRecoveredTransactionId() {
- return maxTransactionId.get();
- }
-
- @Override
- public String toString() {
- return description;
- }
- }
-}
diff --git
a/nifi-commons/nifi-write-ahead-log/src/test/java/org/wali/TestMinimalLockingWriteAheadLog.java
b/nifi-commons/nifi-write-ahead-log/src/test/java/org/wali/TestMinimalLockingWriteAheadLog.java
deleted file mode 100644
index 7954553110..0000000000
---
a/nifi-commons/nifi-write-ahead-log/src/test/java/org/wali/TestMinimalLockingWriteAheadLog.java
+++ /dev/null
@@ -1,977 +0,0 @@
-/*
- * 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.wali;
-
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.BufferedInputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.File;
-import java.io.FileFilter;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.file.Files;
-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.List;
-import java.util.Map;
-import java.util.Random;
-import java.util.SortedSet;
-import java.util.TreeSet;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-@SuppressWarnings("deprecation")
-public class TestMinimalLockingWriteAheadLog {
- private static final Logger logger =
LoggerFactory.getLogger(TestMinimalLockingWriteAheadLog.class);
-
-
- @Test
- public void testTruncatedPartitionHeader() throws IOException {
- final int numPartitions = 4;
-
- final Path path = Paths.get("target/testTruncatedPartitionHeader");
- deleteRecursively(path.toFile());
- assertTrue(path.toFile().mkdirs());
-
- final AtomicInteger counter = new AtomicInteger(0);
- final SerDe<Object> serde = new SerDe<Object>() {
- @Override
- public void readHeader(DataInputStream in) throws IOException {
- if (counter.getAndIncrement() == 1) {
- throw new EOFException("Intentionally thrown for unit
test");
- }
- }
-
- @Override
- public void serializeEdit(Object previousRecordState, Object
newRecordState, DataOutputStream out) throws IOException {
- out.write(1);
- }
-
- @Override
- public void serializeRecord(Object record, DataOutputStream out)
throws IOException {
- out.write(1);
- }
-
- @Override
- public Object deserializeEdit(DataInputStream in, Map<Object,
Object> currentRecordStates, int version) throws IOException {
- final int val = in.read();
- return (val == 1) ? new Object() : null;
- }
-
- @Override
- public Object deserializeRecord(DataInputStream in, int version)
throws IOException {
- final int val = in.read();
- return (val == 1) ? new Object() : null;
- }
-
- @Override
- public Object getRecordIdentifier(Object record) {
- return 1;
- }
-
- @Override
- public UpdateType getUpdateType(Object record) {
- return UpdateType.CREATE;
- }
-
- @Override
- public String getLocation(Object record) {
- return null;
- }
-
- @Override
- public int getVersion() {
- return 0;
- }
- };
-
- final WriteAheadRepository<Object> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- try {
- final Collection<Object> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- repo.update(Collections.singletonList(new Object()), false);
- repo.update(Collections.singletonList(new Object()), false);
- repo.update(Collections.singletonList(new Object()), false);
- } finally {
- repo.shutdown();
- }
-
- final WriteAheadRepository<Object> secondRepo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- try {
- secondRepo.recoverRecords();
- } finally {
- secondRepo.shutdown();
- }
- }
-
- @Test
- @Disabled("For manual performance testing")
- public void testUpdatePerformance() throws IOException,
InterruptedException {
- final int numPartitions = 16;
-
- final Path path = Paths.get("target/minimal-locking-repo");
- deleteRecursively(path.toFile());
- assertTrue(path.toFile().mkdirs());
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- final long updateCountPerThread = 1_000_000;
- final int numThreads = 4;
-
- final Thread[] threads = new Thread[numThreads];
-
- final int batchSize = 1;
-
- long previousBytes = 0;
-
- for (int j = 0; j < 2; j++) {
- for (int i = 0; i < numThreads; i++) {
- final Thread t = new Thread(() -> {
- final List<DummyRecord> batch = new ArrayList<>();
-
- for (int i1 = 0; i1 < updateCountPerThread / batchSize;
i1++) {
- batch.clear();
- for (int j1 = 0; j1 < batchSize; j1++) {
- final DummyRecord record = new
DummyRecord(String.valueOf(i1), UpdateType.CREATE);
- batch.add(record);
- }
-
- assertDoesNotThrow(() -> repo.update(batch, false));
- }
- });
-
- threads[i] = t;
- }
-
- final long start = System.nanoTime();
- for (final Thread t : threads) {
- t.start();
- }
- for (final Thread t : threads) {
- t.join();
- }
-
- long bytes = 0L;
- for (final File file : path.toFile().listFiles()) {
- if (file.getName().startsWith("partition-")) {
- for (final File journalFile : file.listFiles()) {
- bytes += journalFile.length();
- }
- }
- }
-
- bytes -= previousBytes;
- previousBytes = bytes;
-
- final long millis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
- final long eventsPerSecond = (updateCountPerThread * numThreads *
1000) / millis;
- final String eps =
NumberFormat.getInstance().format(eventsPerSecond);
- final long bytesPerSecond = bytes * 1000 / millis;
- final String bps =
NumberFormat.getInstance().format(bytesPerSecond);
-
- if (j == 0) {
- System.out.println(millis + " ms to insert " +
updateCountPerThread * numThreads + " updates using " + numThreads + " threads,
*as a warmup!* "
- + eps + " events per second, " + bps + " bytes per
second");
- } else {
- System.out.println(millis + " ms to insert " +
updateCountPerThread * numThreads + " updates using " + numThreads + " threads,
"
- + eps + " events per second, " + bps + " bytes per
second");
- }
- }
- }
-
-
-
- @Test
- public void testRepoDoesntContinuallyGrowOnOutOfMemoryError() throws
IOException, InterruptedException {
- final int numPartitions = 8;
-
- final Path path = Paths.get("target/minimal-locking-repo");
- deleteRecursively(path.toFile());
- assertTrue(path.toFile().mkdirs());
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- try {
- final Collection<DummyRecord> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- serde.setThrowOOMEAfterNSerializeEdits(100);
- for (int i = 0; i < 108; i++) {
- try {
- final DummyRecord record = new
DummyRecord(String.valueOf(i), UpdateType.CREATE);
- repo.update(Collections.singleton(record), false);
- } catch (final OutOfMemoryError oome) {
- logger.info("Received OOME on record " + i);
- }
- }
-
- long expectedSize = sizeOf(path.toFile());
- for (int i = 0; i < 1000; i++) {
- final DummyRecord record = new DummyRecord(String.valueOf(i),
UpdateType.CREATE);
- assertThrows(IOException.class, () ->
repo.update(Collections.singleton(record), false));
- }
-
- long newSize = sizeOf(path.toFile());
- assertEquals(expectedSize, newSize);
-
- assertThrows(OutOfMemoryError.class, () -> repo.checkpoint());
-
- expectedSize = sizeOf(path.toFile());
-
- for (int i = 0; i < 100000; i++) {
- final DummyRecord record = new DummyRecord(String.valueOf(i),
UpdateType.CREATE);
- assertThrows(IOException.class, () ->
repo.update(Collections.singleton(record), false));
- }
-
- newSize = sizeOf(path.toFile());
- assertEquals(expectedSize, newSize);
- } finally {
- repo.shutdown();
- }
- }
-
- /**
- * This test is intended to continually update the Write-ahead log using
many threads, then
- * stop and restore the repository to check for any corruption. There were
reports of potential threading
- * issues leading to repository corruption. This was an attempt to
replicate. It should not be run as a
- * unit test, really, but will be left, as it can be valuable to exercise
the implementation
- *
- * @throws IOException if unable to read from/write to the write-ahead log
- * @throws InterruptedException if a thread is interrupted
- */
- @Test
- @Disabled
- public void tryToCauseThreadingIssue() throws IOException,
InterruptedException {
- System.setProperty("org.slf4j.simpleLogger.log.org.wali", "INFO");
-
- final int numThreads = 12;
- final long iterationsPerThread = 1000000;
- final int numAttempts = 1000;
-
- final Path path = Paths.get("D:/dummy/minimal-locking-repo");
- path.toFile().mkdirs();
-
- final AtomicReference<WriteAheadRepository<DummyRecord>> writeRepoRef
= new AtomicReference<>();
- final AtomicBoolean checkpointing = new AtomicBoolean(false);
-
- final Thread bgThread = new Thread(() -> {
- while (true) {
- checkpointing.set(true);
-
- final WriteAheadRepository<DummyRecord> repo =
writeRepoRef.get();
- if (repo != null) {
- assertDoesNotThrow(() -> repo.checkpoint());
- }
-
- checkpointing.set(false);
-
- try {
- TimeUnit.SECONDS.sleep(5);
- } catch (InterruptedException e) {
- }
- }
- });
- bgThread.setDaemon(true);
- bgThread.start();
-
- for (int x = 0; x < numAttempts; x++) {
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> writeRepo = new
MinimalLockingWriteAheadLog<>(path, 256, serde, null);
- final Collection<DummyRecord> writeRecords =
writeRepo.recoverRecords();
- for (final DummyRecord record : writeRecords) {
- assertEquals("B", record.getProperty("A"));
- }
-
- writeRepoRef.set(writeRepo);
-
- final Thread[] threads = new Thread[numThreads];
- for (int i = 0; i < numThreads; i++) {
- final Thread t = new
InlineCreationInsertThread(iterationsPerThread, writeRepo);
- t.start();
- threads[i] = t;
- }
-
- for (final Thread t : threads) {
- t.join();
- }
-
- writeRepoRef.set(null);
- writeRepo.shutdown();
-
- boolean cp = checkpointing.get();
- while (cp) {
- Thread.sleep(100L);
- cp = checkpointing.get();
- }
-
- final WriteAheadRepository<DummyRecord> readRepo = new
MinimalLockingWriteAheadLog<>(path, 256, serde, null);
- // ensure that we are able to recover the records properly
- final Collection<DummyRecord> readRecords =
readRepo.recoverRecords();
- for (final DummyRecord record : readRecords) {
- assertEquals("B", record.getProperty("A"));
- }
- readRepo.shutdown();
- }
- }
-
- @Test
- public void testWrite() throws IOException, InterruptedException {
- final int numPartitions = 8;
-
- final Path path = Paths.get("target/minimal-locking-repo");
- deleteRecursively(path.toFile());
- assertTrue(path.toFile().mkdirs());
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- final List<InsertThread> threads = new ArrayList<>();
- for (int i = 0; i < 10; i++) {
- threads.add(new InsertThread(10000, 1000000 * i, repo));
- }
-
- final long start = System.nanoTime();
- for (final InsertThread thread : threads) {
- thread.start();
- }
- for (final InsertThread thread : threads) {
- thread.join();
- }
- final long nanos = System.nanoTime() - start;
- final long millis = TimeUnit.MILLISECONDS.convert(nanos,
TimeUnit.NANOSECONDS);
- System.out.println("Took " + millis + " millis to insert 1,000,000
records each in its own transaction");
- repo.shutdown();
-
- final WriteAheadRepository<DummyRecord> recoverRepo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> recoveredRecords =
recoverRepo.recoverRecords();
- assertFalse(recoveredRecords.isEmpty());
- assertEquals(100000, recoveredRecords.size());
- for (final DummyRecord record : recoveredRecords) {
- final Map<String, String> recoveredProps = record.getProperties();
- assertEquals(1, recoveredProps.size());
- assertEquals("B", recoveredProps.get("A"));
- }
- }
-
- @Test
- public void testRecoverAfterIOException() throws IOException {
- final int numPartitions = 5;
- final Path path =
Paths.get("target/minimal-locking-repo-test-recover-after-ioe");
- deleteRecursively(path.toFile());
- Files.createDirectories(path);
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- serde.setThrowIOEAfterNSerializeEdits(7); // serialize the 2
transactions, then the first edit of the third transaction; then throw
IOException
-
- final List<DummyRecord> firstTransaction = new ArrayList<>();
- firstTransaction.add(new DummyRecord("1", UpdateType.CREATE));
- firstTransaction.add(new DummyRecord("2", UpdateType.CREATE));
- firstTransaction.add(new DummyRecord("3", UpdateType.CREATE));
-
- final List<DummyRecord> secondTransaction = new ArrayList<>();
- secondTransaction.add(new DummyRecord("1",
UpdateType.UPDATE).setProperty("abc", "123"));
- secondTransaction.add(new DummyRecord("2",
UpdateType.UPDATE).setProperty("cba", "123"));
- secondTransaction.add(new DummyRecord("3",
UpdateType.UPDATE).setProperty("aaa", "123"));
-
- final List<DummyRecord> thirdTransaction = new ArrayList<>();
- thirdTransaction.add(new DummyRecord("1", UpdateType.DELETE));
- thirdTransaction.add(new DummyRecord("2", UpdateType.DELETE));
-
- repo.update(firstTransaction, true);
- repo.update(secondTransaction, true);
- assertThrows(IOException.class, () -> repo.update(thirdTransaction,
true));
-
- repo.shutdown();
-
- serde.setThrowIOEAfterNSerializeEdits(-1);
- final WriteAheadRepository<DummyRecord> recoverRepo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> recoveredRecords =
recoverRepo.recoverRecords();
- assertFalse(recoveredRecords.isEmpty());
- assertEquals(3, recoveredRecords.size());
-
- boolean record1 = false, record2 = false, record3 = false;
- for (final DummyRecord record : recoveredRecords) {
- switch (record.getId()) {
- case "1":
- record1 = true;
- assertEquals("123", record.getProperty("abc"));
- break;
- case "2":
- record2 = true;
- assertEquals("123", record.getProperty("cba"));
- break;
- case "3":
- record3 = true;
- assertEquals("123", record.getProperty("aaa"));
- break;
- }
- }
-
- assertTrue(record1);
- assertTrue(record2);
- assertTrue(record3);
- }
-
-
- @Test
- public void testRecoverFileThatHasTrailingNULBytesAndTruncation() throws
IOException {
- final int numPartitions = 5;
- final Path path =
Paths.get("target/testRecoverFileThatHasTrailingNULBytesAndTruncation");
- deleteRecursively(path.toFile());
- Files.createDirectories(path);
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- final List<DummyRecord> firstTransaction = new ArrayList<>();
- firstTransaction.add(new DummyRecord("1", UpdateType.CREATE));
- firstTransaction.add(new DummyRecord("2", UpdateType.CREATE));
- firstTransaction.add(new DummyRecord("3", UpdateType.CREATE));
-
- final List<DummyRecord> secondTransaction = new ArrayList<>();
- secondTransaction.add(new DummyRecord("1",
UpdateType.UPDATE).setProperty("abc", "123"));
- secondTransaction.add(new DummyRecord("2",
UpdateType.UPDATE).setProperty("cba", "123"));
- secondTransaction.add(new DummyRecord("3",
UpdateType.UPDATE).setProperty("aaa", "123"));
-
- final List<DummyRecord> thirdTransaction = new ArrayList<>();
- thirdTransaction.add(new DummyRecord("1", UpdateType.DELETE));
- thirdTransaction.add(new DummyRecord("2", UpdateType.DELETE));
-
- repo.update(firstTransaction, true);
- repo.update(secondTransaction, true);
- repo.update(thirdTransaction, true);
-
- repo.shutdown();
-
- final File partition3Dir = path.resolve("partition-2").toFile();
- final File journalFile = partition3Dir.listFiles()[0];
- final byte[] contents = Files.readAllBytes(journalFile.toPath());
-
- // Truncate the contents of the journal file by 8 bytes. Then replace
with 28 trailing NUL bytes,
- // as this is what we often see when we have a sudden power loss.
- final byte[] truncated = Arrays.copyOfRange(contents, 0,
contents.length - 8);
- final byte[] withNuls = new byte[truncated.length + 28];
- System.arraycopy(truncated, 0, withNuls, 0, truncated.length);
-
- try (final OutputStream fos = new FileOutputStream(journalFile)) {
- fos.write(withNuls);
- }
-
- final WriteAheadRepository<DummyRecord> recoverRepo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> recoveredRecords =
recoverRepo.recoverRecords();
- assertFalse(recoveredRecords.isEmpty());
- assertEquals(3, recoveredRecords.size());
-
- boolean record1 = false, record2 = false, record3 = false;
- for (final DummyRecord record : recoveredRecords) {
- switch (record.getId()) {
- case "1":
- record1 = true;
- assertEquals("123", record.getProperty("abc"));
- break;
- case "2":
- record2 = true;
- assertEquals("123", record.getProperty("cba"));
- break;
- case "3":
- record3 = true;
- assertEquals("123", record.getProperty("aaa"));
- break;
- }
- }
-
- assertTrue(record1);
- assertTrue(record2);
- assertTrue(record3);
- }
-
- @Test
- public void testRecoverFileThatHasTrailingNULBytesNoTruncation() throws
IOException {
- final int numPartitions = 5;
- final Path path =
Paths.get("target/testRecoverFileThatHasTrailingNULBytesNoTruncation");
- deleteRecursively(path.toFile());
- Files.createDirectories(path);
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- final List<DummyRecord> firstTransaction = new ArrayList<>();
- firstTransaction.add(new DummyRecord("1", UpdateType.CREATE));
- firstTransaction.add(new DummyRecord("2", UpdateType.CREATE));
- firstTransaction.add(new DummyRecord("3", UpdateType.CREATE));
-
- final List<DummyRecord> secondTransaction = new ArrayList<>();
- secondTransaction.add(new DummyRecord("1",
UpdateType.UPDATE).setProperty("abc", "123"));
- secondTransaction.add(new DummyRecord("2",
UpdateType.UPDATE).setProperty("cba", "123"));
- secondTransaction.add(new DummyRecord("3",
UpdateType.UPDATE).setProperty("aaa", "123"));
-
- final List<DummyRecord> thirdTransaction = new ArrayList<>();
- thirdTransaction.add(new DummyRecord("1", UpdateType.DELETE));
- thirdTransaction.add(new DummyRecord("2", UpdateType.DELETE));
-
- repo.update(firstTransaction, true);
- repo.update(secondTransaction, true);
- repo.update(thirdTransaction, true);
-
- repo.shutdown();
-
- final File partition3Dir = path.resolve("partition-2").toFile();
- final File journalFile = partition3Dir.listFiles()[0];
-
- // Truncate the contents of the journal file by 8 bytes. Then replace
with 28 trailing NUL bytes,
- // as this is what we often see when we have a sudden power loss.
- final byte[] withNuls = new byte[28];
-
- try (final OutputStream fos = new FileOutputStream(journalFile, true))
{
- fos.write(withNuls);
- }
-
- final WriteAheadRepository<DummyRecord> recoverRepo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> recoveredRecords =
recoverRepo.recoverRecords();
- assertFalse(recoveredRecords.isEmpty());
- assertEquals(1, recoveredRecords.size());
-
- boolean record1 = false, record2 = false, record3 = false;
- for (final DummyRecord record : recoveredRecords) {
- switch (record.getId()) {
- case "1":
- record1 = record.getUpdateType() != UpdateType.DELETE;
- assertEquals("123", record.getProperty("abc"));
- break;
- case "2":
- record2 = record.getUpdateType() != UpdateType.DELETE;
- assertEquals("123", record.getProperty("cba"));
- break;
- case "3":
- record3 = true;
- assertEquals("123", record.getProperty("aaa"));
- break;
- }
- }
-
- assertFalse(record1);
- assertFalse(record2);
- assertTrue(record3);
- }
-
- @Test
- public void testCannotModifyLogAfterAllAreBlackListed() throws IOException
{
- final int numPartitions = 5;
- final Path path =
Paths.get("target/minimal-locking-repo-test-cannot-modify-after-all-blacklisted");
- deleteRecursively(path.toFile());
- Files.createDirectories(path);
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- serde.setThrowIOEAfterNSerializeEdits(3); // serialize the first
transaction, then fail on all subsequent transactions
-
- final List<DummyRecord> firstTransaction = new ArrayList<>();
- firstTransaction.add(new DummyRecord("1", UpdateType.CREATE));
- firstTransaction.add(new DummyRecord("2", UpdateType.CREATE));
- firstTransaction.add(new DummyRecord("3", UpdateType.CREATE));
-
- final List<DummyRecord> secondTransaction = new ArrayList<>();
- secondTransaction.add(new DummyRecord("1",
UpdateType.UPDATE).setProperty("abc", "123"));
- secondTransaction.add(new DummyRecord("2",
UpdateType.UPDATE).setProperty("cba", "123"));
- secondTransaction.add(new DummyRecord("3",
UpdateType.UPDATE).setProperty("aaa", "123"));
-
- final List<DummyRecord> thirdTransaction = new ArrayList<>();
- thirdTransaction.add(new DummyRecord("1", UpdateType.DELETE));
- thirdTransaction.add(new DummyRecord("2", UpdateType.DELETE));
-
- repo.update(firstTransaction, true);
-
- assertThrows(IOException.class, () -> repo.update(secondTransaction,
true));
-
- for (int i = 0; i < 4; i++) {
- assertThrows(IOException.class, () ->
repo.update(thirdTransaction, true));
- }
-
- serde.setThrowIOEAfterNSerializeEdits(-1);
- final List<DummyRecord> fourthTransaction = new ArrayList<>();
- fourthTransaction.add(new DummyRecord("1", UpdateType.DELETE));
-
- IOException e = assertThrows(IOException.class, () ->
repo.update(fourthTransaction, true));
- assertTrue(e.getMessage().contains("All Partitions have been
blacklisted"));
-
- repo.shutdown();
- serde.setThrowIOEAfterNSerializeEdits(-1);
-
- final WriteAheadRepository<DummyRecord> recoverRepo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serde, null);
- final Collection<DummyRecord> recoveredRecords =
recoverRepo.recoverRecords();
- assertFalse(recoveredRecords.isEmpty());
- assertEquals(3, recoveredRecords.size());
- }
-
- @Test
- public void testStriping() throws IOException {
- final int numPartitions = 6;
- final Path path = Paths.get("target/minimal-locking-repo-striped");
- deleteRecursively(path.toFile());
- Files.createDirectories(path);
-
- final SortedSet<Path> paths = new TreeSet<>();
- paths.add(path.resolve("stripe-1"));
- paths.add(path.resolve("stripe-2"));
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> repo = new
MinimalLockingWriteAheadLog<>(paths, numPartitions, serde, null);
- final Collection<DummyRecord> initialRecs = repo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- final InsertThread inserter = new InsertThread(100000, 0, repo);
- inserter.run();
-
- for (final Path partitionPath : paths) {
- final File[] files = partitionPath.toFile().listFiles(new
FileFilter() {
- @Override
- public boolean accept(File pathname) {
- return pathname.getName().startsWith("partition");
- }
- });
- assertEquals(3, files.length);
-
- for (final File file : files) {
- final File[] journalFiles = file.listFiles();
- assertEquals(1, journalFiles.length);
- }
- }
-
- repo.checkpoint();
-
- }
-
- @Test
- public void testShutdownWhileBlacklisted() throws IOException {
- final Path path =
Paths.get("target/minimal-locking-repo-shutdown-blacklisted");
- deleteRecursively(path.toFile());
- Files.createDirectories(path);
-
- final SerDe<SimpleRecord> failOnThirdWriteSerde = new
SerDe<SimpleRecord>() {
- private int writes = 0;
-
- @Override
- public void serializeEdit(SimpleRecord previousRecordState,
SimpleRecord newRecordState, DataOutputStream out) throws IOException {
- serializeRecord(newRecordState, out);
- }
-
- @Override
- public void serializeRecord(SimpleRecord record, DataOutputStream
out) throws IOException {
- int size = (int) record.getSize();
- out.writeLong(record.getSize());
-
- for (int i = 0; i < size; i++) {
- out.write('A');
- }
-
- if (++writes == 3) {
- throw new IOException("Intentional Exception for Unit
Testing");
- }
-
- out.writeLong(record.getId());
- }
-
- @Override
- public SimpleRecord deserializeEdit(DataInputStream in,
Map<Object, SimpleRecord> currentRecordStates, int version) throws IOException {
- return deserializeRecord(in, version);
- }
-
- @Override
- public SimpleRecord deserializeRecord(DataInputStream in, int
version) throws IOException {
- long size = in.readLong();
-
- for (int i = 0; i < (int) size; i++) {
- in.read();
- }
-
- long id = in.readLong();
- return new SimpleRecord(id, size);
- }
-
- @Override
- public Object getRecordIdentifier(SimpleRecord record) {
- return record.getId();
- }
-
- @Override
- public UpdateType getUpdateType(SimpleRecord record) {
- return UpdateType.CREATE;
- }
-
- @Override
- public String getLocation(SimpleRecord record) {
- return null;
- }
-
- @Override
- public int getVersion() {
- return 0;
- }
- };
-
- final WriteAheadRepository<SimpleRecord> writeRepo = new
MinimalLockingWriteAheadLog<>(path, 1, failOnThirdWriteSerde, null);
- final Collection<SimpleRecord> initialRecs =
writeRepo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
-
- writeRepo.update(Collections.singleton(new SimpleRecord(1L, 1L)),
false);
- writeRepo.update(Collections.singleton(new SimpleRecord(2L, 2L)),
false);
- // Use a size of 8194 because the BufferedOutputStream has a buffer
size of 8192 and we want
- // to exceed this for testing purposes.
- assertThrows(IOException.class,
- () -> writeRepo.update(Collections.singleton(new
SimpleRecord(3L, 8194L)), false));
-
- final Path partitionDir = path.resolve("partition-0");
- final File journalFile = partitionDir.toFile().listFiles()[0];
- final long journalFileSize = journalFile.length();
- verifyBlacklistedJournalContents(journalFile, failOnThirdWriteSerde);
-
- writeRepo.shutdown();
-
- // Ensure that calling shutdown() didn't write anything to the journal
file
- final long newJournalSize = journalFile.length();
- assertEquals(newJournalSize, journalFile.length(), "Calling Shutdown
wrote " + (newJournalSize - journalFileSize) + " bytes to the journal file");
- }
-
- private void verifyBlacklistedJournalContents(final File journalFile,
final SerDe<?> serde) throws IOException {
- try (final FileInputStream fis = new FileInputStream(journalFile);
- final InputStream bis = new BufferedInputStream(fis);
- final DataInputStream in = new DataInputStream(bis)) {
-
- // Verify header info.
- final String waliClassName = in.readUTF();
- assertEquals(MinimalLockingWriteAheadLog.class.getName(),
waliClassName);
-
- final int waliVersion = in.readInt();
- assertTrue(waliVersion > 0);
-
- final String serdeClassName = in.readUTF();
- assertEquals(serde.getClass().getName(), serdeClassName);
-
- final int serdeVersion = in.readInt();
- assertEquals(serde.getVersion(), serdeVersion);
-
- for (int i = 0; i < 2; i++) {
- long transactionId = in.readLong();
- assertEquals(i, transactionId);
-
- // read what serde wrote
- long size = in.readLong();
-
- assertEquals((i + 1), size);
-
- for (int j = 0; j < (int) size; j++) {
- final int c = in.read();
- assertEquals('A', c);
- }
-
- long id = in.readLong();
- assertEquals((i + 1), id);
-
- int transactionIndicator = in.read();
- assertEquals(2, transactionIndicator);
- }
-
- // In previous implementations, we would still have a partial
record written out.
- // In the current version, however, the serde above would result
in the data serialization
- // failing and as a result no data would be written to the stream,
so the stream should
- // now be out of data
- final int nextByte = in.read();
- assertEquals(-1, nextByte);
- }
- }
-
-
-
- @Test
- public void testDecreaseNumberOfPartitions() throws IOException {
- final Path path =
Paths.get("target/minimal-locking-repo-decrease-partitions");
- deleteRecursively(path.toFile());
- Files.createDirectories(path);
-
- final DummyRecordSerde serde = new DummyRecordSerde();
- final WriteAheadRepository<DummyRecord> writeRepo = new
MinimalLockingWriteAheadLog<>(path, 256, serde, null);
- final Collection<DummyRecord> initialRecs = writeRepo.recoverRecords();
- assertTrue(initialRecs.isEmpty());
-
- final DummyRecord record1 = new DummyRecord("1", UpdateType.CREATE);
- writeRepo.update(Collections.singleton(record1), false);
-
- for (int i=0; i < 8; i++) {
- final DummyRecord r = new DummyRecord("1", UpdateType.UPDATE);
- r.setProperty("i", String.valueOf(i));
- writeRepo.update(Collections.singleton(r), false);
- }
-
- writeRepo.shutdown();
-
- final WriteAheadRepository<DummyRecord> recoverRepo = new
MinimalLockingWriteAheadLog<>(path, 6, serde, null);
- final Collection<DummyRecord> records = recoverRepo.recoverRecords();
- final List<DummyRecord> list = new ArrayList<>(records);
- assertEquals(1, list.size());
-
- final DummyRecord recoveredRecord = list.get(0);
- assertEquals("1", recoveredRecord.getId());
- assertEquals("7",recoveredRecord.getProperty("i"));
- }
-
-
- private static class InsertThread extends Thread {
-
- private final List<List<DummyRecord>> records;
- private final WriteAheadRepository<DummyRecord> repo;
-
- public InsertThread(final int numInsertions, final int startIndex,
final WriteAheadRepository<DummyRecord> repo) {
- records = new ArrayList<>();
- for (int i = 0; i < numInsertions; i++) {
- final DummyRecord record = new DummyRecord(String.valueOf(i +
startIndex), UpdateType.CREATE);
- record.setProperty("A", "B");
- final List<DummyRecord> list = new ArrayList<>();
- list.add(record);
- records.add(list);
- }
- this.repo = repo;
- }
-
- @Override
- public void run() {
- int counter = 0;
- for (final List<DummyRecord> list : records) {
- final boolean forceSync = (++counter == records.size());
- assertDoesNotThrow(() -> repo.update(list, forceSync));
- }
- }
- }
-
-
- private static class InlineCreationInsertThread extends Thread {
- private final long iterations;
- private final WriteAheadRepository<DummyRecord> repo;
-
- public InlineCreationInsertThread(final long numInsertions, final
WriteAheadRepository<DummyRecord> repo) {
- this.iterations = numInsertions;
- this.repo = repo;
- }
-
- @Override
- public void run() {
- final List<DummyRecord> list = new ArrayList<>(1);
- list.add(null);
- final UpdateType[] updateTypes = new UpdateType[] {
UpdateType.CREATE, UpdateType.DELETE, UpdateType.UPDATE };
- final Random random = new Random();
-
- for (long i = 0; i < iterations; i++) {
- final int updateTypeIndex = random.nextInt(updateTypes.length);
- final UpdateType updateType = updateTypes[updateTypeIndex];
-
- final DummyRecord record = new DummyRecord(String.valueOf(i),
updateType);
- record.setProperty("A", "B");
- list.set(0, record);
-
- try {
- repo.update(list, false);
- } catch (final Throwable t) {
- t.printStackTrace();
- }
- }
- }
- }
-
- private void deleteRecursively(final File file) {
- final File[] children = file.listFiles();
- if (children != null) {
- for (final File child : children) {
- deleteRecursively(child);
- }
- }
-
- file.delete();
- }
-
- private long sizeOf(final File file) {
- long size = 0L;
- if (file.isDirectory()) {
- final File[] children = file.listFiles();
- if (children != null) {
- for (final File child : children) {
- size += sizeOf(child);
- }
- }
- }
-
- size += file.length();
-
- return size;
- }
-
- static class SimpleRecord {
- private long id;
- private long size;
-
- public SimpleRecord(final long id, final long size) {
- this.id = id;
- this.size = size;
- }
-
- public long getId() {
- return id;
- }
-
- public long getSize() {
- return size;
- }
- }
-}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java
index e1b78a0c82..146d990b6e 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/WriteAheadFlowFileRepository.java
@@ -31,7 +31,6 @@ import org.apache.nifi.wali.SequentialAccessWriteAheadLog;
import org.apache.nifi.wali.SnapshotCapture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.wali.MinimalLockingWriteAheadLog;
import org.wali.SyncListener;
import org.wali.WriteAheadRepository;
@@ -47,10 +46,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Optional;
import java.util.Set;
-import java.util.SortedSet;
-import java.util.TreeSet;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -94,7 +90,6 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
static final String SEQUENTIAL_ACCESS_WAL =
"org.apache.nifi.wali.SequentialAccessWriteAheadLog";
static final String ENCRYPTED_SEQUENTIAL_ACCESS_WAL =
"org.apache.nifi.wali.EncryptedSequentialAccessWriteAheadLog";
- private static final String MINIMAL_LOCKING_WALI =
"org.wali.MinimalLockingWriteAheadLog";
private static final String DEFAULT_WAL_IMPLEMENTATION =
SEQUENTIAL_ACCESS_WAL;
private static final int DEFAULT_CACHE_SIZE = 10_000_000;
@@ -110,7 +105,6 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
private final long checkpointDelayMillis;
private final List<File> flowFileRepositoryPaths = new ArrayList<>();
- private final List<File> recoveryFiles = new ArrayList<>();
private final ScheduledExecutorService checkpointExecutor;
private final int maxCharactersToCache;
@@ -176,24 +170,8 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
this.walImplementation = writeAheadLogImpl;
this.maxCharactersToCache =
nifiProperties.getIntegerProperty(FLOWFILE_REPO_CACHE_SIZE, DEFAULT_CACHE_SIZE);
- // We used to use one implementation (minimal locking) of the
write-ahead log, but we now want to use the other
- // (sequential access), we must address this. Since the
MinimalLockingWriteAheadLog supports multiple partitions,
- // we need to ensure that we recover records from all partitions, so
we build up a List of Files for the
- // recovery files.
- for (final String propertyName : nifiProperties.getPropertyKeys()) {
- if (propertyName.startsWith(FLOWFILE_REPOSITORY_DIRECTORY_PREFIX))
{
- final String dirName =
nifiProperties.getProperty(propertyName);
- recoveryFiles.add(new File(dirName));
- }
- }
-
- if (isSequentialAccessWAL(walImplementation)) {
- final String directoryName =
nifiProperties.getProperty(FLOWFILE_REPOSITORY_DIRECTORY_PREFIX);
- flowFileRepositoryPaths.add(new File(directoryName));
- } else {
- flowFileRepositoryPaths.addAll(recoveryFiles);
- }
-
+ final String directoryName =
nifiProperties.getProperty(FLOWFILE_REPOSITORY_DIRECTORY_PREFIX);
+ flowFileRepositoryPaths.add(new File(directoryName));
checkpointDelayMillis =
FormatUtils.getTimeDuration(nifiProperties.getFlowFileRepositoryCheckpointInterval(),
TimeUnit.MILLISECONDS);
@@ -207,16 +185,6 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
});
}
- /**
- * Returns true if the provided implementation is a sequential access
write ahead log (plaintext or encrypted).
- *
- * @param walImplementation the implementation to check
- * @return true if this implementation is sequential access
- */
- private static boolean isSequentialAccessWAL(String walImplementation) {
- return walImplementation.equals(SEQUENTIAL_ACCESS_WAL) ||
walImplementation.equals(ENCRYPTED_SEQUENTIAL_ACCESS_WAL);
- }
-
@Override
public void initialize(final ResourceClaimManager claimManager) throws
IOException {
final FieldCache fieldCache = new
CaffeineFieldCache(maxCharactersToCache);
@@ -251,15 +219,9 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
this.serdeFactory = serdeFactory;
// The specified implementation can be plaintext or encrypted; the
only difference is the serde factory
- if (isSequentialAccessWAL(walImplementation)) {
+ if (walImplementation.equals(SEQUENTIAL_ACCESS_WAL) ||
walImplementation.equals(ENCRYPTED_SEQUENTIAL_ACCESS_WAL)) {
// TODO: May need to instantiate ESAWAL for clarity?
wal = new
SequentialAccessWriteAheadLog<>(flowFileRepositoryPaths.get(0), serdeFactory,
this);
- } else if (walImplementation.equals(MINIMAL_LOCKING_WALI)) {
- final SortedSet<Path> paths = flowFileRepositoryPaths.stream()
- .map(File::toPath)
- .collect(Collectors.toCollection(TreeSet::new));
-
- wal = new MinimalLockingWriteAheadLog<>(paths, 1, serdeFactory,
this);
} else {
throw new IllegalStateException("Cannot create Write-Ahead Log
because the configured property '" + WRITE_AHEAD_LOG_IMPL + "' has an invalid
value of '" + walImplementation
+ "'. Please update nifi.properties to indicate a valid
value for this property.");
@@ -285,10 +247,6 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
@Override
public Map<ResourceClaim, Set<ResourceClaimReference>>
findResourceClaimReferences(final Set<ResourceClaim> resourceClaims, final
FlowFileSwapManager swapManager) {
- if (!(isSequentialAccessWAL(walImplementation))) {
- return null;
- }
-
final Map<ResourceClaim, Set<ResourceClaimReference>> references = new
HashMap<>();
final SnapshotCapture<SerializedRepositoryRecord> snapshot =
((SequentialAccessWriteAheadLog<SerializedRepositoryRecord>)
wal).captureSnapshot();
@@ -747,95 +705,6 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
}
}
- private Optional<Collection<SerializedRepositoryRecord>>
migrateFromSequentialAccessLog(final
WriteAheadRepository<SerializedRepositoryRecord> toUpdate) throws IOException {
- final String recoveryDirName =
nifiProperties.getProperty(FLOWFILE_REPOSITORY_DIRECTORY_PREFIX);
- final File recoveryDir = new File(recoveryDirName);
- if (!recoveryDir.exists()) {
- return Optional.empty();
- }
-
- final WriteAheadRepository<SerializedRepositoryRecord> recoveryWal =
new SequentialAccessWriteAheadLog<>(recoveryDir, serdeFactory, this);
- logger.info("Encountered FlowFile Repository that was written using
the Sequential Access Write Ahead Log. Will recover from this version.");
-
- final Collection<SerializedRepositoryRecord> recordList;
- try {
- recordList = recoveryWal.recoverRecords();
- } finally {
- recoveryWal.shutdown();
- }
-
- toUpdate.update(recordList, true);
-
- logger.info("Successfully recovered files from existing Write-Ahead
Log and transitioned to new Write-Ahead Log. Will not delete old files.");
-
- final File journalsDir = new File(recoveryDir, "journals");
- deleteRecursively(journalsDir);
-
- final File checkpointFile = new File(recoveryDir, "checkpoint");
- if (!checkpointFile.delete() && checkpointFile.exists()) {
- logger.warn("Failed to delete old file {}; this file should be
cleaned up manually", checkpointFile);
- }
-
- final File partialFile = new File(recoveryDir, "checkpoint.partial");
- if (!partialFile.delete() && partialFile.exists()) {
- logger.warn("Failed to delete old file {}; this file should be
cleaned up manually", partialFile);
- }
-
- return Optional.of(recordList);
- }
-
- @SuppressWarnings("deprecation")
- private Optional<Collection<SerializedRepositoryRecord>>
migrateFromMinimalLockingLog(final
WriteAheadRepository<SerializedRepositoryRecord> toUpdate) throws IOException {
- final List<File> partitionDirs = new ArrayList<>();
- for (final File recoveryFile : recoveryFiles) {
- final File[] partitions = recoveryFile.listFiles(file ->
file.getName().startsWith("partition-"));
- for (final File partition : partitions) {
- partitionDirs.add(partition);
- }
- }
-
- if (partitionDirs == null || partitionDirs.isEmpty()) {
- return Optional.empty();
- }
-
- logger.info("Encountered FlowFile Repository that was written using
the 'Minimal Locking Write-Ahead Log'. "
- + "Will recover from this version and re-write the repository
using the new version of the Write-Ahead Log.");
-
- final SortedSet<Path> paths = recoveryFiles.stream()
- .map(File::toPath)
- .collect(Collectors.toCollection(TreeSet::new));
-
- final Collection<SerializedRepositoryRecord> recordList;
- final MinimalLockingWriteAheadLog<SerializedRepositoryRecord>
minimalLockingWal = new MinimalLockingWriteAheadLog<>(paths,
partitionDirs.size(), serdeFactory, null);
- try {
- recordList = minimalLockingWal.recoverRecords();
- } finally {
- minimalLockingWal.shutdown();
- }
-
- toUpdate.update(recordList, true);
-
- // Delete the old repository
- logger.info("Successfully recovered files from existing Write-Ahead
Log and transitioned to new implementation. Will now delete old files.");
- for (final File partitionDir : partitionDirs) {
- deleteRecursively(partitionDir);
- }
-
- for (final File recoveryFile : recoveryFiles) {
- final File snapshotFile = new File(recoveryFile, "snapshot");
- if (!snapshotFile.delete() && snapshotFile.exists()) {
- logger.warn("Failed to delete old file {}; this file should be
cleaned up manually", snapshotFile);
- }
-
- final File partialFile = new File(recoveryFile,
"snapshot.partial");
- if (!partialFile.delete() && partialFile.exists()) {
- logger.warn("Failed to delete old file {}; this file should be
cleaned up manually", partialFile);
- }
- }
-
- return Optional.of(recordList);
- }
-
@Override
public Set<String> findQueuesWithFlowFiles(final FlowFileSwapManager
swapManager) throws IOException {
if (recoveredRecords == null) {
@@ -885,23 +754,6 @@ public class WriteAheadFlowFileRepository implements
FlowFileRepository, SyncLis
logger.debug("Recovered {} Swap Files: {}",
swapLocationSuffixes.size(), swapLocationSuffixes);
}
- // If we didn't recover any records from our write-ahead log, attempt
to recover records from the other implementation
- // of the write-ahead log. We do this in case the user changed the
"nifi.flowfile.repository.wal.impl" property.
- // In such a case, we still want to recover the records from the
previous FlowFile Repository and write them into the new one.
- // Since these implementations do not write to the same files, they
will not interfere with one another. If we do recover records,
- // then we will update the new WAL (with fsync()) and delete the old
repository so that we won't recover it again.
- if (recordList == null || recordList.isEmpty()) {
- if (isSequentialAccessWAL(walImplementation)) {
- // Configured to use Sequential Access WAL but it has no
records. Check if there are records in
- // a MinimalLockingWriteAheadLog that we can recover.
- recordList = migrateFromMinimalLockingLog(wal).orElse(new
ArrayList<>());
- } else {
- // Configured to use Minimal Locking WAL but it has no
records. Check if there are records in
- // a SequentialAccess Log that we can recover.
- recordList = migrateFromSequentialAccessLog(wal).orElse(new
ArrayList<>());
- }
- }
-
fieldCache.clear();
final Map<String, FlowFileQueue> queueMap = new HashMap<>();
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/local/WriteAheadLocalStateProvider.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/local/WriteAheadLocalStateProvider.java
index 682fc5a2d7..69276abbd4 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/local/WriteAheadLocalStateProvider.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/local/WriteAheadLocalStateProvider.java
@@ -43,9 +43,11 @@ 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.util.StandardValidators;
+import org.apache.nifi.wali.SequentialAccessWriteAheadLog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.wali.MinimalLockingWriteAheadLog;
+import org.wali.SerDe;
+import org.wali.SerDeFactory;
import org.wali.UpdateType;
import org.wali.WriteAheadRepository;
@@ -131,7 +133,7 @@ public class WriteAheadLocalStateProvider extends
AbstractStateProvider {
}
versionGenerator = new AtomicLong(-1L);
- writeAheadLog = new MinimalLockingWriteAheadLog<>(basePath.toPath(),
numPartitions, serde, null);
+ writeAheadLog = new SequentialAccessWriteAheadLog<>(basePath, new
SerdeFactory(serde));
final Collection<StateMapUpdate> updates =
writeAheadLog.recoverRecords();
long maxRecordVersion = EMPTY_VERSION;
@@ -319,4 +321,34 @@ public class WriteAheadLocalStateProvider extends
AbstractStateProvider {
return t;
}
}
+
+ private static class SerdeFactory implements SerDeFactory<StateMapUpdate> {
+
+ private StateMapSerDe serde;
+
+ public SerdeFactory(StateMapSerDe serde) {
+ this.serde = serde;
+ }
+
+ @Override
+ public SerDe<StateMapUpdate> createSerDe(String encodingName) {
+ return this.serde;
+ }
+
+ @Override
+ public Object getRecordIdentifier(StateMapUpdate record) {
+ return this.serde.getRecordIdentifier(record);
+ }
+
+ @Override
+ public UpdateType getUpdateType(StateMapUpdate record) {
+ return this.serde.getUpdateType(record);
+ }
+
+ @Override
+ public String getLocation(StateMapUpdate record) {
+ return this.serde.getLocation(record);
+ }
+
+ }
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
index c4f17b8962..3c7fc6b2fd 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
@@ -43,13 +43,13 @@ import org.apache.nifi.processor.FlowFileFilter;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.file.FileUtils;
+import org.apache.nifi.wali.SequentialAccessWriteAheadLog;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.mockito.stubbing.Answer;
-import org.wali.MinimalLockingWriteAheadLog;
import org.wali.WriteAheadRepository;
import java.io.File;
@@ -359,7 +359,7 @@ public class TestWriteAheadFlowFileRepository {
final ResourceClaimManager claimManager = new
StandardResourceClaimManager();
final StandardRepositoryRecordSerdeFactory serdeFactory = new
StandardRepositoryRecordSerdeFactory(claimManager);
- final WriteAheadRepository<SerializedRepositoryRecord> repo = new
MinimalLockingWriteAheadLog<>(path, numPartitions, serdeFactory, null);
+ final WriteAheadRepository<SerializedRepositoryRecord> repo = new
SequentialAccessWriteAheadLog<>(path.toFile(), serdeFactory);
final Collection<SerializedRepositoryRecord> initialRecs =
repo.recoverRecords();
assertTrue(initialRecs.isEmpty());
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java
index f36cfaa7c7..64651d069b 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java
@@ -33,10 +33,10 @@ import org.apache.nifi.remote.protocol.HandshakeProperties;
import org.apache.nifi.remote.protocol.RequestType;
import org.apache.nifi.remote.protocol.Response;
import org.apache.nifi.remote.protocol.ResponseCode;
-import org.apache.nifi.stream.io.ByteArrayInputStream;
-import org.apache.nifi.stream.io.ByteArrayOutputStream;
import org.apache.nifi.util.StringUtils;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
diff --git
a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java
b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java
index c1eebd61ee..03ed1deaf9 100644
---
a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java
+++
b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java
@@ -29,11 +29,13 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.nifi.wali.SequentialAccessWriteAheadLog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.wali.MinimalLockingWriteAheadLog;
import org.wali.SerDe;
+import org.wali.SerDeFactory;
import org.wali.UpdateType;
import org.wali.WriteAheadRepository;
@@ -48,7 +50,7 @@ public class PersistentMapCache implements MapCache {
public PersistentMapCache(final String serviceIdentifier, final File
persistencePath, final MapCache cacheToWrap) throws IOException {
try {
- wali = new MinimalLockingWriteAheadLog<>(persistencePath.toPath(),
1, new Serde(), null);
+ wali = new SequentialAccessWriteAheadLog<>(persistencePath, new
SerdeFactory());
} catch (OverlappingFileLockException ex) {
logger.error("OverlappingFileLockException thrown: Check lock
location - possible duplicate persistencePath conflict in PersistentMapCache.");
// Propagate the exception
@@ -276,4 +278,34 @@ public class PersistentMapCache implements MapCache {
return 1;
}
}
+
+ private static class SerdeFactory implements SerDeFactory<MapWaliRecord> {
+
+ private Serde serde;
+
+ public SerdeFactory() {
+ this.serde = new Serde();
+ }
+
+ @Override
+ public SerDe<MapWaliRecord> createSerDe(String encodingName) {
+ return this.serde;
+ }
+
+ @Override
+ public Object getRecordIdentifier(MapWaliRecord record) {
+ return this.serde.getRecordIdentifier(record);
+ }
+
+ @Override
+ public UpdateType getUpdateType(MapWaliRecord record) {
+ return this.serde.getUpdateType(record);
+ }
+
+ @Override
+ public String getLocation(MapWaliRecord record) {
+ return this.serde.getLocation(record);
+ }
+
+ }
}
diff --git
a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/set/PersistentSetCache.java
b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/set/PersistentSetCache.java
index c2c3a4112a..8b6cede7f3 100644
---
a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/set/PersistentSetCache.java
+++
b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/set/PersistentSetCache.java
@@ -29,8 +29,9 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
-import org.wali.MinimalLockingWriteAheadLog;
+import org.apache.nifi.wali.SequentialAccessWriteAheadLog;
import org.wali.SerDe;
+import org.wali.SerDeFactory;
import org.wali.UpdateType;
import org.wali.WriteAheadRepository;
@@ -42,7 +43,7 @@ public class PersistentSetCache implements SetCache {
private final AtomicLong modifications = new AtomicLong(0L);
public PersistentSetCache(final String serviceIdentifier, final File
persistencePath, final SetCache cacheToWrap) throws IOException {
- wali = new MinimalLockingWriteAheadLog<>(persistencePath.toPath(), 1,
new Serde(), null);
+ wali = new SequentialAccessWriteAheadLog<>(persistencePath, new
SerdeFactory());
wrapped = cacheToWrap;
}
@@ -192,4 +193,34 @@ public class PersistentSetCache implements SetCache {
return 1;
}
}
+
+ private static class SerdeFactory implements SerDeFactory<SetRecord> {
+
+ private Serde serde;
+
+ public SerdeFactory() {
+ this.serde = new Serde();
+ }
+
+ @Override
+ public SerDe<SetRecord> createSerDe(String encodingName) {
+ return this.serde;
+ }
+
+ @Override
+ public Object getRecordIdentifier(SetRecord record) {
+ return this.serde.getRecordIdentifier(record);
+ }
+
+ @Override
+ public UpdateType getUpdateType(SetRecord record) {
+ return this.serde.getUpdateType(record);
+ }
+
+ @Override
+ public String getLocation(SetRecord record) {
+ return this.serde.getLocation(record);
+ }
+
+ }
}
diff --git
a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/map/TestPersistentMapCache.java
b/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/map/TestPersistentMapCache.java
deleted file mode 100644
index 5371e42977..0000000000
---
a/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/map/TestPersistentMapCache.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.distributed.cache.server.map;
-
-import org.apache.nifi.distributed.cache.server.EvictionPolicy;
-import org.junit.jupiter.api.Test;
-
-import java.io.File;
-import java.nio.channels.OverlappingFileLockException;
-
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-public class TestPersistentMapCache {
-
- /**
- * Test OverlappingFileLockException is caught when persistent path is
duplicated.
- */
- @Test
- public void testDuplicatePersistenceDirectory() {
- assertThrows(OverlappingFileLockException.class, () -> {
- File duplicatedFilePath = new File("/tmp/path1");
- final MapCache cache = new SimpleMapCache("simpleCache", 2,
EvictionPolicy.FIFO);
- PersistentMapCache pmc1 = new PersistentMapCache("id1",
duplicatedFilePath, cache);
- PersistentMapCache pmc2 = new PersistentMapCache("id2",
duplicatedFilePath, cache);
- });
- }
-}
\ No newline at end of file
