public MetaStore(RaftStorage storage, Serializer serializer) {
  this.serializer = checkNotNull(serializer, "serializer cannot be null");

  if (!(storage.directory().isDirectory() || storage.directory().mkdirs())) {
    throw new IllegalArgumentException(String.format("Can't create storage directory [%s].", storage.directory()));
  }

  // Note that for raft safety, irrespective of the storage level, <term, vote> metadata is always persisted on disk.
  File metaFile = new File(storage.directory(), String.format("%s.meta", storage.prefix()));
  metadataBuffer = FileBuffer.allocate(metaFile, 12);

  if (storage.storageLevel() == StorageLevel.MEMORY) {
    configurationBuffer = HeapBuffer.allocate(32);
  } else {
    File confFile = new File(storage.directory(), String.format("%s.conf", storage.prefix()));
    configurationBuffer = FileBuffer.allocate(confFile, 32);
  }
}
 

/**
 * Creates a Raft server.
 */
private RaftServer createServer(RaftMember member) {
  RaftServerProtocol protocol;
  if (USE_NETTY) {
    Address address = Address.from(++port);
    MessagingService messagingManager = new NettyMessagingService("test", address, new MessagingConfig()).start().join();
    messagingServices.add(messagingManager);
    addressMap.put(member.memberId(), address);
    protocol = new RaftServerMessagingProtocol(messagingManager, PROTOCOL_SERIALIZER, addressMap::get);
  } else {
    protocol = protocolFactory.newServerProtocol(member.memberId());
  }

  RaftServer.Builder builder = RaftServer.builder(member.memberId())
      .withProtocol(protocol)
      .withStorage(RaftStorage.builder()
          .withStorageLevel(StorageLevel.DISK)
          .withDirectory(new File(String.format("target/fuzz-logs/%s", member.memberId())))
          .withNamespace(STORAGE_NAMESPACE)
          .withMaxSegmentSize(1024 * 1024)
          .build());

  RaftServer server = builder.build();
  servers.add(server);
  return server;
}
 

public SegmentedJournal(
    String name,
    StorageLevel storageLevel,
    File directory,
    Namespace namespace,
    int maxSegmentSize,
    int maxEntrySize,
    int maxEntriesPerSegment,
    double indexDensity,
    boolean flushOnCommit) {
  this.name = checkNotNull(name, "name cannot be null");
  this.storageLevel = checkNotNull(storageLevel, "storageLevel cannot be null");
  this.directory = checkNotNull(directory, "directory cannot be null");
  this.namespace = checkNotNull(namespace, "namespace cannot be null");
  this.maxSegmentSize = maxSegmentSize;
  this.maxEntrySize = maxEntrySize;
  this.maxEntriesPerSegment = maxEntriesPerSegment;
  this.indexDensity = indexDensity;
  this.flushOnCommit = flushOnCommit;
  open();
  this.writer = openWriter();
}
 

@Override
public Namespace namespace() {
  return Namespace.builder()
      .nextId(Namespaces.BEGIN_USER_CUSTOM_ID + 100)
      .register(RaftPartitionGroupConfig.class)
      .register(RaftStorageConfig.class)
      .register(RaftCompactionConfig.class)
      .register(StorageLevel.class)
      .build();
}
 

/**
 * Returns a boolean indicating whether the node is running out of memory.
 */
private boolean isRunningOutOfMemory() {
  StorageLevel level = raft.getStorage().storageLevel();
  if (level == StorageLevel.MEMORY || level == StorageLevel.MAPPED) {
    long freeMemory = raft.getStorage().statistics().getFreeMemory();
    long totalMemory = raft.getStorage().statistics().getTotalMemory();
    if (freeMemory > 0 && totalMemory > 0) {
      return freeMemory / (double) totalMemory < raft.getStorage().freeMemoryBuffer();
    }
  }
  return false;
}
 

private RaftStorage(
    String prefix,
    StorageLevel storageLevel,
    File directory,
    Namespace namespace,
    int maxSegmentSize,
    int maxEntrySize,
    int maxEntriesPerSegment,
    boolean dynamicCompaction,
    double freeDiskBuffer,
    double freeMemoryBuffer,
    boolean flushOnCommit,
    boolean retainStaleSnapshots) {
  this.prefix = prefix;
  this.storageLevel = storageLevel;
  this.directory = directory;
  this.namespace = namespace;
  this.maxSegmentSize = maxSegmentSize;
  this.maxEntrySize = maxEntrySize;
  this.maxEntriesPerSegment = maxEntriesPerSegment;
  this.dynamicCompaction = dynamicCompaction;
  this.freeDiskBuffer = freeDiskBuffer;
  this.freeMemoryBuffer = freeMemoryBuffer;
  this.flushOnCommit = flushOnCommit;
  this.retainStaleSnapshots = retainStaleSnapshots;
  this.statistics = new StorageStatistics(directory);
  directory.mkdirs();
}
 

