import akka.actor.ActorSelection;
import akka.actor.Cancellable;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Optional;
import com.google.common.io.ByteSource;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.OptionalInt;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import org.eclipse.jdt.annotation.Nullable;
private final MessageSlicer appendEntriesMessageSlicer;
private Cancellable heartbeatSchedule = null;
- private Optional<SnapshotHolder> snapshotHolder = Optional.absent();
+ private Optional<SnapshotHolder> snapshotHolder = Optional.empty();
private int minReplicationCount;
protected AbstractLeader(final RaftActorContext context, final RaftState state,
@VisibleForTesting
void setSnapshotHolder(final @Nullable SnapshotHolder snapshotHolder) {
- this.snapshotHolder = Optional.fromNullable(snapshotHolder);
+ this.snapshotHolder = Optional.ofNullable(snapshotHolder);
}
@VisibleForTesting
nextSnapshotChunk.length);
int nextChunkIndex = installSnapshotState.incrementChunkIndex();
- Optional<ServerConfigurationPayload> serverConfig = Optional.absent();
+ Optional<ServerConfigurationPayload> serverConfig = Optional.empty();
if (installSnapshotState.isLastChunk(nextChunkIndex)) {
- serverConfig = Optional.fromNullable(context.getPeerServerInfo(true));
+ serverConfig = Optional.ofNullable(context.getPeerServerInfo(true));
}
sendSnapshotChunk(followerActor, followerLogInfo, nextSnapshotChunk, nextChunkIndex, serverConfig);
snapshotChunk,
chunkIndex,
installSnapshotState.getTotalChunks(),
- Optional.of(installSnapshotState.getLastChunkHashCode()),
+ OptionalInt.of(installSnapshotState.getLastChunkHashCode()),
serverConfig
).toSerializable(followerLogInfo.getRaftVersion()),
actor()
return false;
}
- private void sendOutOfSyncAppendEntriesReply(final ActorRef sender, boolean forceInstallSnapshot,
- short leaderRaftVersion) {
+ private void sendOutOfSyncAppendEntriesReply(final ActorRef sender, final boolean forceInstallSnapshot,
+ final short leaderRaftVersion) {
// We found that the log was out of sync so just send a negative reply.
final AppendEntriesReply reply = new AppendEntriesReply(context.getId(), currentTerm(), false, lastIndex(),
lastTerm(), context.getPayloadVersion(), forceInstallSnapshot, needsLeaderAddress(),
installSnapshot.getLastIncludedTerm(),
context.getTermInformation().getCurrentTerm(),
context.getTermInformation().getVotedFor(),
- installSnapshot.getServerConfig().orNull());
+ installSnapshot.getServerConfig().orElse(null));
ApplySnapshot.Callback applySnapshotCallback = new ApplySnapshot.Callback() {
@Override
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-
package org.opendaylight.controller.cluster.raft.behaviors;
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
+import static java.util.Objects.requireNonNull;
+
import com.google.common.io.ByteSource;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.util.Arrays;
+import java.util.OptionalInt;
import org.opendaylight.controller.cluster.io.FileBackedOutputStream;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
import org.slf4j.Logger;
private int lastChunkHashCode = LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE;
private long count;
- SnapshotTracker(Logger log, int totalChunks, String leaderId, RaftActorContext context) {
+ SnapshotTracker(final Logger log, final int totalChunks, final String leaderId, final RaftActorContext context) {
this.log = log;
this.totalChunks = totalChunks;
- this.leaderId = Preconditions.checkNotNull(leaderId);
+ this.leaderId = requireNonNull(leaderId);
fileBackedStream = context.getFileBackedOutputStreamFactory().newInstance();
bufferedStream = new BufferedOutputStream(fileBackedStream);
}
* @return true if this is the last chunk is received
* @throws InvalidChunkException if the chunk index is invalid or out of order
*/
- boolean addChunk(int chunkIndex, byte[] chunk, Optional<Integer> maybeLastChunkHashCode)
+ boolean addChunk(final int chunkIndex, final byte[] chunk, final OptionalInt maybeLastChunkHashCode)
throws InvalidChunkException, IOException {
log.debug("addChunk: chunkIndex={}, lastChunkIndex={}, collectedChunks.size={}, lastChunkHashCode={}",
chunkIndex, lastChunkIndex, count, this.lastChunkHashCode);
throw new InvalidChunkException("Expected chunkIndex " + (lastChunkIndex + 1) + " got " + chunkIndex);
}
- if (maybeLastChunkHashCode.isPresent() && maybeLastChunkHashCode.get() != this.lastChunkHashCode) {
+ if (maybeLastChunkHashCode.isPresent() && maybeLastChunkHashCode.getAsInt() != this.lastChunkHashCode) {
throw new InvalidChunkException("The hash code of the recorded last chunk does not match "
+ "the senders hash code, expected " + this.lastChunkHashCode + " was "
- + maybeLastChunkHashCode.get());
+ + maybeLastChunkHashCode.getAsInt());
}
bufferedStream.write(chunk);
public static class InvalidChunkException extends IOException {
private static final long serialVersionUID = 1L;
- InvalidChunkException(String message) {
+ InvalidChunkException(final String message) {
super(message);
}
}
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-
package org.opendaylight.controller.cluster.raft.messages;
-import com.google.common.base.Optional;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
+import java.util.Optional;
+import java.util.OptionalInt;
import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
/**
private final byte[] data;
private final int chunkIndex;
private final int totalChunks;
- private final Optional<Integer> lastChunkHashCode;
+ @SuppressFBWarnings(value = "SE_BAD_FIELD", justification = "Handled via writeReplace()")
+ private final OptionalInt lastChunkHashCode;
+ @SuppressFBWarnings(value = "SE_BAD_FIELD", justification = "Handled via writeReplace()")
private final Optional<ServerConfigurationPayload> serverConfig;
@SuppressFBWarnings(value = "EI_EXPOSE_REP2", justification = "Stores a reference to an externally mutable byte[] "
+ "object but this is OK since this class is merely a DTO and does not process byte[] internally. "
+ "Also it would be inefficient to create a copy as the byte[] could be large.")
- public InstallSnapshot(long term, String leaderId, long lastIncludedIndex, long lastIncludedTerm, byte[] data,
- int chunkIndex, int totalChunks, Optional<Integer> lastChunkHashCode,
- Optional<ServerConfigurationPayload> serverConfig) {
+ public InstallSnapshot(final long term, final String leaderId, final long lastIncludedIndex,
+ final long lastIncludedTerm, final byte[] data, final int chunkIndex, final int totalChunks,
+ final OptionalInt lastChunkHashCode, final Optional<ServerConfigurationPayload> serverConfig) {
super(term);
this.leaderId = leaderId;
this.lastIncludedIndex = lastIncludedIndex;
this.serverConfig = serverConfig;
}
- public InstallSnapshot(long term, String leaderId, long lastIncludedIndex,
- long lastIncludedTerm, byte[] data, int chunkIndex, int totalChunks) {
- this(term, leaderId, lastIncludedIndex, lastIncludedTerm, data, chunkIndex, totalChunks,
- Optional.<Integer>absent(), Optional.<ServerConfigurationPayload>absent());
+ public InstallSnapshot(final long term, final String leaderId, final long lastIncludedIndex,
+ final long lastIncludedTerm, final byte[] data, final int chunkIndex,
+ final int totalChunks) {
+ this(term, leaderId, lastIncludedIndex, lastIncludedTerm, data, chunkIndex, totalChunks, OptionalInt.empty(),
+ Optional.empty());
}
public String getLeaderId() {
return totalChunks;
}
- public Optional<Integer> getLastChunkHashCode() {
+ public OptionalInt getLastChunkHashCode() {
return lastChunkHashCode;
}
}
- public <T> Object toSerializable(short version) {
+ public <T> Object toSerializable(final short version) {
return this;
}
return "InstallSnapshot [term=" + getTerm() + ", leaderId=" + leaderId + ", lastIncludedIndex="
+ lastIncludedIndex + ", lastIncludedTerm=" + lastIncludedTerm + ", datasize=" + data.length
+ ", Chunk=" + chunkIndex + "/" + totalChunks + ", lastChunkHashCode=" + lastChunkHashCode
- + ", serverConfig=" + serverConfig.orNull() + "]";
+ + ", serverConfig=" + serverConfig.orElse(null) + "]";
}
private Object writeReplace() {
public Proxy() {
}
- Proxy(InstallSnapshot installSnapshot) {
+ Proxy(final InstallSnapshot installSnapshot) {
this.installSnapshot = installSnapshot;
}
@Override
- public void writeExternal(ObjectOutput out) throws IOException {
+ public void writeExternal(final ObjectOutput out) throws IOException {
out.writeLong(installSnapshot.getTerm());
out.writeObject(installSnapshot.leaderId);
out.writeLong(installSnapshot.lastIncludedIndex);
out.writeByte(installSnapshot.lastChunkHashCode.isPresent() ? 1 : 0);
if (installSnapshot.lastChunkHashCode.isPresent()) {
- out.writeInt(installSnapshot.lastChunkHashCode.get().intValue());
+ out.writeInt(installSnapshot.lastChunkHashCode.getAsInt());
}
out.writeByte(installSnapshot.serverConfig.isPresent() ? 1 : 0);
}
@Override
- public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ public void readExternal(final ObjectInput in) throws IOException, ClassNotFoundException {
long term = in.readLong();
String leaderId = (String) in.readObject();
long lastIncludedIndex = in.readLong();
int chunkIndex = in.readInt();
int totalChunks = in.readInt();
- Optional<Integer> lastChunkHashCode = Optional.absent();
- boolean chunkHashCodePresent = in.readByte() == 1;
- if (chunkHashCodePresent) {
- lastChunkHashCode = Optional.of(in.readInt());
- }
-
- Optional<ServerConfigurationPayload> serverConfig = Optional.absent();
- boolean serverConfigPresent = in.readByte() == 1;
- if (serverConfigPresent) {
- serverConfig = Optional.of((ServerConfigurationPayload)in.readObject());
- }
+ OptionalInt lastChunkHashCode = in.readByte() == 1 ? OptionalInt.of(in.readInt()) : OptionalInt.empty();
+ Optional<ServerConfigurationPayload> serverConfig = in.readByte() == 1
+ ? Optional.of((ServerConfigurationPayload)in.readObject()) : Optional.empty();
byte[] data = (byte[])in.readObject();
assertEquals(1, installSnapshot.getChunkIndex());
assertEquals(3, installSnapshot.getTotalChunks());
assertEquals(LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE,
- installSnapshot.getLastChunkHashCode().get().intValue());
+ installSnapshot.getLastChunkHashCode().getAsInt());
final int hashCode = Arrays.hashCode(installSnapshot.getData());
assertEquals(2, installSnapshot.getChunkIndex());
assertEquals(3, installSnapshot.getTotalChunks());
- assertEquals(hashCode, installSnapshot.getLastChunkHashCode().get().intValue());
+ assertEquals(hashCode, installSnapshot.getLastChunkHashCode().getAsInt());
}
@Test
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-
package org.opendaylight.controller.cluster.raft.behaviors;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.verify;
import akka.protobuf.ByteString;
-import com.google.common.base.Optional;
import com.google.common.io.ByteSource;
import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
+import java.util.OptionalInt;
import org.apache.commons.lang3.SerializationUtils;
import org.junit.Before;
import org.junit.Test;
@Test
public void testAddChunks() throws IOException {
try (SnapshotTracker tracker = new SnapshotTracker(LOG, 3, "leader", mockContext)) {
- tracker.addChunk(1, chunk1, Optional.of(LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE));
- tracker.addChunk(2, chunk2, Optional.of(Arrays.hashCode(chunk1)));
- tracker.addChunk(3, chunk3, Optional.of(Arrays.hashCode(chunk2)));
+ tracker.addChunk(1, chunk1, OptionalInt.of(LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE));
+ tracker.addChunk(2, chunk2, OptionalInt.of(Arrays.hashCode(chunk1)));
+ tracker.addChunk(3, chunk3, OptionalInt.of(Arrays.hashCode(chunk2)));
ByteSource snapshotBytes = tracker.getSnapshotBytes();
assertEquals("Deserialized", data, SerializationUtils.deserialize(snapshotBytes.read()));
@Test(expected = SnapshotTracker.InvalidChunkException.class)
public void testAddChunkWhenAlreadySealed() throws IOException {
try (SnapshotTracker tracker = new SnapshotTracker(LOG, 2, "leader", mockContext)) {
- tracker.addChunk(1, chunk1, Optional.<Integer>absent());
- tracker.addChunk(2, chunk2, Optional.<Integer>absent());
- tracker.addChunk(3, chunk3, Optional.<Integer>absent());
+ tracker.addChunk(1, chunk1, OptionalInt.empty());
+ tracker.addChunk(2, chunk2, OptionalInt.empty());
+ tracker.addChunk(3, chunk3, OptionalInt.empty());
}
}
@Test(expected = SnapshotTracker.InvalidChunkException.class)
public void testInvalidFirstChunkIndex() throws IOException {
try (SnapshotTracker tracker = new SnapshotTracker(LOG, 2, "leader", mockContext)) {
- tracker.addChunk(LeaderInstallSnapshotState.FIRST_CHUNK_INDEX - 1, chunk1, Optional.<Integer>absent());
+ tracker.addChunk(LeaderInstallSnapshotState.FIRST_CHUNK_INDEX - 1, chunk1, OptionalInt.empty());
}
}
@Test(expected = SnapshotTracker.InvalidChunkException.class)
public void testOutOfSequenceChunk() throws IOException {
try (SnapshotTracker tracker = new SnapshotTracker(LOG, 2, "leader", mockContext)) {
- tracker.addChunk(1, chunk1, Optional.<Integer>absent());
- tracker.addChunk(3, chunk3, Optional.<Integer>absent());
+ tracker.addChunk(1, chunk1, OptionalInt.empty());
+ tracker.addChunk(3, chunk3, OptionalInt.empty());
}
}
@Test(expected = SnapshotTracker.InvalidChunkException.class)
public void testInvalidLastChunkHashCode() throws IOException {
try (SnapshotTracker tracker = new SnapshotTracker(LOG, 2, "leader", mockContext)) {
- tracker.addChunk(1, chunk1, Optional.of(LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE));
- tracker.addChunk(2, chunk2, Optional.of(1));
+ tracker.addChunk(1, chunk1, OptionalInt.of(LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE));
+ tracker.addChunk(2, chunk2, OptionalInt.of(1));
}
}
@Test(expected = IllegalStateException.class)
public void testGetSnapshotBytesWhenNotSealed() throws IOException {
try (SnapshotTracker tracker = new SnapshotTracker(LOG, 2, "leader", mockContext)) {
- tracker.addChunk(1, chunk1, Optional.<Integer>absent());
+ tracker.addChunk(1, chunk1, OptionalInt.empty());
tracker.getSnapshotBytes();
}
}
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
-import com.google.common.base.Optional;
import java.io.Serializable;
import java.util.Arrays;
+import java.util.Optional;
+import java.util.OptionalInt;
import org.apache.commons.lang.SerializationUtils;
import org.junit.Test;
import org.opendaylight.controller.cluster.raft.RaftVersions;
ServerConfigurationPayload serverConfig = new ServerConfigurationPayload(Arrays.asList(
new ServerInfo("leader", true), new ServerInfo("follower", false)));
- InstallSnapshot expected = new InstallSnapshot(3L, "leaderId", 11L, 2L, data, 5, 6,
- Optional.<Integer>of(54321), Optional.of(serverConfig));
+ InstallSnapshot expected = new InstallSnapshot(3L, "leaderId", 11L, 2L, data, 5, 6, OptionalInt.of(54321),
+ Optional.of(serverConfig));
Object serialized = expected.toSerializable(RaftVersions.CURRENT_VERSION);
assertEquals("Serialized type", InstallSnapshot.class, serialized.getClass());
verifyInstallSnapshot(expected, actual);
}
- private static void verifyInstallSnapshot(InstallSnapshot expected, InstallSnapshot actual) {
+ private static void verifyInstallSnapshot(final InstallSnapshot expected, final InstallSnapshot actual) {
assertEquals("getTerm", expected.getTerm(), actual.getTerm());
assertEquals("getChunkIndex", expected.getChunkIndex(), actual.getChunkIndex());
assertEquals("getTotalChunks", expected.getTotalChunks(), actual.getTotalChunks());
assertEquals("getLastChunkHashCode present", expected.getLastChunkHashCode().isPresent(),
actual.getLastChunkHashCode().isPresent());
if (expected.getLastChunkHashCode().isPresent()) {
- assertEquals("getLastChunkHashCode", expected.getLastChunkHashCode().get(),
- actual.getLastChunkHashCode().get());
+ assertEquals("getLastChunkHashCode", expected.getLastChunkHashCode(),
+ actual.getLastChunkHashCode());
}
assertEquals("getServerConfig present", expected.getServerConfig().isPresent(),
Code Review / controller.git / commitdiff