X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardTest.java;h=4f4162f607b76c60a22914b74c651626747b61a1;hb=5273c33b6f2051a7e3b1afcc4eeae4e457b6f26c;hp=24a5225b8744edb82981f8eb7dc4ab07eee4a6b4;hpb=2088e1ffef92b11e9d877b2a51280b142a84c58b;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java index 24a5225b87..4f4162f607 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java @@ -1,3 +1,11 @@ +/* + * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved. + * + * This program and the accompanying materials are made available under the + * 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.datastore; import static org.junit.Assert.assertEquals; @@ -12,6 +20,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.verify; import static org.opendaylight.controller.cluster.datastore.DataStoreVersions.CURRENT_VERSION; + import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.PoisonPill; @@ -140,7 +149,7 @@ public class ShardTest extends AbstractShardTest { "testRegisterChangeListener-DataChangeListener"); shard.tell(new RegisterChangeListener(TestModel.TEST_PATH, - dclActor, AsyncDataBroker.DataChangeScope.BASE), getRef()); + dclActor, AsyncDataBroker.DataChangeScope.BASE, true), getRef()); final RegisterChangeListenerReply reply = expectMsgClass(duration("3 seconds"), RegisterChangeListenerReply.class); @@ -233,7 +242,7 @@ public class ShardTest extends AbstractShardTest { // Now send the RegisterChangeListener and wait for the reply. shard.tell(new RegisterChangeListener(path, dclActor, - AsyncDataBroker.DataChangeScope.SUBTREE), getRef()); + AsyncDataBroker.DataChangeScope.SUBTREE, false), getRef()); final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"), RegisterChangeListenerReply.class); @@ -338,11 +347,11 @@ public class ShardTest extends AbstractShardTest { writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); assertEquals("Got first ElectionTimeout", true, - onFirstElectionTimeout.await(5, TimeUnit.SECONDS)); + onFirstElectionTimeout.await(5, TimeUnit.SECONDS)); shard.tell(new RegisterDataTreeChangeListener(path, dclActor), getRef()); final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"), - RegisterDataTreeChangeListenerReply.class); + RegisterDataTreeChangeListenerReply.class); assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath()); shard.tell(new FindLeader(), getRef()); @@ -442,13 +451,13 @@ public class ShardTest extends AbstractShardTest { //waitUntilLeader(shard); assertEquals("Recovery complete", true, - Uninterruptibles.awaitUninterruptibly(recoveryComplete, 5, TimeUnit.SECONDS)); + Uninterruptibles.awaitUninterruptibly(recoveryComplete, 5, TimeUnit.SECONDS)); final String address = "akka://foobar"; shard.underlyingActor().onReceiveCommand(new PeerAddressResolved(shardID.toString(), address)); assertEquals("getPeerAddresses", address, - ((TestShard)shard.underlyingActor()).getPeerAddresses().get(shardID.toString())); + ((TestShard) shard.underlyingActor()).getPeerAddresses().get(shardID.toString())); shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); }}; @@ -543,8 +552,8 @@ public class ShardTest extends AbstractShardTest { final NormalizedNode root = readStore(testStore, YangInstanceIdentifier.builder().build()); InMemorySnapshotStore.addSnapshot(shardID.toString(), Snapshot.create( - SerializationUtils.serializeNormalizedNode(root), - Collections.emptyList(), 0, 1, -1, -1)); + SerializationUtils.serializeNormalizedNode(root), + Collections.emptyList(), 0, 1, -1, -1)); return testStore; } @@ -586,7 +595,7 @@ public class ShardTest extends AbstractShardTest { } InMemoryJournal.addEntry(shardID.toString(), nListEntries + 2, - new ApplyJournalEntries(nListEntries)); + new ApplyJournalEntries(nListEntries)); testRecovery(listEntryKeys); } @@ -602,8 +611,8 @@ public class ShardTest extends AbstractShardTest { InMemoryJournal.addEntry(shardID.toString(), 0, DUMMY_DATA); InMemoryJournal.addEntry(shardID.toString(), 1, new ReplicatedLogImplEntry(0, 1, newModificationPayload( - new WriteModification(TestModel.OUTER_LIST_PATH, - ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build())))); + new WriteModification(TestModel.OUTER_LIST_PATH, + ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build())))); final int nListEntries = 16; final Set listEntryKeys = new HashSet<>(); @@ -928,12 +937,12 @@ public class ShardTest extends AbstractShardTest { // Send a couple more BatchedModifications. shard.tell(newBatchedModifications(transactionID, TestModel.OUTER_LIST_PATH, - ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false, false, 2), getRef()); + ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false, false, 2), getRef()); expectMsgClass(duration, BatchedModificationsReply.class); shard.tell(newBatchedModifications(transactionID, YangInstanceIdentifier.builder( TestModel.OUTER_LIST_PATH).nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(), - ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1), true, true, 3), getRef()); + ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1), true, true, 3), getRef()); expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS); @@ -1260,7 +1269,7 @@ public class ShardTest extends AbstractShardTest { final MutableCompositeModification modification = new MutableCompositeModification(); final NormalizedNode containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); final ShardDataTreeCohort cohort = setupMockWriteTransaction("cohort", dataStore, - TestModel.TEST_PATH, containerNode, modification); + TestModel.TEST_PATH, containerNode, modification); final FiniteDuration duration = duration("5 seconds"); @@ -1268,7 +1277,7 @@ public class ShardTest extends AbstractShardTest { // by the ShardTransaction. shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, - cohort, modification, true, false), getRef()); + cohort, modification, true, false), getRef()); expectMsgClass(duration, ReadyTransactionReply.class); // Send the CanCommitTransaction message. @@ -1551,7 +1560,7 @@ public class ShardTest extends AbstractShardTest { shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef()); final CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable( - expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS)); + expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS)); assertEquals("Can commit", true, canCommitReply.getCanCommit()); // Send the CanCommitTransaction message for the 2nd Tx. This should get queued and @@ -1673,7 +1682,7 @@ public class ShardTest extends AbstractShardTest { doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit(); shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, - cohort, modification, true, false), getRef()); + cohort, modification, true, false), getRef()); expectMsgClass(duration, ReadyTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef()); @@ -1818,12 +1827,12 @@ public class ShardTest extends AbstractShardTest { modification, preCommit); shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, - cohort, modification, true, false), getRef()); + cohort, modification, true, false), getRef()); expectMsgClass(duration, ReadyTransactionReply.class); shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef()); final CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable( - expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS)); + expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS)); assertEquals("Can commit", true, canCommitReply.getCanCommit()); shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef()); @@ -2004,7 +2013,7 @@ public class ShardTest extends AbstractShardTest { TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification1); shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, - cohort1, modification1, true, false), getRef()); + cohort1, modification1, true, false), getRef()); expectMsgClass(duration, ReadyTransactionReply.class); final String transactionID2 = "tx2"; @@ -2022,7 +2031,7 @@ public class ShardTest extends AbstractShardTest { TestModel.TEST2_PATH, ImmutableNodes.containerNode(TestModel.TEST2_QNAME), modification3); shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION, - cohort3, modification3, true, false), getRef()); + cohort3, modification3, true, false), getRef()); expectMsgClass(duration, ReadyTransactionReply.class); // All Tx's are readied. We'll send canCommit for the last one but not the others. The others @@ -2404,7 +2413,7 @@ public class ShardTest extends AbstractShardTest { shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(false).build(); final Props nonPersistentProps = Shard.props(shardID, Collections.emptyMap(), - nonPersistentContext, SCHEMA_CONTEXT); + nonPersistentContext, SCHEMA_CONTEXT); new ShardTestKit(getSystem()) {{ final TestActorRef shard1 = TestActorRef.create(getSystem(), @@ -2440,12 +2449,12 @@ public class ShardTest extends AbstractShardTest { shard.tell(dataStoreContextBuilder.persistent(false).build(), ActorRef.noSender()); assertEquals("isRecoveryApplicable", false, - shard.underlyingActor().persistence().isRecoveryApplicable()); + shard.underlyingActor().persistence().isRecoveryApplicable()); shard.tell(dataStoreContextBuilder.persistent(true).build(), ActorRef.noSender()); assertEquals("isRecoveryApplicable", true, - shard.underlyingActor().persistence().isRecoveryApplicable()); + shard.underlyingActor().persistence().isRecoveryApplicable()); shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); }}; @@ -2469,11 +2478,11 @@ public class ShardTest extends AbstractShardTest { MessageCollectorActor.expectFirstMatching(listener, RegisterRoleChangeListenerReply.class); ShardLeaderStateChanged leaderStateChanged = MessageCollectorActor.expectFirstMatching(listener, - ShardLeaderStateChanged.class); + ShardLeaderStateChanged.class); assertEquals("getLocalShardDataTree present", true, leaderStateChanged.getLocalShardDataTree().isPresent()); assertSame("getLocalShardDataTree", shard.underlyingActor().getDataStore().getDataTree(), - leaderStateChanged.getLocalShardDataTree().get()); + leaderStateChanged.getLocalShardDataTree().get()); MessageCollectorActor.clearMessages(listener); @@ -2513,4 +2522,144 @@ public class ShardTest extends AbstractShardTest { store.validate(modification); store.commit(store.prepare(modification)); } + + @Test + public void testClusteredDataChangeListernerDelayedRegistration() throws Exception { + new ShardTestKit(getSystem()) {{ + final CountDownLatch onFirstElectionTimeout = new CountDownLatch(1); + final CountDownLatch onChangeListenerRegistered = new CountDownLatch(1); + final Creator creator = new Creator() { + boolean firstElectionTimeout = true; + + @Override + public Shard create() throws Exception { + return new Shard(shardID, Collections.emptyMap(), + dataStoreContextBuilder.persistent(false).build(), SCHEMA_CONTEXT) { + @Override + public void onReceiveCommand(final Object message) throws Exception { + if(message instanceof ElectionTimeout && firstElectionTimeout) { + firstElectionTimeout = false; + final ActorRef self = getSelf(); + new Thread() { + @Override + public void run() { + Uninterruptibles.awaitUninterruptibly( + onChangeListenerRegistered, 5, TimeUnit.SECONDS); + self.tell(message, self); + } + }.start(); + + onFirstElectionTimeout.countDown(); + } else { + super.onReceiveCommand(message); + } + } + }; + } + }; + + final MockDataChangeListener listener = new MockDataChangeListener(1); + final ActorRef dclActor = getSystem().actorOf(DataChangeListener.props(listener), + "testDataChangeListenerOnFollower-DataChangeListener"); + + final TestActorRef shard = TestActorRef.create(getSystem(), + Props.create(new DelegatingShardCreator(creator)), + "testDataChangeListenerOnFollower"); + + assertEquals("Got first ElectionTimeout", true, + onFirstElectionTimeout.await(5, TimeUnit.SECONDS)); + + shard.tell(new FindLeader(), getRef()); + final FindLeaderReply findLeadeReply = + expectMsgClass(duration("5 seconds"), FindLeaderReply.class); + assertNull("Expected the shard not to be the leader", findLeadeReply.getLeaderActor()); + + final YangInstanceIdentifier path = TestModel.TEST_PATH; + + shard.tell(new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.BASE, true), getRef()); + final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterChangeListenerReply.class); + assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath()); + + writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + + onChangeListenerRegistered.countDown(); + + listener.waitForChangeEvents(); + + dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender()); + shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); + }}; + } + + @Test + public void testClusteredDataChangeListernerRegistration() throws Exception { + new ShardTestKit(getSystem()) {{ + final ShardIdentifier member1ShardID = ShardIdentifier.builder().memberName("member-1") + .shardName("inventory").type("config").build(); + + final ShardIdentifier member2ShardID = ShardIdentifier.builder().memberName("member-2") + .shardName("inventory").type("config").build(); + final Creator followerShardCreator = new Creator() { + + @Override + public Shard create() throws Exception { + return new Shard(member1ShardID, Collections.singletonMap(member2ShardID.toString(), + "akka://test/user/" + member2ShardID.toString()), + dataStoreContextBuilder.persistent(false).build(), SCHEMA_CONTEXT) { + @Override + public void onReceiveCommand(final Object message) throws Exception { + + if(!(message instanceof ElectionTimeout)) { + super.onReceiveCommand(message); + } + } + }; + } + }; + + final Creator leaderShardCreator = new Creator() { + + @Override + public Shard create() throws Exception { + return new Shard(member2ShardID, Collections.singletonMap(member1ShardID.toString(), + "akka://test/user/" + member1ShardID.toString()), + dataStoreContextBuilder.persistent(false).build(), SCHEMA_CONTEXT) { }; + } + }; + + + final TestActorRef shard = TestActorRef.create(getSystem(), + Props.create(new DelegatingShardCreator(followerShardCreator)), + member1ShardID.toString()); + + final TestActorRef shardLeader = TestActorRef.create(getSystem(), + Props.create(new DelegatingShardCreator(leaderShardCreator)), + member2ShardID.toString()); + // Sleep to let election happen + Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS); + + shard.tell(new FindLeader(), getRef()); + final FindLeaderReply findLeaderReply = + expectMsgClass(duration("5 seconds"), FindLeaderReply.class); + assertEquals("Shard leader does not match", shardLeader.path().toString(), findLeaderReply.getLeaderActor()); + + final YangInstanceIdentifier path = TestModel.TEST_PATH; + final MockDataChangeListener listener = new MockDataChangeListener(1); + final ActorRef dclActor = getSystem().actorOf(DataChangeListener.props(listener), + "testDataChangeListenerOnFollower-DataChangeListener"); + + shard.tell(new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.BASE, true), getRef()); + final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterChangeListenerReply.class); + assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath()); + + writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + + listener.waitForChangeEvents(); + + dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender()); + shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); + }}; + } }