/**
 * Opens the snapshot manager.
 */
private void open() {
  // load persisted snapshots only if storage level is persistent
  if (storage.storageLevel() != StorageLevel.MEMORY) {
    for (Snapshot snapshot : loadSnapshots()) {
      completeSnapshot(snapshot);
    }
  }
}
 

/**
 * Creates a new snapshot.
 *
 * @param index     The snapshot index.
 * @param timestamp The snapshot timestamp.
 * @return The snapshot.
 */
public Snapshot newSnapshot(long index, WallClockTimestamp timestamp) {
  SnapshotDescriptor descriptor = SnapshotDescriptor.builder()
      .withIndex(index)
      .withTimestamp(timestamp.unixTimestamp())
      .build();

  if (storage.storageLevel() == StorageLevel.MEMORY) {
    return createMemorySnapshot(descriptor);
  } else {
    return createDiskSnapshot(descriptor);
  }
}
 

private RaftStorage createStorage(MemberId memberId, Function<RaftStorage.Builder, RaftStorage.Builder> configurator) {
  final RaftStorage.Builder defaults =
      RaftStorage.builder()
          .withStorageLevel(StorageLevel.DISK)
          .withDirectory(new File(String.format("target/test-logs/%s", memberId)))
          .withMaxEntriesPerSegment(10)
          .withMaxSegmentSize(1024 * 10)
          .withNamespace(NAMESPACE);
  return configurator.apply(defaults).build();
}
 

/**
 * Returns a new snapshot store.
 */
protected SnapshotStore createSnapshotStore() {
  RaftStorage storage = RaftStorage.builder()
      .withPrefix("test")
      .withStorageLevel(StorageLevel.MEMORY)
      .build();
  return new SnapshotStore(storage);
}
 

/**
 * Returns a new snapshot store.
 */
protected SnapshotStore createSnapshotStore() {
  RaftStorage storage = RaftStorage.builder()
      .withPrefix("test")
      .withDirectory(new File(String.format("target/test-logs/%s", testId)))
      .withStorageLevel(StorageLevel.DISK)
      .build();
  return new SnapshotStore(storage);
}
 

@Override
public Namespace namespace() {
  return Namespace.builder()
      .nextId(Namespaces.BEGIN_USER_CUSTOM_ID + 300)
      .register(LogPartitionGroupConfig.class)
      .register(LogStorageConfig.class)
      .register(LogCompactionConfig.class)
      .register(MemorySize.class)
      .register(StorageLevel.class)
      .build();
}
 

public JournalSegment(
    JournalSegmentFile file,
    JournalSegmentDescriptor descriptor,
    StorageLevel storageLevel,
    int maxEntrySize,
    double indexDensity,
    Namespace namespace) {
  this.file = file;
  this.descriptor = descriptor;
  this.storageLevel = storageLevel;
  this.maxEntrySize = maxEntrySize;
  this.index = new SparseJournalIndex(indexDensity);
  this.namespace = namespace;
  this.writer = new MappableJournalSegmentWriter<>(openChannel(file.file()), this, maxEntrySize, index, namespace);
}
 

/**
 * Maps the log segment into memory.
 */
private void map() {
  if (storageLevel == StorageLevel.MAPPED) {
    MappedByteBuffer buffer = writer.map();
    readers.forEach(reader -> reader.map(buffer));
  }
}
 

/**
 * Unmaps the log segment from memory.
 */
private void unmap() {
  if (storageLevel == StorageLevel.MAPPED) {
    writer.unmap();
    readers.forEach(reader -> reader.unmap());
  }
}
 

/**
 * Creates a Raft server.
 */
