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
}
@Override
- protected RaftActorBehavior handleRequestVoteReply(final ActorRef sender,
- final RequestVoteReply requestVoteReply) {
+ protected RaftActorBehavior handleRequestVoteReply(final ActorRef sender, final RequestVoteReply requestVoteReply) {
return this;
}
- protected void beforeSendHeartbeat(){}
+ protected void beforeSendHeartbeat() {
+ // No-op
+ }
@Override
public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) {
return;
}
+ installSnapshotState.resetChunkTimer();
followerLogInformation.markFollowerActive();
if (installSnapshotState.getChunkIndex() == reply.getChunkIndex()) {
LeaderInstallSnapshotState installSnapshotState = followerLogInformation.getInstallSnapshotState();
if (installSnapshotState != null) {
+
// if install snapshot is in process , then sent next chunk if possible
- if (isFollowerActive && installSnapshotState.canSendNextChunk()) {
- sendSnapshotChunk(followerActor, followerLogInformation);
- } else if (sendHeartbeat) {
+ if (isFollowerActive) {
+ // 30 seconds with default settings, can be modified via heartbeat or election timeout factor
+ FiniteDuration snapshotReplyTimeout = context.getConfigParams().getHeartBeatInterval()
+ .$times(context.getConfigParams().getElectionTimeoutFactor() * 3);
+
+ if (installSnapshotState.isChunkTimedOut(snapshotReplyTimeout)) {
+ sendAppendEntries = !resendSnapshotChunk(followerActor, followerLogInformation);
+ } else if (installSnapshotState.canSendNextChunk()) {
+ sendSnapshotChunk(followerActor, followerLogInformation);
+ }
+ } else if (sendHeartbeat || followerLogInformation.hasStaleCommitIndex(context.getCommitIndex())) {
// we send a heartbeat even if we have not received a reply for the last chunk
sendAppendEntries = true;
}
log.debug("{}: sendAppendEntries: {} is present for follower {}", logName(),
followerNextIndex, followerId);
- if (followerLogInformation.okToReplicate()) {
+ if (followerLogInformation.okToReplicate(context.getCommitIndex())) {
entries = getEntriesToSend(followerLogInformation, followerActor);
sendAppendEntries = true;
}
context.getReplicatedLog().size());
}
- } else if (sendHeartbeat) {
+ } else if (sendHeartbeat || followerLogInformation.hasStaleCommitIndex(context.getCommitIndex())) {
// we send an AppendEntries, even if the follower is inactive
// in-order to update the followers timestamp, in case it becomes active again
sendAppendEntries = true;
appendEntries);
}
+ followerLogInformation.setSentCommitIndex(leaderCommitIndex);
followerActor.tell(appendEntries, actor());
}
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));
}
- followerActor.tell(
- new InstallSnapshot(currentTerm(), context.getId(),
+ sendSnapshotChunk(followerActor, followerLogInfo, nextSnapshotChunk, nextChunkIndex, serverConfig);
+
+ log.debug("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}", logName(), followerActor.path(),
+ installSnapshotState.getChunkIndex(), installSnapshotState.getTotalChunks());
+
+ } catch (IOException e) {
+ log.warn("{}: Unable to send chunk: {}/{}. Reseting snapshot progress. Snapshot state: {}", logName(),
+ installSnapshotState.getChunkIndex(), installSnapshotState.getTotalChunks(),
+ installSnapshotState);
+ installSnapshotState.reset();
+ }
+ }
+ }
+
+ private void sendSnapshotChunk(final ActorSelection followerActor, final FollowerLogInformation followerLogInfo,
+ final byte[] snapshotChunk, final int chunkIndex,
+ final Optional<ServerConfigurationPayload> serverConfig) {
+ LeaderInstallSnapshotState installSnapshotState = followerLogInfo.getInstallSnapshotState();
+
+ installSnapshotState.startChunkTimer();
+ followerActor.tell(
+ new InstallSnapshot(currentTerm(), context.getId(),
snapshotHolder.get().getLastIncludedIndex(),
snapshotHolder.get().getLastIncludedTerm(),
- nextSnapshotChunk,
- nextChunkIndex,
+ snapshotChunk,
+ chunkIndex,
installSnapshotState.getTotalChunks(),
- Optional.of(installSnapshotState.getLastChunkHashCode()),
+ OptionalInt.of(installSnapshotState.getLastChunkHashCode()),
serverConfig
- ).toSerializable(followerLogInfo.getRaftVersion()),
- actor()
- );
-
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
+ ).toSerializable(followerLogInfo.getRaftVersion()),
+ actor()
+ );
+ }
- log.debug("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}", logName(), followerActor.path(),
- installSnapshotState.getChunkIndex(), installSnapshotState.getTotalChunks());
+ private boolean resendSnapshotChunk(final ActorSelection followerActor,
+ final FollowerLogInformation followerLogInfo) {
+ if (!snapshotHolder.isPresent()) {
+ // Seems like we should never hit this case, but just in case we do, reset the snapshot progress so that it
+ // can restart from the next AppendEntries.
+ log.warn("{}: Attempting to resend snapshot with no snapshot holder present.", logName());
+ followerLogInfo.clearLeaderInstallSnapshotState();
+ return false;
}
+
+ LeaderInstallSnapshotState installSnapshotState = followerLogInfo.getInstallSnapshotState();
+ // we are resending, timer needs to be reset
+ installSnapshotState.resetChunkTimer();
+ installSnapshotState.markSendStatus(false);
+
+ sendSnapshotChunk(followerActor, followerLogInfo);
+
+ return true;
}
private void sendHeartBeat() {
Code Review / controller.git / blobdiff