package org.opendaylight.controller.cluster.raft;
import static org.junit.Assert.assertEquals;
-import com.google.common.base.Optional;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Sets;
-import java.util.Arrays;
+
+import akka.actor.ActorRef;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.junit.Test;
-import org.opendaylight.controller.cluster.raft.ServerConfigurationPayload.ServerInfo;
+import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
+import org.opendaylight.controller.cluster.raft.AbstractRaftActorIntegrationTest.TestRaftActor.Builder;
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.SnapshotComplete;
-import org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
+import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
+import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
+import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
+import scala.concurrent.duration.FiniteDuration;
/**
* Integration test for various scenarios involving non-voting followers.
public class NonVotingFollowerIntegrationTest extends AbstractRaftActorIntegrationTest {
private TestRaftActor followerInstance;
private TestRaftActor leaderInstance;
+ private final Builder follower1Builder = TestRaftActor.newBuilder();
/**
* Tests non-voting follower re-sync after the non-persistent leader restarts with an empty log. In this
//
// We also add another voting follower actor into the mix even though it shoildn't affect the
// outcome.
- ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList(
+ ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(List.of(
new ServerInfo(leaderId, true), new ServerInfo(follower1Id, false),
new ServerInfo(follower2Id, true), new ServerInfo("downPeer", false)));
- ReplicatedLogImplEntry persistedServerConfigEntry = new ReplicatedLogImplEntry(0, currentTerm,
+ SimpleReplicatedLogEntry persistedServerConfigEntry = new SimpleReplicatedLogEntry(0, currentTerm,
persistedServerConfig);
InMemoryJournal.clear();
DefaultConfigParamsImpl follower2ConfigParams = newFollowerConfigParams();
follower2ConfigParams.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
follower2Actor = newTestRaftActor(follower2Id, TestRaftActor.newBuilder().peerAddresses(
- ImmutableMap.of(leaderId, testActorPath(leaderId), follower1Id, follower1Actor.path().toString())).
- config(follower2ConfigParams).persistent(Optional.of(false)));
+ Map.of(leaderId, testActorPath(leaderId), follower1Id, follower1Actor.path().toString()))
+ .config(follower2ConfigParams).persistent(Optional.of(false)));
TestRaftActor follower2Instance = follower2Actor.underlyingActor();
follower2Instance.waitForRecoveryComplete();
follower2CollectorActor = follower2Instance.collectorActor();
- peerAddresses = ImmutableMap.of(follower1Id, follower1Actor.path().toString(),
+ peerAddresses = Map.of(follower1Id, follower1Actor.path().toString(),
follower2Id, follower2Actor.path().toString());
createNewLeaderActor();
testLog.info("testFollowerResyncWithMoreLeaderLogEntriesAndDownPeerAfterNonPersistentLeaderRestart ending");
}
+ @Test
+ public void testFollowerLeaderStateChanges() {
+ testLog.info("testFollowerLeaderStateChanges");
+
+ ActorRef roleChangeNotifier = factory.createActor(
+ MessageCollectorActor.props(), factory.generateActorId("roleChangeNotifier"));
+ follower1Builder.roleChangeNotifier(roleChangeNotifier);
+
+ setupLeaderAndNonVotingFollower();
+
+ ((DefaultConfigParamsImpl)follower1Context.getConfigParams()).setElectionTimeoutFactor(2);
+ ((DefaultConfigParamsImpl)follower1Context.getConfigParams())
+ .setHeartBeatInterval(FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
+
+ MessageCollectorActor.clearMessages(roleChangeNotifier);
+ follower1Actor.tell(ElectionTimeout.INSTANCE, ActorRef.noSender());
+ followerInstance.startDropMessages(AppendEntries.class);
+
+ LeaderStateChanged leaderStateChanged = MessageCollectorActor.expectFirstMatching(roleChangeNotifier,
+ LeaderStateChanged.class);
+ assertEquals("getLeaderId", null, leaderStateChanged.getLeaderId());
+
+ MessageCollectorActor.clearMessages(roleChangeNotifier);
+ followerInstance.stopDropMessages(AppendEntries.class);
+
+ leaderStateChanged = MessageCollectorActor.expectFirstMatching(roleChangeNotifier,
+ LeaderStateChanged.class);
+ assertEquals("getLeaderId", leaderId, leaderStateChanged.getLeaderId());
+ }
+
private void createNewLeaderActor() {
expSnapshotState.clear();
- leaderActor = newTestRaftActor(leaderId, TestRaftActor.newBuilder().peerAddresses(peerAddresses).
- config(leaderConfigParams).persistent(Optional.of(false)));
+ leaderActor = newTestRaftActor(leaderId, TestRaftActor.newBuilder().peerAddresses(peerAddresses)
+ .config(leaderConfigParams).persistent(Optional.of(false)));
leaderInstance = leaderActor.underlyingActor();
leaderCollectorActor = leaderInstance.collectorActor();
waitUntilLeader(leaderActor);
private void setupLeaderAndNonVotingFollower() {
snapshotBatchCount = 100;
- int initialTerm = 1;
+ int persistedTerm = 1;
// Set up a persisted ServerConfigurationPayload with the leader voting and the follower non-voting.
- ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList(
+ ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(List.of(
new ServerInfo(leaderId, true), new ServerInfo(follower1Id, false)));
- ReplicatedLogImplEntry persistedServerConfigEntry = new ReplicatedLogImplEntry(0, initialTerm,
+ SimpleReplicatedLogEntry persistedServerConfigEntry = new SimpleReplicatedLogEntry(0, persistedTerm,
persistedServerConfig);
- InMemoryJournal.addEntry(leaderId, 1, new UpdateElectionTerm(initialTerm, leaderId));
+ InMemoryJournal.addEntry(leaderId, 1, new UpdateElectionTerm(persistedTerm, leaderId));
InMemoryJournal.addEntry(leaderId, 2, persistedServerConfigEntry);
- InMemoryJournal.addEntry(follower1Id, 1, new UpdateElectionTerm(initialTerm, leaderId));
+ InMemoryJournal.addEntry(follower1Id, 1, new UpdateElectionTerm(persistedTerm, leaderId));
InMemoryJournal.addEntry(follower1Id, 2, persistedServerConfigEntry);
DefaultConfigParamsImpl followerConfigParams = newFollowerConfigParams();
- follower1Actor = newTestRaftActor(follower1Id, TestRaftActor.newBuilder().peerAddresses(
- ImmutableMap.of(leaderId, testActorPath(leaderId))).config(followerConfigParams).
- persistent(Optional.of(false)));
+ follower1Actor = newTestRaftActor(follower1Id, follower1Builder.peerAddresses(
+ Map.of(leaderId, testActorPath(leaderId))).config(followerConfigParams)
+ .persistent(Optional.of(false)));
- peerAddresses = ImmutableMap.<String, String>builder().
- put(follower1Id, follower1Actor.path().toString()).build();
+ peerAddresses = Map.of(follower1Id, follower1Actor.path().toString());
leaderConfigParams = newLeaderConfigParams();
- leaderActor = newTestRaftActor(leaderId, TestRaftActor.newBuilder().peerAddresses(peerAddresses).
- config(leaderConfigParams).persistent(Optional.of(false)));
+ leaderActor = newTestRaftActor(leaderId, TestRaftActor.newBuilder().peerAddresses(peerAddresses)
+ .config(leaderConfigParams).persistent(Optional.of(false)));
followerInstance = follower1Actor.underlyingActor();
follower1CollectorActor = followerInstance.collectorActor();
// Verify leader's context after startup
- currentTerm = initialTerm + 1;
+ currentTerm = persistedTerm + 1;
assertEquals("Leader term", currentTerm, leaderContext.getTermInformation().getCurrentTerm());
- assertEquals("Leader server config", Sets.newHashSet(persistedServerConfig.getServerConfig()),
- Sets.newHashSet(leaderContext.getPeerServerInfo(true).getServerConfig()));
+ assertEquals("Leader server config", Set.copyOf(persistedServerConfig.getServerConfig()),
+ Set.copyOf(leaderContext.getPeerServerInfo(true).getServerConfig()));
assertEquals("Leader isVotingMember", true, leaderContext.isVotingMember());
// Verify follower's context after startup
MessageCollectorActor.expectFirstMatching(follower1CollectorActor, AppendEntries.class);
assertEquals("Follower term", currentTerm, follower1Context.getTermInformation().getCurrentTerm());
- assertEquals("Follower server config", Sets.newHashSet(persistedServerConfig.getServerConfig()),
- Sets.newHashSet(follower1Context.getPeerServerInfo(true).getServerConfig()));
+ assertEquals("Follower server config", Set.copyOf(persistedServerConfig.getServerConfig()),
+ Set.copyOf(follower1Context.getPeerServerInfo(true).getServerConfig()));
assertEquals("FollowerisVotingMember", false, follower1Context.isVotingMember());
}
}
Code Review / controller.git / blobdiff