X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardTest.java;h=445116246bdb83e87cd16d923ac5d0385cdb9532;hp=45cfd29d255af37b91df19553261570079c65081;hb=7426d405093265655b05c6a3eb197362266edf2e;hpb=013a6679470bf692753f2e04ab4398c97fd9f5d0 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 45cfd29d25..445116246b 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; @@ -105,6 +107,7 @@ import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelpe import org.opendaylight.controller.md.cluster.datastore.model.TestModel; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker; import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; +import org.opendaylight.yangtools.concepts.Identifier; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; import org.opendaylight.yangtools.yang.data.api.schema.MapNode; @@ -372,8 +375,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()))); } }; } @@ -393,8 +397,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()))); } }; } @@ -621,7 +626,6 @@ public class ShardTest extends AbstractShardTest { final ReadyTransactionReply readyReply = ReadyTransactionReply .fromSerializable(expectMsgClass(duration, ReadyTransactionReply.class)); assertEquals("Cohort path", shard.path().toString(), readyReply.getCohortPath()); - // Send the CanCommitTransaction message for the first Tx. shard.tell(new CanCommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), getRef()); @@ -1445,15 +1449,15 @@ public class ShardTest extends AbstractShardTest { { final Creator creator = () -> new Shard(newShardBuilder()) { @Override - void persistPayload(final TransactionIdentifier transactionId, final Payload payload, - boolean batchHint) { + void persistPayload(final Identifier id, final Payload payload, + final boolean batchHint) { // Simulate an AbortTransaction message occurring during // replication, after // persisting and before finishing the commit to the // in-memory store. - doAbortTransaction(transactionId, null); - super.persistPayload(transactionId, payload, batchHint); + doAbortTransaction(id, null); + super.persistPayload(id, payload, batchHint); } }; @@ -2016,13 +2020,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 +2124,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()); @@ -2170,14 +2174,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()); @@ -2204,10 +2208,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()); @@ -2236,6 +2240,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 RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"), + RegisterDataTreeChangeListenerReply.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 +2293,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());