import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
import org.opendaylight.controller.cluster.raft.ClientRequestTrackerImpl;
import org.opendaylight.controller.cluster.raft.FollowerLogInformation;
-import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
import org.opendaylight.controller.cluster.raft.PeerInfo;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
import org.opendaylight.controller.cluster.raft.RaftState;
import scala.concurrent.duration.FiniteDuration;
/**
- * The behavior of a RaftActor when it is in the Leader state
+ * The behavior of a RaftActor when it is in the Leader state.
*
* <p>
* Leaders:
trackers.addAll(initializeFromLeader.trackers);
} else {
for (PeerInfo peerInfo: context.getPeers()) {
- FollowerLogInformation followerLogInformation = new FollowerLogInformationImpl(peerInfo, -1, context);
+ FollowerLogInformation followerLogInformation = new FollowerLogInformation(peerInfo, context);
followerToLog.put(peerInfo.getId(), followerLogInformation);
}
}
}
public void addFollower(final String followerId) {
- FollowerLogInformation followerLogInformation = new FollowerLogInformationImpl(
- context.getPeerInfo(followerId), -1, context);
+ FollowerLogInformation followerLogInformation = new FollowerLogInformation(context.getPeerInfo(followerId),
+ context);
followerToLog.put(followerId, followerLogInformation);
if (heartbeatSchedule == null) {
}
@VisibleForTesting
- void setSnapshot(@Nullable final SnapshotHolder snapshotHolder) {
+ void setSnapshotHolder(@Nullable final SnapshotHolder snapshotHolder) {
this.snapshotHolder = Optional.fromNullable(snapshotHolder);
}
return this;
}
- if (followerLogInformation.timeSinceLastActivity()
- > context.getConfigParams().getElectionTimeOutInterval().toMillis()) {
+ final long lastActivityNanos = followerLogInformation.nanosSinceLastActivity();
+ if (lastActivityNanos > context.getConfigParams().getElectionTimeOutInterval().toNanos()) {
log.warn("{} : handleAppendEntriesReply delayed beyond election timeout, "
+ "appendEntriesReply : {}, timeSinceLastActivity : {}, lastApplied : {}, commitIndex : {}",
- logName(), appendEntriesReply, followerLogInformation.timeSinceLastActivity(),
+ logName(), appendEntriesReply, TimeUnit.NANOSECONDS.toMillis(lastActivityNanos),
context.getLastApplied(), context.getCommitIndex());
}
scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval());
} else if (message instanceof SendInstallSnapshot) {
SendInstallSnapshot sendInstallSnapshot = (SendInstallSnapshot) message;
- setSnapshot(new SnapshotHolder(sendInstallSnapshot.getSnapshot(), sendInstallSnapshot.getSnapshotBytes()));
+ setSnapshotHolder(new SnapshotHolder(sendInstallSnapshot.getSnapshot(),
+ sendInstallSnapshot.getSnapshotBytes()));
sendInstallSnapshot();
} else if (message instanceof Replicate) {
replicate((Replicate) message);
if (!anyFollowersInstallingSnapshot()) {
// once there are no pending followers receiving snapshots
// we can remove snapshot from the memory
- setSnapshot(null);
+ setSnapshotHolder(null);
}
wasLastChunk = true;
}
}
- protected void sendAppendEntries(final long timeSinceLastActivityInterval, final boolean isHeartbeat) {
+ protected void sendAppendEntries(final long timeSinceLastActivityIntervalNanos, final boolean isHeartbeat) {
// Send an AppendEntries to all followers
for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
final String followerId = e.getKey();
final FollowerLogInformation followerLogInformation = e.getValue();
// This checks helps not to send a repeat message to the follower
if (!followerLogInformation.isFollowerActive()
- || followerLogInformation.timeSinceLastActivity() >= timeSinceLastActivityInterval) {
+ || followerLogInformation.nanosSinceLastActivity() >= timeSinceLastActivityIntervalNanos) {
sendUpdatesToFollower(followerId, followerLogInformation, true, isHeartbeat);
}
}
}
}
- private List<ReplicatedLogEntry> getEntriesToSend(FollowerLogInformation followerLogInfo,
- ActorSelection followerActor) {
+ private List<ReplicatedLogEntry> getEntriesToSend(final FollowerLogInformation followerLogInfo,
+ final ActorSelection followerActor) {
// Try to get all the entries in the journal but not exceeding the max data size for a single AppendEntries
// message.
int maxEntries = (int) context.getReplicatedLog().size();
return Collections.emptyList();
}
- private void sendAppendEntriesToFollower(ActorSelection followerActor, List<ReplicatedLogEntry> entries,
- FollowerLogInformation followerLogInformation) {
+ private void sendAppendEntriesToFollower(final ActorSelection followerActor, final List<ReplicatedLogEntry> entries,
+ final FollowerLogInformation followerLogInformation) {
// In certain cases outlined below we don't want to send the actual commit index to prevent the follower from
// possibly committing and applying conflicting entries (those with same index, different term) from a prior
// term that weren't replicated to a majority, which would be a violation of raft.
private void sendHeartBeat() {
if (!followerToLog.isEmpty()) {
log.trace("{}: Sending heartbeat", logName());
- sendAppendEntries(context.getConfigParams().getHeartBeatInterval().toMillis(), true);
+ sendAppendEntries(context.getConfigParams().getHeartBeatInterval().toNanos(), true);
appendEntriesMessageSlicer.checkExpiredSlicedMessageState();
}