X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FRaftActor.java;h=5dc1b9dcdf8877132b4978f1cde7f47da32e3bc7;hb=655216a6c75aa29d31c4c56c56a5000db56ba233;hp=157a53ed2d1797b662f6fadccd3acf4027ae442f;hpb=254cfb302af490110740ebef111ae28783018a02;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java index 157a53ed2d..5dc1b9dcdf 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java @@ -16,10 +16,10 @@ import akka.persistence.SnapshotSelectionCriteria; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Objects; import com.google.common.base.Optional; -import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; import java.io.Serializable; import java.util.Collection; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -114,12 +114,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { private final BehaviorStateHolder reusableBehaviorStateHolder = new BehaviorStateHolder(); - public RaftActor(String id, Map peerAddresses) { - this(id, peerAddresses, Optional.absent()); - } - public RaftActor(String id, Map peerAddresses, - Optional configParams) { + Optional configParams, short payloadVersion) { context = new RaftActorContextImpl(this.getSelf(), this.getContext(), id, new ElectionTermImpl(delegatingPersistenceProvider, id, LOG), @@ -127,7 +123,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { (configParams.isPresent() ? configParams.get(): new DefaultConfigParamsImpl()), delegatingPersistenceProvider, LOG); - context.setReplicatedLog(ReplicatedLogImpl.newInstance(context, delegatingPersistenceProvider, currentBehavior)); + context.setPayloadVersion(payloadVersion); + context.setReplicatedLog(ReplicatedLogImpl.newInstance(context, currentBehavior)); } @Override @@ -136,6 +133,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { context.getConfigParams().getJournalRecoveryLogBatchSize()); super.preStart(); + + snapshotSupport = newRaftActorSnapshotMessageSupport(); } @Override @@ -177,8 +176,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } protected RaftActorRecoverySupport newRaftActorRecoverySupport() { - return new RaftActorRecoverySupport(delegatingPersistenceProvider, context, currentBehavior, - getRaftActorRecoveryCohort()); + return new RaftActorRecoverySupport(context, currentBehavior, getRaftActorRecoveryCohort()); } protected void initializeBehavior(){ @@ -193,15 +191,6 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { @Override public void handleCommand(Object message) { - if(snapshotSupport == null) { - snapshotSupport = newRaftActorSnapshotMessageSupport(); - } - - boolean handled = snapshotSupport.handleSnapshotMessage(message); - if(handled) { - return; - } - if (message instanceof ApplyState){ ApplyState applyState = (ApplyState) message; @@ -235,7 +224,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { ); } else if(message instanceof GetOnDemandRaftState) { onGetOnDemandRaftStats(); - } else { + } else if(!snapshotSupport.handleSnapshotMessage(message)) { reusableBehaviorStateHolder.init(getCurrentBehavior()); setCurrentBehavior(currentBehavior.handleMessage(getSender(), message)); @@ -245,8 +234,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } protected RaftActorSnapshotMessageSupport newRaftActorSnapshotMessageSupport() { - return new RaftActorSnapshotMessageSupport(delegatingPersistenceProvider, context, - currentBehavior, getRaftActorSnapshotCohort()); + return new RaftActorSnapshotMessageSupport(context, currentBehavior, + getRaftActorSnapshotCohort()); } private void onGetOnDemandRaftStats() { @@ -267,7 +256,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { .snapshotIndex(replicatedLog().getSnapshotIndex()) .snapshotTerm(replicatedLog().getSnapshotTerm()) .votedFor(context.getTermInformation().getVotedFor()) - .peerAddresses(ImmutableMap.copyOf(context.getPeerAddresses())); + .peerAddresses(new HashMap<>(context.getPeerAddresses())); ReplicatedLogEntry lastLogEntry = getLastLogEntry(); if (lastLogEntry != null) { @@ -306,7 +295,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { Optional roleChangeNotifier = getRoleChangeNotifier(); if(!Objects.equal(oldBehaviorLeaderId, currentBehavior.getLeaderId())) { if(roleChangeNotifier.isPresent()) { - roleChangeNotifier.get().tell(new LeaderStateChanged(getId(), currentBehavior.getLeaderId()), getSelf()); + roleChangeNotifier.get().tell(newLeaderStateChanged(getId(), currentBehavior.getLeaderId()), getSelf()); } onLeaderChanged(oldBehaviorLeaderId, currentBehavior.getLeaderId()); @@ -319,6 +308,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } } + protected LeaderStateChanged newLeaderStateChanged(String memberId, String leaderId) { + return new LeaderStateChanged(memberId, leaderId); + } + /** * When a derived RaftActor needs to persist something it must call * persistData. @@ -573,13 +566,14 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } /** - * @deprecated Deprecated in favor of {@link org.opendaylight.controller.cluster.raft.base.messages.DeleteEntriesTest} + * @deprecated Deprecated in favor of {@link org.opendaylight.controller.cluster.raft.base.messages.DeleteEntries} * whose type for fromIndex is long instead of int. This class was kept for backwards * compatibility with Helium. */ + // Suppressing this warning as we can't set serialVersionUID to maintain backwards compatibility. + @SuppressWarnings("serial") @Deprecated static class DeleteEntries implements Serializable { - private static final long serialVersionUID = 1L; private final int fromIndex; public DeleteEntries(int fromIndex) { @@ -591,8 +585,14 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } } + /** + * @deprecated Deprecated in favor of non-inner class {@link org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm} + * which has serialVersionUID set. This class was kept for backwards compatibility with Helium. + */ + // Suppressing this warning as we can't set serialVersionUID to maintain backwards compatibility. + @SuppressWarnings("serial") + @Deprecated static class UpdateElectionTerm implements Serializable { - private static final long serialVersionUID = 1L; private final long currentTerm; private final String votedFor;