private RaftServer createServer(Member member, List<Node> members) {
  RaftServerProtocol protocol;
  ManagedMessagingService messagingService;
  if (USE_NETTY) {
    messagingService = (ManagedMessagingService) new NettyMessagingService("test", member.address(), new MessagingConfig())
        .start()
        .join();
    messagingServices.add(messagingService);
    protocol = new RaftServerMessagingProtocol(messagingService, PROTOCOL_SERIALIZER, addressMap::get);
  } else {
    protocol = protocolFactory.newServerProtocol(member.id());
  }

  BootstrapService bootstrapService = new BootstrapService() {
    @Override
    public MessagingService getMessagingService() {
      return messagingService;
    }

    @Override
    public UnicastService getUnicastService() {
      return new UnicastServiceAdapter();
    }

    @Override
    public BroadcastService getBroadcastService() {
      return new BroadcastServiceAdapter();
    }
  };

  RaftServer.Builder builder = RaftServer.builder(member.id())
      .withProtocol(protocol)
      .withThreadModel(ThreadModel.SHARED_THREAD_POOL)
      .withMembershipService(new DefaultClusterMembershipService(
          member,
          Version.from("1.0.0"),
          new DefaultNodeDiscoveryService(bootstrapService, member, new BootstrapDiscoveryProvider(members)),
          bootstrapService,
          new HeartbeatMembershipProtocol(new HeartbeatMembershipProtocolConfig())))
      .withStorage(RaftStorage.builder()
          .withStorageLevel(StorageLevel.DISK)
          .withDirectory(new File(String.format("target/perf-logs/%s", member.id())))
          .withNamespace(STORAGE_NAMESPACE)
          .withMaxSegmentSize(1024 * 1024 * 64)
          .withDynamicCompaction()
          .withFlushOnCommit(false)
          .build());

  RaftServer server = builder.build();
  servers.add(server);
  return server;
}
 

@Override
protected StorageLevel storageLevel() {
  return StorageLevel.DISK;
}
 

@Override
protected StorageLevel storageLevel() {
  return StorageLevel.MAPPED;
}
 

@Override
protected StorageLevel storageLevel() {
  return StorageLevel.MEMORY;
}
 

/**
 * Returns the partition storage level.
 *
 * @return the partition storage level
 */
public StorageLevel getLevel() {
  return level;
}
 

/**
 * Sets the partition storage level.
 *
 * @param storageLevel the partition storage level
 * @return the Raft partition group configuration
 */
public RaftStorageConfig setLevel(StorageLevel storageLevel) {
  this.level = checkNotNull(storageLevel);
  return this;
}
 

/**
 * Sets the storage level.
 *
 * @param storageLevel the storage level
 * @return the Raft partition group builder
 */
public Builder withStorageLevel(StorageLevel storageLevel) {
  config.getStorageConfig().setLevel(storageLevel);
  return this;
}
 

/**
 * Sets the log storage level, returning the builder for method chaining.
 * <p>
 * The storage level indicates how individual entries should be persisted in the journal.
 *
 * @param storageLevel The log storage level.
 * @return The storage builder.
 */
public Builder withStorageLevel(StorageLevel storageLevel) {
  journalBuilder.withStorageLevel(storageLevel);
  return this;
}
 

/**
 * Returns the storage level.
 * <p>
 * The storage level dictates how entries within individual log {@link RaftLog}s should be stored.
 *
 * @return The storage level.
 */
public StorageLevel storageLevel() {
  return storageLevel;
}
 

/**
 * Sets the log storage level, returning the builder for method chaining.
 * <p>
 * The storage level indicates how individual {@link RaftLogEntry entries}
 * should be persisted in the log.
 *
 * @param storageLevel The log storage level.
 * @return The storage builder.
 */
public Builder withStorageLevel(StorageLevel storageLevel) {
  this.storageLevel = checkNotNull(storageLevel, "storageLevel");
  return this;
}
 

/**
 * Sets the log storage level, returning the builder for method chaining.
 * <p>
 * The storage level indicates how individual entries should be persisted in the journal.
 *
 * @param storageLevel The log storage level.
 * @return The storage builder.
 */
public Builder withStorageLevel(StorageLevel storageLevel) {
  this.storageLevel = checkNotNull(storageLevel, "storageLevel cannot be null");
  return this;
}
 

/**
 * Sets the storage level.
 *
 * @param storageLevel the storage level
 * @return the Raft partition group builder
 */
public Builder withStorageLevel(StorageLevel storageLevel) {
  config.getStorageConfig().setLevel(storageLevel);
  return this;
}
 

/**
 * Returns the partition storage level.
 *
 * @return the partition storage level
 */
public StorageLevel getLevel() {
  return level;
}
 

/**
 * Sets the partition storage level.
 *
 * @param storageLevel the partition storage level
 * @return the log partition group configuration
 */
public LogStorageConfig setLevel(StorageLevel storageLevel) {
  this.level = checkNotNull(storageLevel);
  return this;
}
 

/**
 * Returns the storage level.
 * <p>
 * The storage level dictates how entries within individual journal segments should be stored.
 *
 * @return The storage level.
 */
public StorageLevel storageLevel() {
  return storageLevel;
}