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=6429307dc1d3c8cca783eddcecdc1e22c5f710d3;hb=refs%2Fchanges%2F71%2F85371%2F3;hp=a2509fa46c49cb272d0cbd2ce8c26fb36f4bf34f;hpb=466078ab1dc8a8cc2981b161051f6edecd6af85a;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 a2509fa46c..6429307dc1 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 @@ -13,7 +13,7 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -import static org.mockito.Matchers.any; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; @@ -33,6 +33,7 @@ import akka.util.Timeout; import com.google.common.base.Stopwatch; import com.google.common.base.Throwables; import com.google.common.util.concurrent.Uninterruptibles; +import java.time.Duration; import java.util.Collections; import java.util.HashSet; import java.util.Map; @@ -139,7 +140,7 @@ public class ShardTest extends AbstractShardTest { shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, false), testKit.getRef()); - final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(testKit.duration("3 seconds"), + final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(Duration.ofSeconds(3), RegisterDataTreeNotificationListenerReply.class); final String replyPath = reply.getListenerRegistrationPath().toString(); assertTrue("Incorrect reply path: " + replyPath, @@ -149,6 +150,18 @@ public class ShardTest extends AbstractShardTest { writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); listener.waitForChangeEvents(); + listener.verifyOnInitialDataEvent(); + + final MockDataTreeChangeListener listener2 = new MockDataTreeChangeListener(1); + final ActorRef dclActor2 = actorFactory.createActor(DataTreeChangeListenerActor.props(listener2, + TestModel.TEST_PATH), "testRegisterDataTreeChangeListener-DataTreeChangeListener2"); + + shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor2, false), testKit.getRef()); + + testKit.expectMsgClass(Duration.ofSeconds(3), RegisterDataTreeNotificationListenerReply.class); + + listener2.waitForChangeEvents(); + listener2.verifyNoOnInitialDataEvent(); } @SuppressWarnings("serial") @@ -156,7 +169,7 @@ public class ShardTest extends AbstractShardTest { public void testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration() throws Exception { final CountDownLatch onFirstElectionTimeout = new CountDownLatch(1); final CountDownLatch onChangeListenerRegistered = new CountDownLatch(1); - final Creator creator = new Creator() { + final Creator creator = new Creator<>() { boolean firstElectionTimeout = true; @Override @@ -189,21 +202,20 @@ public class ShardTest extends AbstractShardTest { final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener, path), "testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration-DataChangeListener"); - final TestActorRef shard = actorFactory.createTestActor( - Props.create(new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()), + final TestActorRef shard = actorFactory.createTestActor(Props.create(Shard.class, + new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()), "testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration"); final ShardTestKit testKit = new ShardTestKit(getSystem()); assertTrue("Got first ElectionTimeout", onFirstElectionTimeout.await(5, TimeUnit.SECONDS)); shard.tell(new RegisterDataTreeChangeListener(path, dclActor, false), testKit.getRef()); - final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(testKit.duration("5 seconds"), + final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath()); shard.tell(FindLeader.INSTANCE, testKit.getRef()); - final FindLeaderReply findLeadeReply = testKit.expectMsgClass(testKit.duration("5 seconds"), - FindLeaderReply.class); + final FindLeaderReply findLeadeReply = testKit.expectMsgClass(Duration.ofSeconds(5), FindLeaderReply.class); assertFalse("Expected the shard not to be the leader", findLeadeReply.getLeaderActor().isPresent()); onChangeListenerRegistered.countDown(); @@ -224,7 +236,7 @@ public class ShardTest extends AbstractShardTest { shard.tell(new CreateTransaction(nextTransactionId(), TransactionType.READ_ONLY.ordinal(), DataStoreVersions.CURRENT_VERSION).toSerializable(), testKit.getRef()); - final CreateTransactionReply reply = testKit.expectMsgClass(testKit.duration("3 seconds"), + final CreateTransactionReply reply = testKit.expectMsgClass(Duration.ofSeconds(3), CreateTransactionReply.class); final String path = reply.getTransactionPath().toString(); @@ -243,7 +255,7 @@ public class ShardTest extends AbstractShardTest { shard.tell(new CreateTransaction(nextTransactionId(), TransactionType.READ_ONLY.ordinal(), DataStoreVersions.CURRENT_VERSION).toSerializable(), testKit.getRef()); - final CreateTransactionReply reply = testKit.expectMsgClass(testKit.duration("3 seconds"), + final CreateTransactionReply reply = testKit.expectMsgClass(Duration.ofSeconds(3), CreateTransactionReply.class); final String path = reply.getTransactionPath().toString(); @@ -287,7 +299,7 @@ public class ShardTest extends AbstractShardTest { writeToStore(store, TestModel.TEST_PATH, container); - final YangInstanceIdentifier root = YangInstanceIdentifier.EMPTY; + final YangInstanceIdentifier root = YangInstanceIdentifier.empty(); final NormalizedNode expected = readStore(store, root); final Snapshot snapshot = Snapshot.create(new ShardSnapshotState(new MetadataShardDataTreeSnapshot(expected)), @@ -387,8 +399,8 @@ public class ShardTest extends AbstractShardTest { final CountDownLatch commitLatch = new CountDownLatch(2); final long timeoutSec = 5; - final FiniteDuration duration = FiniteDuration.create(timeoutSec, TimeUnit.SECONDS); - final Timeout timeout = new Timeout(duration); + final Duration duration = Duration.ofSeconds(timeoutSec); + final Timeout timeout = Timeout.create(duration); final TestActorRef shard = actorFactory.createTestActor( newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()), @@ -539,7 +551,7 @@ public class ShardTest extends AbstractShardTest { verifyOuterListEntry(shard, 1); - verifyLastApplied(shard, 5); + verifyLastApplied(shard, 3); } @Test @@ -552,7 +564,7 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); final TransactionIdentifier transactionID = nextTransactionId(); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); // Send a BatchedModifications to start a transaction. @@ -601,7 +613,7 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); final TransactionIdentifier transactionID = nextTransactionId(); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); // Send a BatchedModifications to start a transaction. @@ -645,7 +657,7 @@ public class ShardTest extends AbstractShardTest { shard.tell(batched, testKit.getRef()); - final Failure failure = testKit.expectMsgClass(testKit.duration("5 seconds"), Failure.class); + final Failure failure = testKit.expectMsgClass(Duration.ofSeconds(5), Failure.class); if (failure != null) { Throwables.propagateIfPossible(failure.cause(), Exception.class); @@ -675,7 +687,7 @@ public class ShardTest extends AbstractShardTest { BatchedModifications batched = new BatchedModifications(transactionID, CURRENT_VERSION); batched.addModification(new MergeModification(TestModel.TEST_PATH, invalidData)); shard.tell(batched, testKit.getRef()); - Failure failure = testKit.expectMsgClass(testKit.duration("5 seconds"), akka.actor.Status.Failure.class); + Failure failure = testKit.expectMsgClass(Duration.ofSeconds(5), akka.actor.Status.Failure.class); final Throwable cause = failure.cause(); @@ -685,7 +697,7 @@ public class ShardTest extends AbstractShardTest { shard.tell(batched, testKit.getRef()); - failure = testKit.expectMsgClass(testKit.duration("5 seconds"), akka.actor.Status.Failure.class); + failure = testKit.expectMsgClass(Duration.ofSeconds(5), akka.actor.Status.Failure.class); assertEquals("Failure cause", cause, failure.cause()); } @@ -702,7 +714,7 @@ public class ShardTest extends AbstractShardTest { final TransactionIdentifier transactionID1 = new TransactionIdentifier(historyId, 0); final TransactionIdentifier transactionID2 = new TransactionIdentifier(historyId, 1); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); // Send a BatchedModifications to start a chained write // transaction and ready it. @@ -717,12 +729,12 @@ public class ShardTest extends AbstractShardTest { shard.tell(new CreateTransaction(transactionID2, TransactionType.READ_ONLY.ordinal(), DataStoreVersions.CURRENT_VERSION).toSerializable(), testKit.getRef()); - final CreateTransactionReply createReply = testKit.expectMsgClass(testKit.duration("3 seconds"), + final CreateTransactionReply createReply = testKit.expectMsgClass(Duration.ofSeconds(3), CreateTransactionReply.class); getSystem().actorSelection(createReply.getTransactionPath()) .tell(new ReadData(path, DataStoreVersions.CURRENT_VERSION), testKit.getRef()); - final ReadDataReply readReply = testKit.expectMsgClass(testKit.duration("3 seconds"), ReadDataReply.class); + final ReadDataReply readReply = testKit.expectMsgClass(Duration.ofSeconds(3), ReadDataReply.class); assertEquals("Read node", containerNode, readReply.getNormalizedNode()); // Commit the write transaction. @@ -745,7 +757,7 @@ public class ShardTest extends AbstractShardTest { public void testOnBatchedModificationsWhenNotLeader() { final AtomicBoolean overrideLeaderCalls = new AtomicBoolean(); final ShardTestKit testKit = new ShardTestKit(getSystem()); - final Creator creator = new Creator() { + final Creator creator = new Creator<>() { private static final long serialVersionUID = 1L; @Override @@ -765,8 +777,8 @@ public class ShardTest extends AbstractShardTest { } }; - final TestActorRef shard = actorFactory.createTestActor(Props - .create(new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()), + final TestActorRef shard = actorFactory.createTestActor(Props.create(Shard.class, + new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()), "testOnBatchedModificationsWhenNotLeader"); ShardTestKit.waitUntilLeader(shard); @@ -836,7 +848,7 @@ public class ShardTest extends AbstractShardTest { testKit.getRef()); } - testKit.expectMsgClass(testKit.duration("5 seconds"), CommitTransactionReply.class); + testKit.expectMsgClass(Duration.ofSeconds(5), CommitTransactionReply.class); final NormalizedNode actualNode = readStore(shard, TestModel.TEST_PATH); assertEquals(TestModel.TEST_QNAME.getLocalName(), containerNode, actualNode); @@ -932,7 +944,7 @@ public class ShardTest extends AbstractShardTest { // Setup a simulated transactions with a mock cohort. - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); final TransactionIdentifier transactionID = nextTransactionId(); final NormalizedNode containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -975,7 +987,7 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); final TransactionIdentifier transactionID = nextTransactionId(); if (readWrite) { @@ -1014,8 +1026,7 @@ public class ShardTest extends AbstractShardTest { // Committed transaction count should increase as usual assertEquals(1, shardStats.getCommittedTransactionsCount()); - // Commit index should advance as we do not have an empty - // modification + // Commit index should advance 1 to account for disabling metadata assertEquals(1, shardStats.getCommitIndex()); } @@ -1029,8 +1040,8 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); - final Timeout timeout = new Timeout(duration); + final Duration duration = Duration.ofSeconds(5); + final Timeout timeout = Timeout.create(duration); // Setup 2 simulated transactions with mock cohorts. The first // one fails in the @@ -1073,7 +1084,7 @@ public class ShardTest extends AbstractShardTest { // Wait for the 2nd Tx to complete the canCommit phase. final CountDownLatch latch = new CountDownLatch(1); - canCommitFuture.onComplete(new OnComplete() { + canCommitFuture.onComplete(new OnComplete<>() { @Override public void onComplete(final Throwable failure, final Object resp) { latch.countDown(); @@ -1103,8 +1114,8 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); - final Timeout timeout = new Timeout(duration); + final Duration duration = Duration.ofSeconds(5); + final Timeout timeout = Timeout.create(duration); doThrow(new RuntimeException("mock preCommit failure")).when(dataTree) .prepare(any(DataTreeModification.class)); @@ -1143,7 +1154,7 @@ public class ShardTest extends AbstractShardTest { // Wait for the 2nd Tx to complete the canCommit phase. final CountDownLatch latch = new CountDownLatch(1); - canCommitFuture.onComplete(new OnComplete() { + canCommitFuture.onComplete(new OnComplete<>() { @Override public void onComplete(final Throwable failure, final Object resp) { latch.countDown(); @@ -1168,10 +1179,10 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); final TransactionIdentifier transactionID1 = nextTransactionId(); - doThrow(new DataValidationFailedException(YangInstanceIdentifier.EMPTY, "mock canCommit failure")) + doThrow(new DataValidationFailedException(YangInstanceIdentifier.empty(), "mock canCommit failure")) .doNothing().when(dataTree).validate(any(DataTreeModification.class)); shard.tell(newBatchedModifications(transactionID1, TestModel.TEST_PATH, @@ -1212,10 +1223,10 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - doThrow(new DataValidationFailedException(YangInstanceIdentifier.EMPTY, "mock canCommit failure")) + doThrow(new DataValidationFailedException(YangInstanceIdentifier.empty(), "mock canCommit failure")) .doNothing().when(dataTree).validate(any(DataTreeModification.class)); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); final TransactionIdentifier transactionID1 = nextTransactionId(); @@ -1261,13 +1272,13 @@ public class ShardTest extends AbstractShardTest { } }; - final TestActorRef shard = actorFactory.createTestActor(Props - .create(new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()), + final TestActorRef shard = actorFactory.createTestActor(Props.create(Shard.class, + new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()), "testAbortWithCommitPending"); ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); final TransactionIdentifier transactionID = nextTransactionId(); @@ -1301,7 +1312,7 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); writeToStore(shard, TestModel.OUTER_LIST_PATH, @@ -1430,7 +1441,7 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); final TransactionIdentifier transactionID1 = nextTransactionId(); shard.tell(newBatchedModifications(transactionID1, TestModel.TEST_PATH, @@ -1466,7 +1477,7 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); final ShardDataTree dataStore = shard.underlyingActor().getDataStore(); @@ -1521,7 +1532,7 @@ public class ShardTest extends AbstractShardTest { "testCanCommitBeforeReadyFailure"); shard.tell(new CanCommitTransaction(nextTransactionId(), CURRENT_VERSION).toSerializable(), testKit.getRef()); - testKit.expectMsgClass(testKit.duration("5 seconds"), akka.actor.Status.Failure.class); + testKit.expectMsgClass(Duration.ofSeconds(5), akka.actor.Status.Failure.class); } @Test @@ -1532,8 +1543,8 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); - final Timeout timeout = new Timeout(duration); + final Duration duration = Duration.ofSeconds(5); + final Timeout timeout = Timeout.create(duration); // Ready 2 transactions - the first one will be aborted. @@ -1570,7 +1581,8 @@ public class ShardTest extends AbstractShardTest { // Wait for the 2nd Tx to complete the canCommit phase. - canCommitReply = (CanCommitTransactionReply) Await.result(canCommitFuture, duration); + canCommitReply = (CanCommitTransactionReply) Await.result(canCommitFuture, + FiniteDuration.create(5, TimeUnit.SECONDS)); assertTrue("Can commit", canCommitReply.getCanCommit()); } @@ -1583,7 +1595,7 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); // Ready a tx. @@ -1624,7 +1636,7 @@ public class ShardTest extends AbstractShardTest { ShardTestKit.waitUntilLeader(shard); - final FiniteDuration duration = testKit.duration("5 seconds"); + final Duration duration = Duration.ofSeconds(5); // Ready 3 tx's. @@ -1716,18 +1728,15 @@ public class ShardTest extends AbstractShardTest { } } - final ShardTestKit testKit = new ShardTestKit(getSystem()); - final Creator creator = () -> new TestShard(newShardBuilder()); - final TestActorRef shard = actorFactory.createTestActor(Props - .create(new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()), - shardActorName); + final TestActorRef shard = actorFactory.createTestActor(Props.create(Shard.class, + new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()), shardActorName); ShardTestKit.waitUntilLeader(shard); writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); - final NormalizedNode expectedRoot = readStore(shard, YangInstanceIdentifier.EMPTY); + final NormalizedNode expectedRoot = readStore(shard, YangInstanceIdentifier.empty()); // Trigger creation of a snapshot by ensuring final RaftActorContext raftActorContext = ((TestShard) shard.underlyingActor()).getRaftActorContext(); @@ -1770,16 +1779,16 @@ public class ShardTest extends AbstractShardTest { commitTransaction(store, putTransaction); - final NormalizedNode expected = readStore(store, YangInstanceIdentifier.EMPTY); + final NormalizedNode expected = readStore(store, YangInstanceIdentifier.empty()); final DataTreeModification writeTransaction = store.takeSnapshot().newModification(); - writeTransaction.delete(YangInstanceIdentifier.EMPTY); - writeTransaction.write(YangInstanceIdentifier.EMPTY, expected); + writeTransaction.delete(YangInstanceIdentifier.empty()); + writeTransaction.write(YangInstanceIdentifier.empty(), expected); commitTransaction(store, writeTransaction); - final NormalizedNode actual = readStore(store, YangInstanceIdentifier.EMPTY); + final NormalizedNode actual = readStore(store, YangInstanceIdentifier.empty()); assertEquals(expected, actual); } @@ -1895,7 +1904,7 @@ public class ShardTest extends AbstractShardTest { testKit.waitUntilNoLeader(shard); shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), testKit.getRef()); - final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(testKit.duration("5 seconds"), + final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); @@ -1925,7 +1934,7 @@ public class ShardTest extends AbstractShardTest { testKit.waitUntilNoLeader(shard); shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), testKit.getRef()); - final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(testKit.duration("5 seconds"), + final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath()); @@ -1974,7 +1983,7 @@ public class ShardTest extends AbstractShardTest { actorFactory.generateActorId(testName + "-DataTreeChangeListener")); followerShard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), testKit.getRef()); - final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(testKit.duration("5 seconds"), + final RegisterDataTreeNotificationListenerReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), RegisterDataTreeNotificationListenerReply.class); assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());