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=2c526288b5ff9353145d4db6dd51d581783111d9;hb=220bfdf6b68a02b2ad280000d6699bf17239cc90;hp=557a75a075e80cc8979acf35d3242ba16c5c2481;hpb=b525d4eee599fce34a98e8e7e06d513b895c81c8;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 557a75a075..2c526288b5 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 @@ -10,7 +10,23 @@ import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; -import static org.opendaylight.controller.cluster.datastore.messages.CreateTransaction.CURRENT_CLIENT_VERSION; +import static org.opendaylight.controller.cluster.datastore.messages.CreateTransaction.CURRENT_VERSION; +import akka.actor.ActorRef; +import akka.actor.PoisonPill; +import akka.actor.Props; +import akka.dispatch.Dispatchers; +import akka.dispatch.OnComplete; +import akka.japi.Creator; +import akka.pattern.Patterns; +import akka.testkit.TestActorRef; +import akka.util.Timeout; +import com.google.common.base.Function; +import com.google.common.base.Optional; +import com.google.common.util.concurrent.CheckedFuture; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.Uninterruptibles; import java.io.IOException; import java.util.Collections; import java.util.HashSet; @@ -23,6 +39,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.junit.After; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.mockito.InOrder; import org.mockito.invocation.InvocationOnMock; @@ -59,6 +76,7 @@ import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot; import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; import org.opendaylight.controller.cluster.raft.client.messages.FindLeader; import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply; +import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; @@ -84,23 +102,6 @@ import org.opendaylight.yangtools.yang.model.api.SchemaContext; import scala.concurrent.Await; import scala.concurrent.Future; import scala.concurrent.duration.FiniteDuration; -import akka.actor.ActorRef; -import akka.actor.PoisonPill; -import akka.actor.Props; -import akka.dispatch.Dispatchers; -import akka.dispatch.OnComplete; -import akka.japi.Creator; -import akka.pattern.Patterns; -import akka.testkit.TestActorRef; -import akka.util.Timeout; -import com.google.common.base.Function; -import com.google.common.base.Optional; -import com.google.common.util.concurrent.CheckedFuture; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.MoreExecutors; -import com.google.common.util.concurrent.Uninterruptibles; - public class ShardTest extends AbstractActorTest { @@ -306,6 +307,7 @@ public class ShardTest extends AbstractActorTest { }}; } + @SuppressWarnings("serial") @Test public void testPeerAddressResolved() throws Exception { new ShardTestKit(getSystem()) {{ @@ -432,9 +434,9 @@ public class ShardTest extends AbstractActorTest { ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), SCHEMA_CONTEXT)))); - int nListEntries = 11; + int nListEntries = 16; Set listEntryKeys = new HashSet<>(); - for(int i = 1; i <= nListEntries; i++) { + for(int i = 1; i <= nListEntries-5; i++) { listEntryKeys.add(Integer.valueOf(i)); YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH) .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build(); @@ -445,6 +447,19 @@ public class ShardTest extends AbstractActorTest { newPayload(mod))); } + // Add some of the new CompositeModificationByteStringPayload + for(int i = 11; i <= nListEntries; i++) { + listEntryKeys.add(Integer.valueOf(i)); + YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH) + .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build(); + Modification mod = new MergeModification(path, + ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i), + SCHEMA_CONTEXT); + InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1, + newByteStringPayload(mod))); + } + + InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1, new ApplyLogEntries(nListEntries)); @@ -507,7 +522,7 @@ public class ShardTest extends AbstractActorTest { shard.tell(PoisonPill.getInstance(), ActorRef.noSender()); } - private CompositeModificationPayload newPayload(Modification... mods) { + private CompositeModificationPayload newPayload(final Modification... mods) { MutableCompositeModification compMod = new MutableCompositeModification(); for(Modification mod: mods) { compMod.addModification(mod); @@ -516,15 +531,25 @@ public class ShardTest extends AbstractActorTest { return new CompositeModificationPayload(compMod.toSerializable()); } - private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(String cohortName, - InMemoryDOMDataStore dataStore, YangInstanceIdentifier path, NormalizedNode data, - MutableCompositeModification modification) { + private CompositeModificationByteStringPayload newByteStringPayload(final Modification... mods) { + MutableCompositeModification compMod = new MutableCompositeModification(); + for(Modification mod: mods) { + compMod.addModification(mod); + } + + return new CompositeModificationByteStringPayload(compMod.toSerializable()); + } + + + private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName, + final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode data, + final MutableCompositeModification modification) { return setupMockWriteTransaction(cohortName, dataStore, path, data, modification, null); } - private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(String cohortName, - InMemoryDOMDataStore dataStore, YangInstanceIdentifier path, NormalizedNode data, - MutableCompositeModification modification, + private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName, + final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode data, + final MutableCompositeModification modification, final Function> preCommit) { DOMStoreWriteTransaction tx = dataStore.newWriteOnlyTransaction(); @@ -534,14 +559,14 @@ public class ShardTest extends AbstractActorTest { doAnswer(new Answer>() { @Override - public ListenableFuture answer(InvocationOnMock invocation) { + public ListenableFuture answer(final InvocationOnMock invocation) { return realCohort.canCommit(); } }).when(cohort).canCommit(); doAnswer(new Answer>() { @Override - public ListenableFuture answer(InvocationOnMock invocation) throws Throwable { + public ListenableFuture answer(final InvocationOnMock invocation) throws Throwable { if(preCommit != null) { return preCommit.apply(realCohort); } else { @@ -552,14 +577,14 @@ public class ShardTest extends AbstractActorTest { doAnswer(new Answer>() { @Override - public ListenableFuture answer(InvocationOnMock invocation) throws Throwable { + public ListenableFuture answer(final InvocationOnMock invocation) throws Throwable { return realCohort.commit(); } }).when(cohort).commit(); doAnswer(new Answer>() { @Override - public ListenableFuture answer(InvocationOnMock invocation) throws Throwable { + public ListenableFuture answer(final InvocationOnMock invocation) throws Throwable { return realCohort.abort(); } }).when(cohort).abort(); @@ -610,7 +635,7 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction message for the first Tx that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, cohort1, modification1, true), getRef()); ReadyTransactionReply readyReply = ReadyTransactionReply.fromSerializable( expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS)); @@ -625,11 +650,11 @@ public class ShardTest extends AbstractActorTest { // Send the ForwardedReadyTransaction for the next 2 Tx's. - shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION, cohort3, modification3, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); @@ -656,12 +681,12 @@ public class ShardTest extends AbstractActorTest { class OnFutureComplete extends OnComplete { private final Class expRespType; - OnFutureComplete(Class expRespType) { + OnFutureComplete(final Class expRespType) { this.expRespType = expRespType; } @Override - public void onComplete(Throwable error, Object resp) { + public void onComplete(final Throwable error, final Object resp) { if(error != null) { caughtEx.set(new AssertionError(getClass().getSimpleName() + " failure", error)); } else { @@ -674,7 +699,7 @@ public class ShardTest extends AbstractActorTest { } } - void onSuccess(Object resp) throws Exception { + void onSuccess(final Object resp) throws Exception { } } @@ -684,7 +709,7 @@ public class ShardTest extends AbstractActorTest { } @Override - public void onComplete(Throwable error, Object resp) { + public void onComplete(final Throwable error, final Object resp) { super.onComplete(error, resp); commitLatch.countDown(); } @@ -693,13 +718,13 @@ public class ShardTest extends AbstractActorTest { class OnCanCommitFutureComplete extends OnFutureComplete { private final String transactionID; - OnCanCommitFutureComplete(String transactionID) { + OnCanCommitFutureComplete(final String transactionID) { super(CanCommitTransactionReply.SERIALIZABLE_CLASS); this.transactionID = transactionID; } @Override - void onSuccess(Object resp) throws Exception { + void onSuccess(final Object resp) throws Exception { CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(resp); assertEquals("Can commit", true, canCommitReply.getCanCommit()); @@ -794,11 +819,11 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction messages that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, cohort1, modification1, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); @@ -826,7 +851,7 @@ public class ShardTest extends AbstractActorTest { final CountDownLatch latch = new CountDownLatch(1); canCommitFuture.onComplete(new OnComplete() { @Override - public void onComplete(Throwable t, Object resp) { + public void onComplete(final Throwable t, final Object resp) { latch.countDown(); } }, getSystem().dispatcher()); @@ -863,7 +888,7 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction messages that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, cohort, modification, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); @@ -907,7 +932,7 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction messages that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, cohort, modification, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); @@ -921,6 +946,7 @@ public class ShardTest extends AbstractActorTest { } @Test + @Ignore("This test will work only if replication is turned on. Needs modification due to optimizations added to Shard/RaftActor.") public void testAbortBeforeFinishCommit() throws Throwable { new ShardTestKit(getSystem()) {{ final TestActorRef shard = TestActorRef.create(getSystem(), @@ -946,7 +972,7 @@ public class ShardTest extends AbstractActorTest { new AbortTransaction(transactionID).toSerializable(), timeout); abortFuture.onComplete(new OnComplete() { @Override - public void onComplete(Throwable e, Object resp) { + public void onComplete(final Throwable e, final Object resp) { abortComplete.countDown(); } }, getSystem().dispatcher()); @@ -960,7 +986,7 @@ public class ShardTest extends AbstractActorTest { TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification, preCommit); - shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, cohort, modification, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); @@ -1025,11 +1051,11 @@ public class ShardTest extends AbstractActorTest { // Ready the Tx's - shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, cohort1, modification1, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); @@ -1089,15 +1115,15 @@ public class ShardTest extends AbstractActorTest { // Ready the Tx's - shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, cohort1, modification1, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION, cohort3, modification3, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); @@ -1161,11 +1187,11 @@ public class ShardTest extends AbstractActorTest { // Simulate the ForwardedReadyTransaction messages that would be sent // by the ShardTransaction. - shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION, cohort1, modification1, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); - shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_CLIENT_VERSION, + shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION, cohort2, modification2, true), getRef()); expectMsgClass(duration, ReadyTransactionReply.SERIALIZABLE_CLASS); @@ -1193,7 +1219,7 @@ public class ShardTest extends AbstractActorTest { final CountDownLatch latch = new CountDownLatch(1); canCommitFuture.onComplete(new OnComplete() { @Override - public void onComplete(Throwable t, Object resp) { + public void onComplete(final Throwable t, final Object resp) { latch.countDown(); } }, getSystem().dispatcher()); @@ -1218,7 +1244,8 @@ public class ShardTest extends AbstractActorTest { testCreateSnapshot(false, "testCreateSnapshotWithNonPersistentData"); } - public void testCreateSnapshot(boolean persistent, final String shardActorName) throws IOException, InterruptedException { + @SuppressWarnings("serial") + public void testCreateSnapshot(final boolean persistent, final String shardActorName) throws IOException, InterruptedException { final DatastoreContext dataStoreContext = DatastoreContext.newBuilder(). shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(persistent).build(); @@ -1230,7 +1257,7 @@ public class ShardTest extends AbstractActorTest { return new Shard(shardID, Collections.emptyMap(), dataStoreContext, SCHEMA_CONTEXT) { @Override - protected void commitSnapshot(long sequenceNumber) { + protected void commitSnapshot(final long sequenceNumber) { super.commitSnapshot(sequenceNumber); latch.get().countDown(); } @@ -1262,8 +1289,7 @@ public class ShardTest extends AbstractActorTest { */ @Test public void testInMemoryDataStoreRestore() throws ReadFailedException { - InMemoryDOMDataStore store = new InMemoryDOMDataStore("test", MoreExecutors.listeningDecorator( - MoreExecutors.sameThreadExecutor()), MoreExecutors.sameThreadExecutor()); + InMemoryDOMDataStore store = new InMemoryDOMDataStore("test", MoreExecutors.sameThreadExecutor()); store.onGlobalContextUpdated(SCHEMA_CONTEXT); @@ -1273,7 +1299,7 @@ public class ShardTest extends AbstractActorTest { commitTransaction(putTransaction); - NormalizedNode expected = readStore(store); + NormalizedNode expected = readStore(store); DOMStoreWriteTransaction writeTransaction = store.newWriteOnlyTransaction(); @@ -1282,10 +1308,9 @@ public class ShardTest extends AbstractActorTest { commitTransaction(writeTransaction); - NormalizedNode actual = readStore(store); + NormalizedNode actual = readStore(store); assertEquals(expected, actual); - } @Test @@ -1323,7 +1348,7 @@ public class ShardTest extends AbstractActorTest { } - private NormalizedNode readStore(InMemoryDOMDataStore store) throws ReadFailedException { + private NormalizedNode readStore(final InMemoryDOMDataStore store) throws ReadFailedException { DOMStoreReadTransaction transaction = store.newReadOnlyTransaction(); CheckedFuture>, ReadFailedException> read = transaction.read(YangInstanceIdentifier.builder().build()); @@ -1337,7 +1362,7 @@ public class ShardTest extends AbstractActorTest { return normalizedNode; } - private void commitTransaction(DOMStoreWriteTransaction transaction) { + private void commitTransaction(final DOMStoreWriteTransaction transaction) { DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready(); ListenableFuture future = commitCohort.preCommit(); @@ -1353,13 +1378,13 @@ public class ShardTest extends AbstractActorTest { return new AsyncDataChangeListener>() { @Override public void onDataChanged( - AsyncDataChangeEvent> change) { + final AsyncDataChangeEvent> change) { } }; } - static NormalizedNode readStore(TestActorRef shard, YangInstanceIdentifier id) + static NormalizedNode readStore(final TestActorRef shard, final YangInstanceIdentifier id) throws ExecutionException, InterruptedException { DOMStoreReadTransaction transaction = shard.underlyingActor().getDataStore().newReadOnlyTransaction(); @@ -1374,7 +1399,7 @@ public class ShardTest extends AbstractActorTest { return node; } - private void writeToStore(TestActorRef shard, YangInstanceIdentifier id, NormalizedNode node) + private void writeToStore(final TestActorRef shard, final YangInstanceIdentifier id, final NormalizedNode node) throws ExecutionException, InterruptedException { DOMStoreWriteTransaction transaction = shard.underlyingActor().getDataStore().newWriteOnlyTransaction(); @@ -1385,10 +1410,11 @@ public class ShardTest extends AbstractActorTest { commitCohort.commit().get(); } + @SuppressWarnings("serial") private static final class DelegatingShardCreator implements Creator { private final Creator delegate; - DelegatingShardCreator(Creator delegate) { + DelegatingShardCreator(final Creator delegate) { this.delegate = delegate; }