// it can happen that the state has not changed but the leader has changed.
Optional<ActorRef> roleChangeNotifier = getRoleChangeNotifier();
- if(!Objects.equal(oldBehaviorLeaderId, currentBehavior.getLeaderId())) {
+ if(!Objects.equal(oldBehaviorLeaderId, currentBehavior.getLeaderId()) ||
+ oldBehaviorState.getLeaderPayloadVersion() != currentBehavior.getLeaderPayloadVersion()) {
if(roleChangeNotifier.isPresent()) {
- roleChangeNotifier.get().tell(newLeaderStateChanged(getId(), currentBehavior.getLeaderId()), getSelf());
+ roleChangeNotifier.get().tell(newLeaderStateChanged(getId(), currentBehavior.getLeaderId(),
+ currentBehavior.getLeaderPayloadVersion()), getSelf());
}
onLeaderChanged(oldBehaviorLeaderId, currentBehavior.getLeaderId());
}
}
- protected LeaderStateChanged newLeaderStateChanged(String memberId, String leaderId) {
- return new LeaderStateChanged(memberId, leaderId);
+ protected LeaderStateChanged newLeaderStateChanged(String memberId, String leaderId, short leaderPayloadVersion) {
+ return new LeaderStateChanged(memberId, leaderId, leaderPayloadVersion);
}
@Override
private static class BehaviorStateHolder {
private RaftActorBehavior behavior;
private String leaderId;
+ private short leaderPayloadVersion;
void init(RaftActorBehavior behavior) {
this.behavior = behavior;
this.leaderId = behavior != null ? behavior.getLeaderId() : null;
+ this.leaderPayloadVersion = behavior != null ? behavior.getLeaderPayloadVersion() : -1;
}
RaftActorBehavior getBehavior() {
String getLeaderId() {
return leaderId;
}
+
+ short getLeaderPayloadVersion() {
+ return leaderPayloadVersion;
+ }
}
}
public AbstractLeader(RaftActorContext context) {
super(context, RaftState.Leader);
+ setLeaderPayloadVersion(context.getPayloadVersion());
+
final Builder<String, FollowerLogInformation> ftlBuilder = ImmutableMap.builder();
for (String followerId : context.getPeerAddresses().keySet()) {
FollowerLogInformation followerLogInformation =
*/
protected String leaderId = null;
+ private short leaderPayloadVersion = -1;
+
private long replicatedToAllIndex = -1;
private final String logName;
return leaderId;
}
+ @Override
+ public short getLeaderPayloadVersion() {
+ return leaderPayloadVersion;
+ }
+
+ public void setLeaderPayloadVersion(short leaderPayloadVersion) {
+ this.leaderPayloadVersion = leaderPayloadVersion;
+ }
+
protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) {
LOG.info("{} :- Switching from behavior {} to {}", logName(), this.state(), behavior.state());
try {
public long getReplicatedToAllIndex() {
return delegate.getReplicatedToAllIndex();
}
+
+ @Override
+ public short getLeaderPayloadVersion() {
+ return delegate.getLeaderPayloadVersion();
+ }
}
// If we got here then we do appear to be talking to the leader
leaderId = appendEntries.getLeaderId();
+ setLeaderPayloadVersion(appendEntries.getPayloadVersion());
+
// 2. Reply false if log doesn’t contain an entry at prevLogIndex
// whose term matches prevLogTerm (§5.3)
* @return
*/
long getReplicatedToAllIndex();
+
+ /**
+ * Returns the leader's payload data version.
+ */
+ short getLeaderPayloadVersion();
}
public class MockRaftActor extends RaftActor implements RaftActorRecoveryCohort, RaftActorSnapshotCohort {
+ public static final short PAYLOAD_VERSION = 5;
+
final RaftActor actorDelegate;
final RaftActorRecoveryCohort recoveryCohortDelegate;
final RaftActorSnapshotCohort snapshotCohortDelegate;
public MockRaftActor(String id, Map<String, String> peerAddresses, Optional<ConfigParams> config,
DataPersistenceProvider dataPersistenceProvider) {
- super(id, peerAddresses, config, (short) 0);
+ super(id, peerAddresses, config, PAYLOAD_VERSION);
state = new ArrayList<>();
this.actorDelegate = mock(RaftActor.class);
this.recoveryCohortDelegate = mock(RaftActorRecoveryCohort.class);
notifierActor, LeaderStateChanged.class);
assertEquals(raftRoleChanged.getMemberId(), leaderStateChange.getLeaderId());
+ assertEquals(MockRaftActor.PAYLOAD_VERSION, leaderStateChange.getLeaderPayloadVersion());
notifierActor.underlyingActor().clear();
MockRaftActor raftActor = raftActorRef.underlyingActor();
final String newLeaderId = "new-leader";
+ final short newLeaderVersion = 6;
Follower follower = new Follower(raftActor.getRaftActorContext()) {
@Override
public RaftActorBehavior handleMessage(ActorRef sender, Object message) {
leaderId = newLeaderId;
+ setLeaderPayloadVersion(newLeaderVersion);
return this;
}
};
leaderStateChange = MessageCollectorActor.expectFirstMatching(notifierActor, LeaderStateChanged.class);
assertEquals(persistenceId, leaderStateChange.getMemberId());
assertEquals(newLeaderId, leaderStateChange.getLeaderId());
+ assertEquals(newLeaderVersion, leaderStateChange.getLeaderPayloadVersion());
+
+ notifierActor.underlyingActor().clear();
+
+ raftActor.handleCommand("any");
+
+ Uninterruptibles.sleepUninterruptibly(505, TimeUnit.MILLISECONDS);
+ leaderStateChange = MessageCollectorActor.getFirstMatching(notifierActor, LeaderStateChanged.class);
+ assertNull(leaderStateChange);
}};
}
// before the new behavior was created (1 in this case)
// This will not work for a Candidate because as soon as a Candidate
// is created it increments the term
- AppendEntries appendEntries = new AppendEntries(1, "leader-1", 2, 1, entries, 4, -1, (short)0);
+ short leaderPayloadVersion = 10;
+ String leaderId = "leader-1";
+ AppendEntries appendEntries = new AppendEntries(1, leaderId, 2, 1, entries, 4, -1, leaderPayloadVersion);
follower = createBehavior(context);
assertEquals("Entry 3", entries.get(0), log.get(3));
assertEquals("Entry 4", entries.get(1), log.get(4));
+ assertEquals("getLeaderPayloadVersion", leaderPayloadVersion, newBehavior.getLeaderPayloadVersion());
+ assertEquals("getLeaderId", leaderId, newBehavior.getLeaderId());
+
expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
}
Props.create(ForwardMessageToBehaviorActor.class), actorFactory.generateActorId("follower"));
private Leader leader;
+ private final short payloadVersion = 5;
@Override
@After
configParams.setElectionTimeoutFactor(100000);
MockRaftActorContext context = new MockRaftActorContext(id, getSystem(), actorRef);
context.setConfigParams(configParams);
+ context.setPayloadVersion(payloadVersion);
return context;
}
leader = new Leader(leaderActorContext);
+ assertEquals(payloadVersion, leader.getLeaderPayloadVersion());
+
short payloadVersion = 5;
AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, true, 2, 1, payloadVersion);
public class LeaderStateChanged {
private final String memberId;
private final String leaderId;
+ private final short leaderPayloadVersion;
- public LeaderStateChanged(@Nonnull String memberId, @Nullable String leaderId) {
+ public LeaderStateChanged(@Nonnull String memberId, @Nullable String leaderId, short leaderPayloadVersion) {
this.memberId = Preconditions.checkNotNull(memberId);
this.leaderId = leaderId;
+ this.leaderPayloadVersion = leaderPayloadVersion;
}
public @Nonnull String getMemberId() {
return leaderId;
}
+ public short getLeaderPayloadVersion() {
+ return leaderPayloadVersion;
+ }
+
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("LeaderStateChanged [memberId=").append(memberId).append(", leaderId=").append(leaderId)
- .append("]");
+ .append(", leaderPayloadVersion=").append(leaderPayloadVersion).append("]");
return builder.toString();
}
}
}
@Override
- protected LeaderStateChanged newLeaderStateChanged(String memberId, String leaderId) {
+ protected LeaderStateChanged newLeaderStateChanged(String memberId, String leaderId, short leaderPayloadVersion) {
return new ShardLeaderStateChanged(memberId, leaderId,
- isLeader() ? Optional.<DataTree>of(store.getDataTree()) : Optional.<DataTree>absent());
+ isLeader() ? Optional.<DataTree>of(store.getDataTree()) : Optional.<DataTree>absent(),
+ leaderPayloadVersion);
}
private void onDatastoreContext(DatastoreContext context) {
private final Optional<DataTree> localShardDataTree;
public ShardLeaderStateChanged(@Nonnull String memberId, @Nonnull String leaderId,
- @Nonnull Optional<DataTree> localShardDataTree) {
- super(memberId, leaderId);
+ @Nonnull Optional<DataTree> localShardDataTree, short leaderPayloadVersion) {
+ super(memberId, leaderId, leaderPayloadVersion);
this.localShardDataTree = Preconditions.checkNotNull(localShardDataTree);
}
TestActorRef<RoleChangeNotifier> notifierTestActorRef = TestActorRef.create(
getSystem(), RoleChangeNotifier.getProps(actorId), actorId);
- notifierTestActorRef.tell(new LeaderStateChanged("member1", "leader1"), ActorRef.noSender());
+ notifierTestActorRef.tell(new LeaderStateChanged("member1", "leader1", (short)5), ActorRef.noSender());
// listener registers after the sate has been changed, ensure we sent the latest state change after a reply
notifierTestActorRef.tell(new RegisterRoleChangeListener(), getRef());
LeaderStateChanged leaderStateChanged = expectMsgClass(LeaderStateChanged.class);
assertEquals("getMemberId", "member1", leaderStateChanged.getMemberId());
assertEquals("getLeaderId", "leader1", leaderStateChanged.getLeaderId());
+ assertEquals("getLeaderPayloadVersion", 5, leaderStateChanged.getLeaderPayloadVersion());
- notifierTestActorRef.tell(new LeaderStateChanged("member1", "leader2"), ActorRef.noSender());
+ notifierTestActorRef.tell(new LeaderStateChanged("member1", "leader2", (short)6), ActorRef.noSender());
leaderStateChanged = expectMsgClass(LeaderStateChanged.class);
assertEquals("getMemberId", "member1", leaderStateChanged.getMemberId());
assertEquals("getLeaderId", "leader2", leaderStateChanged.getLeaderId());
+ assertEquals("getLeaderPayloadVersion", 6, leaderStateChanged.getLeaderPayloadVersion());
}};
}
}
shardManager.tell(new ActorInitialized(), mockShardActor);
DataTree mockDataTree = mock(DataTree.class);
- shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree)), getRef());
+ shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree),
+ DataStoreVersions.CURRENT_VERSION), getRef());
MessageCollectorActor.expectFirstMatching(mockShardActor, RegisterRoleChangeListener.class);
shardManager.tell((new RoleChangeNotification(memberId, RaftState.Candidate.name(),
String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
shardManager.tell(new RoleChangeNotification(memberId1,
RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor);
- shardManager.tell(new LeaderStateChanged(memberId1, memberId2), mockShardActor);
+ shardManager.tell(new LeaderStateChanged(memberId1, memberId2, DataStoreVersions.CURRENT_VERSION), mockShardActor);
shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
shardManager.tell(new RoleChangeNotification(memberId1,
RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor);
- shardManager.tell(new ShardLeaderStateChanged(memberId1, memberId2, Optional.<DataTree>absent()), mockShardActor);
+ shardManager.tell(new ShardLeaderStateChanged(memberId1, memberId2, Optional.<DataTree>absent(),
+ DataStoreVersions.CURRENT_VERSION), mockShardActor);
shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
expectMsgClass(duration("5 seconds"), NoShardLeaderException.class);
DataTree mockDataTree = mock(DataTree.class);
- shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree)), mockShardActor);
+ shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree),
+ DataStoreVersions.CURRENT_VERSION), mockShardActor);
shardManager.tell(new FindPrimary(Shard.DEFAULT_NAME, false), getRef());
expectNoMsg(FiniteDuration.create(150, TimeUnit.MILLISECONDS));
DataTree mockDataTree = mock(DataTree.class);
- shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree)), mockShardActor);
+ shardManager.tell(new ShardLeaderStateChanged(memberId, memberId, Optional.of(mockDataTree),
+ DataStoreVersions.CURRENT_VERSION), mockShardActor);
LocalPrimaryShardFound primaryFound = expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
assertTrue("Unexpected primary path " + primaryFound.getPrimaryPath(),
String memberId2 = "member-2-shard-astronauts-" + shardMrgIDSuffix;
shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2,
- Optional.of(mock(DataTree.class))), mockShardActor2);
+ Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION), mockShardActor2);
shardManager2.tell(new RoleChangeNotification(memberId2,
RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor2);
String memberId2 = "member-2-shard-default-" + shardMrgIDSuffix;
String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId2,
- Optional.of(mock(DataTree.class))), mockShardActor1);
+ Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION), mockShardActor1);
shardManager1.tell(new RoleChangeNotification(memberId1,
RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor1);
- shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2, Optional.of(mock(DataTree.class))),
+ shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION),
mockShardActor2);
shardManager2.tell(new RoleChangeNotification(memberId2,
RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor2);
String memberId2 = "member-2-shard-default-" + shardMrgIDSuffix;
String memberId1 = "member-1-shard-default-" + shardMrgIDSuffix;
shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId2,
- Optional.of(mock(DataTree.class))), mockShardActor1);
+ Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION), mockShardActor1);
shardManager1.tell(new RoleChangeNotification(memberId1,
RaftState.Candidate.name(), RaftState.Follower.name()), mockShardActor1);
- shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2, Optional.of(mock(DataTree.class))),
+ shardManager2.tell(new ShardLeaderStateChanged(memberId2, memberId2, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION),
mockShardActor2);
shardManager2.tell(new RoleChangeNotification(memberId2,
RaftState.Candidate.name(), RaftState.Leader.name()), mockShardActor2);
assertNull("Expected primaryShardInfoCache entry removed", primaryShardInfoCache.getIfPresent("default"));
- shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId1, Optional.of(mock(DataTree.class))),
- mockShardActor1);
+ shardManager1.tell(new ShardLeaderStateChanged(memberId1, memberId1, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION), mockShardActor1);
shardManager1.tell(new RoleChangeNotification(memberId1,
RaftState.Follower.name(), RaftState.Leader.name()), mockShardActor1);
verify(ready, never()).countDown();
shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId, memberId,
- Optional.of(mock(DataTree.class))));
+ Optional.of(mock(DataTree.class)), DataStoreVersions.CURRENT_VERSION));
verify(ready, times(1)).countDown();
shardManager.underlyingActor().onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", getRef().path().toString()));
shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId,
- "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class))));
+ "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION));
verify(ready, times(1)).countDown();
verify(ready, never()).countDown();
shardManager.underlyingActor().onReceiveCommand(new ShardLeaderStateChanged(memberId,
- "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class))));
+ "member-2-shard-default-" + shardMrgIDSuffix, Optional.of(mock(DataTree.class)),
+ DataStoreVersions.CURRENT_VERSION));
shardManager.underlyingActor().onReceiveCommand(MockClusterWrapper.createMemberUp("member-2", getRef().path().toString()));
Code Review / controller.git / commitdiff