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=3eb84bde134bf648086e7a025ce41b31c8df9879;hb=6602310b78b5bf54899456e2efa867d116731267;hp=d484f99696a6f89b7528eda2c9e64abc8c212eca;hpb=c1336f9b497bc6867536a24f629c3f0b002ccb2f;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 d484f99696..3eb84bde13 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 @@ -59,6 +59,8 @@ import org.opendaylight.controller.cluster.datastore.messages.BatchedModificatio import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply; +import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeNotificationListenerRegistration; +import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeNotificationListenerRegistrationReply; import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction; @@ -69,9 +71,8 @@ import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply; import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener; -import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply; import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener; -import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListenerReply; +import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeNotificationListenerReply; import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; import org.opendaylight.controller.cluster.datastore.modification.MergeModification; @@ -146,8 +147,8 @@ public class ShardTest extends AbstractShardTest { shard.tell(new RegisterChangeListener(TestModel.TEST_PATH, dclActor, AsyncDataBroker.DataChangeScope.BASE, true), getRef()); - final RegisterChangeListenerReply reply = expectMsgClass(duration("3 seconds"), - RegisterChangeListenerReply.class); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("3 seconds"), + RegisterDataTreeNotificationListenerReply.class); final String replyPath = reply.getListenerRegistrationPath().toString(); assertTrue("Incorrect reply path: " + replyPath, replyPath.matches("akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*")); @@ -197,14 +198,11 @@ public class ShardTest extends AbstractShardTest { // original ElectionTimeout message to proceed with the election. firstElectionTimeout = false; final ActorRef self = getSelf(); - new Thread() { - @Override - public void run() { - Uninterruptibles.awaitUninterruptibly( - onChangeListenerRegistered, 5, TimeUnit.SECONDS); - self.tell(message, self); - } - }.start(); + new Thread(() -> { + Uninterruptibles.awaitUninterruptibly( + onChangeListenerRegistered, 5, TimeUnit.SECONDS); + self.tell(message, self); + }).start(); onFirstElectionTimeout.countDown(); } else { @@ -236,8 +234,8 @@ public class ShardTest extends AbstractShardTest { shard.tell(new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.SUBTREE, false), getRef()); - final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"), - RegisterChangeListenerReply.class); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); // Sanity check - verify the shard is not the leader yet. @@ -276,8 +274,8 @@ public class ShardTest extends AbstractShardTest { shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, false), getRef()); - final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("3 seconds"), - RegisterDataTreeChangeListenerReply.class); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("3 seconds"), + RegisterDataTreeNotificationListenerReply.class); final String replyPath = reply.getListenerRegistrationPath().toString(); assertTrue("Incorrect reply path: " + replyPath, replyPath.matches("akka:\\/\\/test\\/user\\/testRegisterDataTreeChangeListener\\/\\$.*")); @@ -306,14 +304,11 @@ public class ShardTest extends AbstractShardTest { 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(); + new Thread(() -> { + Uninterruptibles.awaitUninterruptibly( + onChangeListenerRegistered, 5, TimeUnit.SECONDS); + self.tell(message, self); + }).start(); onFirstElectionTimeout.countDown(); } else { @@ -340,8 +335,8 @@ public class ShardTest extends AbstractShardTest { assertEquals("Got first ElectionTimeout", true, onFirstElectionTimeout.await(5, TimeUnit.SECONDS)); shard.tell(new RegisterDataTreeChangeListener(path, dclActor, false), getRef()); - final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"), - RegisterDataTreeChangeListenerReply.class); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath()); shard.tell(FindLeader.INSTANCE, getRef()); @@ -373,8 +368,9 @@ public class ShardTest extends AbstractShardTest { CreateTransactionReply.class); final String path = reply.getTransactionPath().toString(); - assertTrue("Unexpected transaction path " + path, path - .startsWith("akka://test/user/testCreateTransaction/shard-member-1:ShardTransactionTest@0:")); + assertTrue("Unexpected transaction path " + path, path.startsWith(String.format( + "akka://test/user/testCreateTransaction/shard-%s-%s:ShardTransactionTest@0:", + shardID.getShardName(), shardID.getMemberName().getName()))); } }; } @@ -394,8 +390,9 @@ public class ShardTest extends AbstractShardTest { CreateTransactionReply.class); final String path = reply.getTransactionPath().toString(); - assertTrue("Unexpected transaction path " + path, path.startsWith( - "akka://test/user/testCreateTransactionOnChain/shard-member-1:ShardTransactionTest@0:")); + assertTrue("Unexpected transaction path " + path, path.startsWith(String.format( + "akka://test/user/testCreateTransactionOnChain/shard-%s-%s:ShardTransactionTest@0:", + shardID.getShardName(), shardID.getMemberName().getName()))); } }; } @@ -813,7 +810,7 @@ public class ShardTest extends AbstractShardTest { final Failure failure = expectMsgClass(duration("5 seconds"), akka.actor.Status.Failure.class); if (failure != null) { - Throwables.propagateIfInstanceOf(failure.cause(), Exception.class); + Throwables.throwIfInstanceOf(failure.cause(), Exception.class); Throwables.propagate(failure.cause()); } } @@ -2016,13 +2013,13 @@ public class ShardTest extends AbstractShardTest { .shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(true).build(); final Props persistentProps = Shard.builder().id(shardID).datastoreContext(persistentContext) - .schemaContext(SCHEMA_CONTEXT).props(); + .schemaContextProvider(() -> SCHEMA_CONTEXT).props(); final DatastoreContext nonPersistentContext = DatastoreContext.newBuilder() .shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(false).build(); final Props nonPersistentProps = Shard.builder().id(shardID).datastoreContext(nonPersistentContext) - .schemaContext(SCHEMA_CONTEXT).props(); + .schemaContextProvider(() -> SCHEMA_CONTEXT).props(); new ShardTestKit(getSystem()) { { @@ -2120,10 +2117,10 @@ public class ShardTest extends AbstractShardTest { } @Test - public void testClusteredDataChangeListenerDelayedRegistration() throws Exception { + public void testClusteredDataChangeListenerWithDelayedRegistration() throws Exception { new ShardTestKit(getSystem()) { { - final String testName = "testClusteredDataChangeListenerDelayedRegistration"; + final String testName = "testClusteredDataChangeListenerWithDelayedRegistration"; dataStoreContextBuilder.shardElectionTimeoutFactor(1000) .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()); @@ -2142,8 +2139,8 @@ public class ShardTest extends AbstractShardTest { shard.tell(new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.BASE, true), getRef()); - final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"), - RegisterChangeListenerReply.class); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); shard.tell(DatastoreContext.newBuilderFrom(dataStoreContextBuilder.build()) @@ -2170,14 +2167,14 @@ public class ShardTest extends AbstractShardTest { .datastoreContext(dataStoreContextBuilder.shardElectionTimeoutFactor(1000).build()) .peerAddresses(Collections.singletonMap(leaderShardID.toString(), "akka://test/user/" + leaderShardID.toString())) - .schemaContext(SCHEMA_CONTEXT).props() + .schemaContextProvider(() -> SCHEMA_CONTEXT).props() .withDispatcher(Dispatchers.DefaultDispatcherId()), followerShardID.toString()); final TestActorRef leaderShard = actorFactory .createTestActor(Shard.builder().id(leaderShardID).datastoreContext(newDatastoreContext()) .peerAddresses(Collections.singletonMap(followerShardID.toString(), "akka://test/user/" + followerShardID.toString())) - .schemaContext(SCHEMA_CONTEXT).props() + .schemaContextProvider(() -> SCHEMA_CONTEXT).props() .withDispatcher(Dispatchers.DefaultDispatcherId()), leaderShardID.toString()); leaderShard.tell(TimeoutNow.INSTANCE, ActorRef.noSender()); @@ -2192,9 +2189,9 @@ public class ShardTest extends AbstractShardTest { followerShard.tell( new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.BASE, true), getRef()); - final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"), - RegisterChangeListenerReply.class); - assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath()); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterDataTreeNotificationListenerReply.class); + assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); writeToStore(followerShard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); @@ -2204,10 +2201,10 @@ public class ShardTest extends AbstractShardTest { } @Test - public void testClusteredDataTreeChangeListenerDelayedRegistration() throws Exception { + public void testClusteredDataTreeChangeListenerWithDelayedRegistration() throws Exception { new ShardTestKit(getSystem()) { { - final String testName = "testClusteredDataTreeChangeListenerDelayedRegistration"; + final String testName = "testClusteredDataTreeChangeListenerWithDelayedRegistration"; dataStoreContextBuilder.shardElectionTimeoutFactor(1000) .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()); @@ -2224,8 +2221,8 @@ public class ShardTest extends AbstractShardTest { waitUntilNoLeader(shard); shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), getRef()); - final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"), - RegisterDataTreeChangeListenerReply.class); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); shard.tell(DatastoreContext.newBuilderFrom(dataStoreContextBuilder.build()) @@ -2236,6 +2233,43 @@ public class ShardTest extends AbstractShardTest { }; } + @Test + public void testClusteredDataTreeChangeListenerWithDelayedRegistrationClosed() throws Exception { + new ShardTestKit(getSystem()) { + { + final String testName = "testClusteredDataTreeChangeListenerWithDelayedRegistrationClosed"; + dataStoreContextBuilder.shardElectionTimeoutFactor(1000) + .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()); + + final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(0); + final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener, + TestModel.TEST_PATH), actorFactory.generateActorId(testName + "-DataTreeChangeListener")); + + setupInMemorySnapshotStore(); + + final TestActorRef shard = actorFactory.createTestActor( + newShardBuilder().props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(testName + "-shard")); + + waitUntilNoLeader(shard); + + shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), getRef()); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterDataTreeNotificationListenerReply.class); + assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); + + final ActorSelection regActor = getSystem().actorSelection(reply.getListenerRegistrationPath()); + regActor.tell(CloseDataTreeNotificationListenerRegistration.getInstance(), getRef()); + expectMsgClass(CloseDataTreeNotificationListenerRegistrationReply.class); + + shard.tell(DatastoreContext.newBuilderFrom(dataStoreContextBuilder.build()) + .customRaftPolicyImplementation(null).build(), ActorRef.noSender()); + + listener.expectNoMoreChanges("Received unexpected change after close"); + } + }; + } + @Test public void testClusteredDataTreeChangeListenerRegistration() throws Exception { new ShardTestKit(getSystem()) { @@ -2252,14 +2286,14 @@ public class ShardTest extends AbstractShardTest { .datastoreContext(dataStoreContextBuilder.shardElectionTimeoutFactor(1000).build()) .peerAddresses(Collections.singletonMap(leaderShardID.toString(), "akka://test/user/" + leaderShardID.toString())) - .schemaContext(SCHEMA_CONTEXT).props() + .schemaContextProvider(() -> SCHEMA_CONTEXT).props() .withDispatcher(Dispatchers.DefaultDispatcherId()), followerShardID.toString()); final TestActorRef leaderShard = actorFactory .createTestActor(Shard.builder().id(leaderShardID).datastoreContext(newDatastoreContext()) .peerAddresses(Collections.singletonMap(followerShardID.toString(), "akka://test/user/" + followerShardID.toString())) - .schemaContext(SCHEMA_CONTEXT).props() + .schemaContextProvider(() -> SCHEMA_CONTEXT).props() .withDispatcher(Dispatchers.DefaultDispatcherId()), leaderShardID.toString()); leaderShard.tell(TimeoutNow.INSTANCE, ActorRef.noSender()); @@ -2272,8 +2306,8 @@ public class ShardTest extends AbstractShardTest { actorFactory.generateActorId(testName + "-DataTreeChangeListener")); followerShard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), getRef()); - final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"), - RegisterDataTreeChangeListenerReply.class); + final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); writeToStore(followerShard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));