* 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 akka.actor.ActorRef;
import akka.cluster.ClusterEvent.CurrentClusterState;
import akka.cluster.Member;
import akka.cluster.MemberStatus;
-import akka.japi.Procedure;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Stopwatch;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import javax.annotation.Nullable;
+import java.util.function.Consumer;
+import org.eclipse.jdt.annotation.Nullable;
import org.opendaylight.controller.cluster.messaging.MessageAssembler;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
import org.opendaylight.controller.cluster.raft.RaftState;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
+import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
}
@VisibleForTesting
- protected final void setLeaderId(@Nullable final String leaderId) {
+ protected final void setLeaderId(final @Nullable String leaderId) {
this.leaderId = leaderId;
}
// applied to the state already, as the persistence callback occurs async, and we want those entries
// purged from the persisted log as well.
final AtomicBoolean shouldCaptureSnapshot = new AtomicBoolean(false);
- final Procedure<ReplicatedLogEntry> appendAndPersistCallback = logEntry -> {
+ final Consumer<ReplicatedLogEntry> appendAndPersistCallback = logEntry -> {
final List<ReplicatedLogEntry> entries = appendEntries.getEntries();
final ReplicatedLogEntry lastEntryToAppend = entries.get(entries.size() - 1);
if (shouldCaptureSnapshot.get() && logEntry == lastEntryToAppend) {
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(),
return this;
}
+ @Override
+ final ApplyState getApplyStateFor(final ReplicatedLogEntry entry) {
+ return new ApplyState(null, null, entry);
+ }
+
@Override
public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) {
if (message instanceof ElectionTimeout || message instanceof TimeoutNow) {
// If RPC request or response contains term T > currentTerm:
// set currentTerm = T, convert to follower (ยง5.1)
// This applies to all RPC messages and responses
- if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
+ if (rpc.getTerm() > context.getTermInformation().getCurrentTerm() && shouldUpdateTerm(rpc)) {
log.info("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term",
logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
installSnapshot.getLastIncludedTerm(),
context.getTermInformation().getCurrentTerm(),
context.getTermInformation().getVotedFor(),
- installSnapshot.getServerConfig().orNull());
+ installSnapshot.getServerConfig().orElse(null));
ApplySnapshot.Callback applySnapshotCallback = new ApplySnapshot.Callback() {
@Override