X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FAbstractLeaderElectionScenarioTest.java;h=8c7c9cb7c4a3ed640d5d04c0a58534a99a819abb;hp=50f0c7b6af0b7d133e862430a9287777910fc8b1;hb=51bf6c7493843ad23f074930b56925162732a8b2;hpb=3c82a8f501a71ec8a40b170fc7ef12f8683c1842 diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeaderElectionScenarioTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeaderElectionScenarioTest.java index 50f0c7b6af..8c7c9cb7c4 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeaderElectionScenarioTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeaderElectionScenarioTest.java @@ -10,12 +10,19 @@ package org.opendaylight.controller.cluster.raft.behaviors; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; + import akka.actor.ActorRef; import akka.actor.ActorSystem; import akka.actor.Props; +import akka.actor.Status; +import akka.dispatch.ControlMessage; import akka.dispatch.Dispatchers; +import akka.dispatch.Mailboxes; +import akka.pattern.Patterns; import akka.testkit.JavaTestKit; import akka.testkit.TestActorRef; +import akka.util.Timeout; +import com.google.common.base.Throwables; import com.google.common.util.concurrent.Uninterruptibles; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -25,13 +32,15 @@ import org.junit.After; import org.junit.Before; import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl; import org.opendaylight.controller.cluster.raft.MockRaftActorContext; -import org.opendaylight.controller.cluster.raft.RaftActorContext; import org.opendaylight.controller.cluster.raft.RaftState; +import org.opendaylight.controller.cluster.raft.TestActorFactory; import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import scala.concurrent.Await; +import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; /** @@ -43,41 +52,72 @@ public class AbstractLeaderElectionScenarioTest { static final int HEARTBEAT_INTERVAL = 50; static class MemberActor extends MessageCollectorActor { - - volatile RaftActorBehavior behavior; + private volatile RaftActorBehavior behavior; Map, CountDownLatch> messagesReceivedLatches = new ConcurrentHashMap<>(); Map, Boolean> dropMessagesToBehavior = new ConcurrentHashMap<>(); CountDownLatch behaviorStateChangeLatch; public static Props props() { - return Props.create(MemberActor.class).withDispatcher(Dispatchers.DefaultDispatcherId()); + return Props.create(MemberActor.class).withDispatcher(Dispatchers.DefaultDispatcherId()) + .withMailbox(Mailboxes.DefaultMailboxId()); } @Override public void onReceive(Object message) throws Exception { // Ignore scheduled SendHeartBeat messages. - if(message instanceof SendHeartBeat) { + if (message instanceof SendHeartBeat) { + return; + } + + if (message instanceof SetBehavior) { + behavior = ((SetBehavior)message).behavior; + ((SetBehavior)message).context.setCurrentBehavior(behavior); return; } + if (message instanceof GetBehaviorState) { + if (behavior != null) { + getSender().tell(behavior.state(), self()); + } else { + getSender().tell(new Status.Failure(new IllegalStateException( + "RaftActorBehavior is not set in MemberActor")), self()); + } + } + + if (message instanceof SendImmediateHeartBeat) { + message = SendHeartBeat.INSTANCE; + } + try { - if(behavior != null && !dropMessagesToBehavior.containsKey(message.getClass())) { - RaftActorBehavior oldBehavior = behavior; - behavior = behavior.handleMessage(getSender(), message); - if(behavior != oldBehavior && behaviorStateChangeLatch != null) { - behaviorStateChangeLatch.countDown(); + if (behavior != null && !dropMessagesToBehavior.containsKey(message.getClass())) { + final RaftActorBehavior nextBehavior = behavior.handleMessage(getSender(), message); + if (nextBehavior != null) { + RaftActorBehavior oldBehavior = behavior; + behavior = nextBehavior; + if (behavior != oldBehavior && behaviorStateChangeLatch != null) { + behaviorStateChangeLatch.countDown(); + } } } } finally { super.onReceive(message); CountDownLatch latch = messagesReceivedLatches.get(message.getClass()); - if(latch != null) { + if (latch != null) { latch.countDown(); } } } + @Override + public void postStop() throws Exception { + super.postStop(); + + if (behavior != null) { + behavior.close(); + } + } + void expectBehaviorStateChange() { behaviorStateChangeLatch = new CountDownLatch(1); } @@ -126,20 +166,45 @@ public class AbstractLeaderElectionScenarioTest { } void forwardCapturedMessagesToBehavior(Class msgClass, ActorRef sender) throws Exception { - for(Object m: getAllMatching(getSelf(), msgClass)) { + for (Object m: getAllMatching(getSelf(), msgClass)) { getSelf().tell(m, sender); } } T getCapturedMessage(Class msgClass) throws Exception { - Object message = getFirstMatching(getSelf(), msgClass); + T message = getFirstMatching(getSelf(), msgClass); assertNotNull("Message of type " + msgClass + " not received", message); - return (T) message; + return message; + } + } + + static class SendImmediateHeartBeat implements ControlMessage { + public static final SendImmediateHeartBeat INSTANCE = new SendImmediateHeartBeat(); + + private SendImmediateHeartBeat() { + } + } + + static class GetBehaviorState implements ControlMessage { + public static final GetBehaviorState INSTANCE = new GetBehaviorState(); + + private GetBehaviorState() { + } + } + + static class SetBehavior implements ControlMessage { + RaftActorBehavior behavior; + MockRaftActorContext context; + + SetBehavior(RaftActorBehavior behavior, MockRaftActorContext context) { + this.behavior = behavior; + this.context = context; } } protected final Logger testLog = LoggerFactory.getLogger(MockRaftActorContext.class); protected final ActorSystem system = ActorSystem.create("test"); + protected final TestActorFactory factory = new TestActorFactory(system); protected TestActorRef member1ActorRef; protected TestActorRef member2ActorRef; protected TestActorRef member3ActorRef; @@ -162,7 +227,7 @@ public class AbstractLeaderElectionScenarioTest { } @After - public void tearDown() { + public void tearDown() throws Exception { JavaTestKit.shutdownActorSystem(system); } @@ -182,31 +247,62 @@ public class AbstractLeaderElectionScenarioTest { return context; } + @SuppressWarnings("checkstyle:IllegalCatch") void verifyBehaviorState(String name, MemberActor actor, RaftState expState) { - assertEquals(name + " behavior state", expState, actor.behavior.state()); + try { + RaftState actualState = (RaftState) Await.result(Patterns.ask(actor.self(), GetBehaviorState.INSTANCE, + Timeout.apply(5, TimeUnit.SECONDS)), Duration.apply(5, TimeUnit.SECONDS)); + assertEquals(name + " behavior state", expState, actualState); + } catch (Exception e) { + Throwables.propagate(e); + } } - void initializeLeaderBehavior(MemberActor actor, RaftActorContext context, - int numActiveFollowers) throws Exception { + void initializeLeaderBehavior(MemberActor actor, MockRaftActorContext context, int numActiveFollowers) + throws Exception { // Leader sends immediate heartbeats - we don't care about it so ignore it. + // Sometimes the initial AppendEntries messages go to dead letters, probably b/c the follower actors + // haven't been fully created/initialized by akka. So we try up to 3 times to create the Leader as + // a workaround. - actor.expectMessageClass(AppendEntriesReply.class, numActiveFollowers); - Leader leader = new Leader(context); - actor.waitForExpectedMessages(AppendEntriesReply.class); - actor.behavior = leader; + Leader leader = null; + AssertionError lastAssertError = null; + for (int i = 1; i <= 3; i++) { + actor.expectMessageClass(AppendEntriesReply.class, numActiveFollowers); + + leader = new Leader(context); + try { + actor.waitForExpectedMessages(AppendEntriesReply.class); + lastAssertError = null; + break; + } catch (AssertionError e) { + lastAssertError = e; + } + } + + if (lastAssertError != null) { + throw lastAssertError; + } + + context.setCurrentBehavior(leader); + + // Delay assignment of the leader behavior so the AppendEntriesReply isn't forwarded to the behavior. + actor.self().tell(new SetBehavior(leader, context), ActorRef.noSender()); actor.forwardCapturedMessagesToBehavior(AppendEntriesReply.class, ActorRef.noSender()); actor.clear(); + } TestActorRef newMemberActor(String name) throws Exception { - TestActorRef actor = TestActorRef.create(system, MemberActor.props(), name); + TestActorRef actor = factory.createTestActor(MemberActor.props() + .withDispatcher(Dispatchers.DefaultDispatcherId()), name); MessageCollectorActor.waitUntilReady(actor); return actor; } void sendHeartbeat(TestActorRef leaderActor) { Uninterruptibles.sleepUninterruptibly(HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS); - leaderActor.underlyingActor().behavior.handleMessage(leaderActor, new SendHeartBeat()); + leaderActor.tell(SendImmediateHeartBeat.INSTANCE, ActorRef.noSender()); } }