*/
package org.opendaylight.controller.cluster.persistence;
+import static com.google.common.base.Preconditions.checkArgument;
+
import akka.actor.ExtendedActorSystem;
import akka.dispatch.Futures;
import akka.persistence.SelectedSnapshot;
import akka.persistence.serialization.Snapshot;
import akka.persistence.serialization.SnapshotSerializer;
import akka.persistence.snapshot.japi.SnapshotStore;
+import akka.serialization.JavaSerializer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.io.ByteStreams;
import com.typesafe.config.Config;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.util.Deque;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.Callable;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import javax.annotation.Nullable;
+import org.eclipse.jdt.annotation.Nullable;
+import org.opendaylight.controller.cluster.io.InputOutputStreamFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.ExecutionContext;
*/
public class LocalSnapshotStore extends SnapshotStore {
private static final Logger LOG = LoggerFactory.getLogger(LocalSnapshotStore.class);
-
private static final int PERSISTENCE_ID_START_INDEX = "snapshot-".length();
+ private final InputOutputStreamFactory streamFactory;
private final ExecutionContext executionContext;
private final int maxLoadAttempts;
private final File snapshotDir;
this.executionContext = context().system().dispatchers().lookup(config.getString("stream-dispatcher"));
snapshotDir = new File(config.getString("dir"));
- int localMaxLoadAttempts = config.getInt("max-load-attempts");
+ final int localMaxLoadAttempts = config.getInt("max-load-attempts");
maxLoadAttempts = localMaxLoadAttempts > 0 ? localMaxLoadAttempts : 1;
+ if (config.getBoolean("use-lz4-compression")) {
+ final String size = config.getString("lz4-blocksize");
+ streamFactory = InputOutputStreamFactory.lz4(size);
+ LOG.debug("Using LZ4 Input/Output Stream, blocksize: {}", size);
+ } else {
+ streamFactory = InputOutputStreamFactory.simple();
+ LOG.debug("Using plain Input/Output Stream");
+ }
+
LOG.debug("LocalSnapshotStore ctor: snapshotDir: {}, maxLoadAttempts: {}", snapshotDir, maxLoadAttempts);
}
}
private Object deserialize(final File file) throws IOException {
- try (ObjectInputStream in = new ObjectInputStream(new FileInputStream(file))) {
- return in.readObject();
- } catch (ClassNotFoundException e) {
- throw new IOException("Error loading snapshot file " + file, e);
- } catch (IOException e) {
- LOG.debug("Error loading snapshot file {}", file, e);
-
- return tryDeserializeAkkaSnapshot(file);
- }
+ return JavaSerializer.currentSystem().withValue((ExtendedActorSystem) context().system(),
+ (Callable<Object>) () -> {
+ try (ObjectInputStream in = new ObjectInputStream(streamFactory.createInputStream(file))) {
+ return in.readObject();
+ } catch (ClassNotFoundException e) {
+ throw new IOException("Error loading snapshot file " + file, e);
+ } catch (IOException e) {
+ LOG.debug("Error loading snapshot file {}", file, e);
+ return tryDeserializeAkkaSnapshot(file);
+ }
+ });
}
private Object tryDeserializeAkkaSnapshot(final File file) throws IOException {
LOG.debug("Saving to temp file: {}", temp);
- try (ObjectOutputStream out = new ObjectOutputStream(new FileOutputStream(temp))) {
+ try (ObjectOutputStream out = new ObjectOutputStream(streamFactory.createOutputStream(temp))) {
out.writeObject(snapshot);
} catch (IOException e) {
LOG.error("Error saving snapshot file {}. Deleting file..", temp, e);
LOG.debug("Deleting files: {}", files);
- files.forEach(File::delete);
+ files.forEach(file -> {
+ try {
+ Files.delete(file.toPath());
+ } catch (IOException | SecurityException e) {
+ LOG.error("Unable to delete snapshot file: {}, persistenceId: {} ", file, persistenceId);
+ }
+ });
return null;
}
LOG.debug("Deleting files: {}", files);
- files.forEach(File::delete);
+ files.forEach(file -> {
+ try {
+ Files.delete(file.toPath());
+ } catch (IOException | SecurityException e) {
+ LOG.error("Unable to delete snapshot file: {}", file);
+ }
+ });
return null;
}
private Collection<File> getSnapshotFiles(final String persistenceId) {
String encodedPersistenceId = encode(persistenceId);
- File[] files = snapshotDir.listFiles((FilenameFilter) (dir, name) -> {
+ File[] files = snapshotDir.listFiles((dir, name) -> {
int persistenceIdEndIndex = name.lastIndexOf('-', name.lastIndexOf('-') - 1);
return PERSISTENCE_ID_START_INDEX + encodedPersistenceId.length() == persistenceIdEndIndex
&& name.startsWith(encodedPersistenceId, PERSISTENCE_ID_START_INDEX) && !name.endsWith(".tmp");
.filter(criteria::matches).collect(Collectors.toList());
}
- private static Stream<SnapshotMetadata> toStream(@Nullable final SnapshotMetadata md) {
+ private static Stream<SnapshotMetadata> toStream(final @Nullable SnapshotMetadata md) {
return md != null ? Stream.of(md) : Stream.empty();
}
- @Nullable
- private static SnapshotMetadata extractMetadata(final File file) {
+ private static @Nullable SnapshotMetadata extractMetadata(final File file) {
String name = file.getName();
int sequenceNumberEndIndex = name.lastIndexOf('-');
int persistenceIdEndIndex = name.lastIndexOf('-', sequenceNumberEndIndex - 1);
@VisibleForTesting
static int compare(final SnapshotMetadata m1, final SnapshotMetadata m2) {
- return (int) (!m1.persistenceId().equals(m2.persistenceId())
- ? m1.persistenceId().compareTo(m2.persistenceId()) :
- m1.sequenceNr() != m2.sequenceNr() ? m1.sequenceNr() - m2.sequenceNr() :
- m1.timestamp() != m2.timestamp() ? m1.timestamp() - m2.timestamp() : 0);
+ checkArgument(m1.persistenceId().equals(m2.persistenceId()),
+ "Persistence id does not match. id1: %s, id2: %s", m1.persistenceId(), m2.persistenceId());
+ final int cmp = Long.compare(m1.timestamp(), m2.timestamp());
+ return cmp != 0 ? cmp : Long.compare(m1.sequenceNr(), m2.sequenceNr());
}
}
Code Review / controller.git / blobdiff