opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardDataTreeTest.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertNotNull;
  12 import static org.junit.Assert.assertTrue;
  13 import static org.junit.Assert.fail;
  14 import static org.mockito.ArgumentMatchers.any;
  15 import static org.mockito.ArgumentMatchers.anyBoolean;
  16 import static org.mockito.ArgumentMatchers.eq;
  17 import static org.mockito.Mockito.atLeastOnce;
  18 import static org.mockito.Mockito.doNothing;
  19 import static org.mockito.Mockito.doReturn;
  20 import static org.mockito.Mockito.inOrder;
  21 import static org.mockito.Mockito.mock;
  22 import static org.mockito.Mockito.never;
  23 import static org.mockito.Mockito.reset;
  24 import static org.mockito.Mockito.verify;
  25 import static org.mockito.Mockito.verifyNoMoreInteractions;
  26 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedCanCommit;
  27 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedCommit;
  28 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedPreCommit;
  29 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediate3PhaseCommit;
  30 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediateCanCommit;
  31 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediateCommit;
  32 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediatePayloadReplication;
  33 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediatePreCommit;
  34
  35 import com.google.common.base.Ticker;
  36 import com.google.common.primitives.UnsignedLong;
  37 import com.google.common.util.concurrent.FutureCallback;
  38 import java.io.IOException;
  39 import java.math.BigInteger;
  40 import java.util.ArrayList;
  41 import java.util.Collection;
  42 import java.util.HashMap;
  43 import java.util.List;
  44 import java.util.Map;
  45 import java.util.Optional;
  46 import java.util.function.Consumer;
  47 import org.junit.Before;
  48 import org.junit.Test;
  49 import org.mockito.ArgumentCaptor;
  50 import org.mockito.InOrder;
  51 import org.mockito.Mockito;
  52 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
  53 import org.opendaylight.controller.cluster.datastore.persisted.CommitTransactionPayload;
  54 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
  55 import org.opendaylight.controller.cluster.datastore.persisted.PayloadVersion;
  56 import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
  57 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  58 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
  59 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  60 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
  61 import org.opendaylight.yangtools.yang.common.Uint64;
  62 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  63 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  64 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
  65 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  66 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  67 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  68 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  69 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  70 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
  71 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration;
  72 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  73 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
  74 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
  75 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
  76 import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
  77 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
  78 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  79 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
  80 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
  81 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  82
  83 public class ShardDataTreeTest extends AbstractTest {
  84     private static final DatastoreContext DATASTORE_CONTEXT = DatastoreContext.newBuilder().build();
  85
  86     private final Shard mockShard = Mockito.mock(Shard.class);
  87     private ShardDataTree shardDataTree;
  88     private EffectiveModelContext fullSchema;
  89
  90     @Before
  91     public void setUp() {
  92         doReturn(Ticker.systemTicker()).when(mockShard).ticker();
  93         doReturn(mock(ShardStats.class)).when(mockShard).getShardMBean();
  94         doReturn(DATASTORE_CONTEXT).when(mockShard).getDatastoreContext();
  95
  96         fullSchema = SchemaContextHelper.full();
  97
  98         shardDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
  99     }
 100
 101     @Test
 102     public void testWrite() {
 103         modify(false, true, true);
 104     }
 105
 106     @Test
 107     public void testMerge() {
 108         modify(true, true, true);
 109     }
 110
 111     private void modify(final boolean merge, final boolean expectedCarsPresent, final boolean expectedPeoplePresent) {
 112         immediatePayloadReplication(shardDataTree, mockShard);
 113
 114         assertEquals(fullSchema, shardDataTree.getSchemaContext());
 115
 116         final ReadWriteShardDataTreeTransaction transaction =
 117                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 118
 119         final DataTreeModification snapshot = transaction.getSnapshot();
 120
 121         assertNotNull(snapshot);
 122
 123         if (merge) {
 124             snapshot.merge(CarsModel.BASE_PATH, CarsModel.create());
 125             snapshot.merge(PeopleModel.BASE_PATH, PeopleModel.create());
 126         } else {
 127             snapshot.write(CarsModel.BASE_PATH, CarsModel.create());
 128             snapshot.write(PeopleModel.BASE_PATH, PeopleModel.create());
 129         }
 130
 131         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction, Optional.empty());
 132
 133         immediateCanCommit(cohort);
 134         immediatePreCommit(cohort);
 135         immediateCommit(cohort);
 136
 137         final ReadOnlyShardDataTreeTransaction readOnlyShardDataTreeTransaction =
 138                 shardDataTree.newReadOnlyTransaction(nextTransactionId());
 139
 140         final DataTreeSnapshot snapshot1 = readOnlyShardDataTreeTransaction.getSnapshot();
 141
 142         final Optional<NormalizedNode<?, ?>> optional = snapshot1.readNode(CarsModel.BASE_PATH);
 143
 144         assertEquals(expectedCarsPresent, optional.isPresent());
 145
 146         final Optional<NormalizedNode<?, ?>> optional1 = snapshot1.readNode(PeopleModel.BASE_PATH);
 147
 148         assertEquals(expectedPeoplePresent, optional1.isPresent());
 149     }
 150
 151     @Test
 152     public void bug4359AddRemoveCarOnce() {
 153         immediatePayloadReplication(shardDataTree, mockShard);
 154
 155         final List<DataTreeCandidate> candidates = new ArrayList<>();
 156         candidates.add(addCar(shardDataTree));
 157         candidates.add(removeCar(shardDataTree));
 158
 159         final NormalizedNode<?, ?> expected = getCars(shardDataTree);
 160
 161         applyCandidates(shardDataTree, candidates);
 162
 163         final NormalizedNode<?, ?> actual = getCars(shardDataTree);
 164
 165         assertEquals(expected, actual);
 166     }
 167
 168     @Test
 169     public void bug4359AddRemoveCarTwice() {
 170         immediatePayloadReplication(shardDataTree, mockShard);
 171
 172         final List<DataTreeCandidate> candidates = new ArrayList<>();
 173         candidates.add(addCar(shardDataTree));
 174         candidates.add(removeCar(shardDataTree));
 175         candidates.add(addCar(shardDataTree));
 176         candidates.add(removeCar(shardDataTree));
 177
 178         final NormalizedNode<?, ?> expected = getCars(shardDataTree);
 179
 180         applyCandidates(shardDataTree, candidates);
 181
 182         final NormalizedNode<?, ?> actual = getCars(shardDataTree);
 183
 184         assertEquals(expected, actual);
 185     }
 186
 187     @Test
 188     public void testListenerNotifiedOnApplySnapshot() throws Exception {
 189         immediatePayloadReplication(shardDataTree, mockShard);
 190
 191         DOMDataTreeChangeListener listener = mock(DOMDataTreeChangeListener.class);
 192         shardDataTree.registerTreeChangeListener(CarsModel.CAR_LIST_PATH.node(CarsModel.CAR_QNAME), listener,
 193             Optional.empty(), noop -> { });
 194
 195         addCar(shardDataTree, "optima");
 196
 197         verifyOnDataTreeChanged(listener, dtc -> {
 198             assertEquals("getModificationType", ModificationType.WRITE, dtc.getRootNode().getModificationType());
 199             assertEquals("getRootPath", CarsModel.newCarPath("optima"), dtc.getRootPath());
 200         });
 201
 202         addCar(shardDataTree, "sportage");
 203
 204         verifyOnDataTreeChanged(listener, dtc -> {
 205             assertEquals("getModificationType", ModificationType.WRITE, dtc.getRootNode().getModificationType());
 206             assertEquals("getRootPath", CarsModel.newCarPath("sportage"), dtc.getRootPath());
 207         });
 208
 209         ShardDataTree newDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
 210         immediatePayloadReplication(newDataTree, mockShard);
 211         addCar(newDataTree, "optima");
 212         addCar(newDataTree, "murano");
 213
 214         shardDataTree.applySnapshot(newDataTree.takeStateSnapshot());
 215
 216         Map<YangInstanceIdentifier, ModificationType> expChanges = new HashMap<>();
 217         expChanges.put(CarsModel.newCarPath("optima"), ModificationType.WRITE);
 218         expChanges.put(CarsModel.newCarPath("murano"), ModificationType.WRITE);
 219         expChanges.put(CarsModel.newCarPath("sportage"), ModificationType.DELETE);
 220         verifyOnDataTreeChanged(listener, dtc -> {
 221             ModificationType expType = expChanges.remove(dtc.getRootPath());
 222             assertNotNull("Got unexpected change for " + dtc.getRootPath(), expType);
 223             assertEquals("getModificationType", expType, dtc.getRootNode().getModificationType());
 224         });
 225
 226         if (!expChanges.isEmpty()) {
 227             fail("Missing change notifications: " + expChanges);
 228         }
 229     }
 230
 231     @Test
 232     public void testPipelinedTransactionsWithCoordinatedCommits() throws Exception {
 233         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 234             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 235
 236         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 237             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 238
 239         NormalizedNode<?, ?> peopleNode = PeopleModel.create();
 240         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 241             snapshot.write(PeopleModel.BASE_PATH, peopleNode));
 242
 243         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 244         MapEntryNode carNode = CarsModel.newCarEntry("optima", Uint64.valueOf(100));
 245         final ShardDataTreeCohort cohort4 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 246
 247         immediateCanCommit(cohort1);
 248         final FutureCallback<Void> canCommitCallback2 = coordinatedCanCommit(cohort2);
 249         final FutureCallback<Void> canCommitCallback3 = coordinatedCanCommit(cohort3);
 250         final FutureCallback<Void> canCommitCallback4 = coordinatedCanCommit(cohort4);
 251
 252         final FutureCallback<DataTreeCandidate> preCommitCallback1 = coordinatedPreCommit(cohort1);
 253         verify(preCommitCallback1).onSuccess(cohort1.getCandidate());
 254         verify(canCommitCallback2).onSuccess(null);
 255
 256         final FutureCallback<DataTreeCandidate> preCommitCallback2 = coordinatedPreCommit(cohort2);
 257         verify(preCommitCallback2).onSuccess(cohort2.getCandidate());
 258         verify(canCommitCallback3).onSuccess(null);
 259
 260         final FutureCallback<DataTreeCandidate> preCommitCallback3 = coordinatedPreCommit(cohort3);
 261         verify(preCommitCallback3).onSuccess(cohort3.getCandidate());
 262         verify(canCommitCallback4).onSuccess(null);
 263
 264         final FutureCallback<DataTreeCandidate> preCommitCallback4 = coordinatedPreCommit(cohort4);
 265         verify(preCommitCallback4).onSuccess(cohort4.getCandidate());
 266
 267         final FutureCallback<UnsignedLong> commitCallback2 = coordinatedCommit(cohort2);
 268         verify(mockShard, never()).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 269                 anyBoolean());
 270         verifyNoMoreInteractions(commitCallback2);
 271
 272         final FutureCallback<UnsignedLong> commitCallback4 = coordinatedCommit(cohort4);
 273         verify(mockShard, never()).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 274                 anyBoolean());
 275         verifyNoMoreInteractions(commitCallback4);
 276
 277         final FutureCallback<UnsignedLong> commitCallback1 = coordinatedCommit(cohort1);
 278         InOrder inOrder = inOrder(mockShard);
 279         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 280                 eq(true));
 281         inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
 282                 eq(false));
 283         verifyNoMoreInteractions(commitCallback1);
 284         verifyNoMoreInteractions(commitCallback2);
 285
 286         final FutureCallback<UnsignedLong> commitCallback3 = coordinatedCommit(cohort3);
 287         inOrder = inOrder(mockShard);
 288         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 289                 eq(true));
 290         inOrder.verify(mockShard).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 291                 eq(false));
 292         verifyNoMoreInteractions(commitCallback3);
 293         verifyNoMoreInteractions(commitCallback4);
 294
 295         final ShardDataTreeCohort cohort5 = newShardDataTreeCohort(snapshot ->
 296             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 297         final FutureCallback<Void> canCommitCallback5 = coordinatedCanCommit(cohort5);
 298
 299         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 300         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 301                 cohort1.getCandidate());
 302         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 303         shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
 304         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 305         shardDataTree.applyReplicatedPayload(cohort4.getIdentifier(), mockPayload);
 306
 307         inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3, commitCallback4);
 308         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 309         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 310         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 311         inOrder.verify(commitCallback4).onSuccess(any(UnsignedLong.class));
 312
 313         verify(canCommitCallback5).onSuccess(null);
 314
 315         final DataTreeSnapshot snapshot =
 316                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 317         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 318         assertTrue("Car node present", optional.isPresent());
 319         assertEquals("Car node", carNode, optional.get());
 320
 321         optional = snapshot.readNode(PeopleModel.BASE_PATH);
 322         assertTrue("People node present", optional.isPresent());
 323         assertEquals("People node", peopleNode, optional.get());
 324     }
 325
 326     @Test
 327     public void testPipelinedTransactionsWithImmediateCommits() throws Exception {
 328         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 329             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 330
 331         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 332             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 333
 334         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 335         MapEntryNode carNode = CarsModel.newCarEntry("optima", Uint64.valueOf(100));
 336         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 337
 338         final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
 339         final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
 340         final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
 341
 342         InOrder inOrder = inOrder(mockShard);
 343         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 344                 eq(true));
 345         inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
 346                 eq(true));
 347         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 348                 eq(false));
 349
 350         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 351         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 352                 cohort1.getCandidate());
 353         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 354         shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
 355         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 356
 357         inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
 358         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 359         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 360         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 361
 362         final DataTreeSnapshot snapshot =
 363                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 364         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 365         assertTrue("Car node present", optional.isPresent());
 366         assertEquals("Car node", carNode, optional.get());
 367     }
 368
 369     @Test
 370     public void testPipelinedTransactionsWithImmediateReplication() {
 371         immediatePayloadReplication(shardDataTree, mockShard);
 372
 373         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 374             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 375
 376         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 377             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 378
 379         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 380         MapEntryNode carNode = CarsModel.newCarEntry("optima", Uint64.valueOf(100));
 381         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 382
 383         final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
 384         final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
 385         final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
 386
 387         InOrder inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
 388         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 389         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 390         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 391
 392         final DataTreeSnapshot snapshot = shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 393         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(CarsModel.BASE_PATH);
 394         assertTrue("Car node present", optional.isPresent());
 395     }
 396
 397     @SuppressWarnings("unchecked")
 398     @Test
 399     public void testAbortWithPendingCommits() throws Exception {
 400         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 401             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 402
 403         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 404             snapshot.write(PeopleModel.BASE_PATH, PeopleModel.create()));
 405
 406         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 407             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 408
 409         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 410         MapEntryNode carNode = CarsModel.newCarEntry("optima", Uint64.valueOf(100));
 411         final ShardDataTreeCohort cohort4 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 412
 413         coordinatedCanCommit(cohort2);
 414         immediateCanCommit(cohort1);
 415         coordinatedCanCommit(cohort3);
 416         coordinatedCanCommit(cohort4);
 417
 418         coordinatedPreCommit(cohort1);
 419         coordinatedPreCommit(cohort2);
 420         coordinatedPreCommit(cohort3);
 421
 422         FutureCallback<Void> mockAbortCallback = mock(FutureCallback.class);
 423         doNothing().when(mockAbortCallback).onSuccess(null);
 424         cohort2.abort(mockAbortCallback);
 425         verify(mockAbortCallback).onSuccess(null);
 426
 427         coordinatedPreCommit(cohort4);
 428         coordinatedCommit(cohort1);
 429         coordinatedCommit(cohort3);
 430         coordinatedCommit(cohort4);
 431
 432         InOrder inOrder = inOrder(mockShard);
 433         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 434                 eq(false));
 435         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 436                 eq(false));
 437         inOrder.verify(mockShard).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 438                 eq(false));
 439
 440         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 441         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 442                 cohort1.getCandidate());
 443         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 444         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 445         shardDataTree.applyReplicatedPayload(cohort4.getIdentifier(), mockPayload);
 446
 447         final DataTreeSnapshot snapshot =
 448                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 449         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 450         assertTrue("Car node present", optional.isPresent());
 451         assertEquals("Car node", carNode, optional.get());
 452     }
 453
 454     @SuppressWarnings("unchecked")
 455     @Test
 456     public void testAbortWithFailedRebase() {
 457         immediatePayloadReplication(shardDataTree, mockShard);
 458
 459         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 460             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 461
 462         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 463             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 464
 465         NormalizedNode<?, ?> peopleNode = PeopleModel.create();
 466         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 467             snapshot.write(PeopleModel.BASE_PATH, peopleNode));
 468
 469         immediateCanCommit(cohort1);
 470         FutureCallback<Void> canCommitCallback2 = coordinatedCanCommit(cohort2);
 471
 472         coordinatedPreCommit(cohort1);
 473         verify(canCommitCallback2).onSuccess(null);
 474
 475         FutureCallback<Void> mockAbortCallback = mock(FutureCallback.class);
 476         doNothing().when(mockAbortCallback).onSuccess(null);
 477         cohort1.abort(mockAbortCallback);
 478         verify(mockAbortCallback).onSuccess(null);
 479
 480         FutureCallback<DataTreeCandidate> preCommitCallback2 = coordinatedPreCommit(cohort2);
 481         verify(preCommitCallback2).onFailure(any(Throwable.class));
 482
 483         immediateCanCommit(cohort3);
 484         immediatePreCommit(cohort3);
 485         immediateCommit(cohort3);
 486
 487         final DataTreeSnapshot snapshot =
 488                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 489         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(PeopleModel.BASE_PATH);
 490         assertTrue("People node present", optional.isPresent());
 491         assertEquals("People node", peopleNode, optional.get());
 492     }
 493
 494     @Test
 495     public void testUintCommitPayload() throws IOException {
 496         shardDataTree.applyRecoveryPayload(CommitTransactionPayload.create(nextTransactionId(),
 497             DataTreeCandidates.fromNormalizedNode(YangInstanceIdentifier.empty(), bigIntegerRoot()),
 498             PayloadVersion.SODIUM_SR1));
 499
 500         assertCarsUint64();
 501     }
 502
 503     @Test
 504     public void testUintSnapshot() throws IOException, DataValidationFailedException {
 505         shardDataTree.applyRecoverySnapshot(new ShardSnapshotState(new MetadataShardDataTreeSnapshot(bigIntegerRoot()),
 506             true));
 507
 508         assertCarsUint64();
 509     }
 510
 511     @Test
 512     public void testUintReplay() throws DataValidationFailedException, IOException {
 513         // Commit two writes and one merge, saving the data tree candidate for each.
 514         //        write(foo=1)
 515         //        write(foo=2)
 516         //        merge(foo=3)
 517         final DataTree dataTree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_OPERATIONAL,
 518             fullSchema);
 519         DataTreeModification mod = dataTree.takeSnapshot().newModification();
 520         mod.write(CarsModel.BASE_PATH, Builders.containerBuilder()
 521                 .withNodeIdentifier(new NodeIdentifier(CarsModel.BASE_QNAME))
 522                 .withChild(Builders.mapBuilder()
 523                     .withNodeIdentifier(new NodeIdentifier(CarsModel.CAR_QNAME))
 524                     .withChild(createCar("one", BigInteger.ONE))
 525                     .build())
 526                 .build());
 527         mod.ready();
 528         dataTree.validate(mod);
 529         final DataTreeCandidate first = dataTree.prepare(mod);
 530         dataTree.commit(first);
 531
 532         mod = dataTree.takeSnapshot().newModification();
 533         mod.write(CarsModel.newCarPath("two"), createCar("two", BigInteger.TWO));
 534         mod.ready();
 535         dataTree.validate(mod);
 536         final DataTreeCandidate second = dataTree.prepare(mod);
 537         dataTree.commit(second);
 538
 539         mod = dataTree.takeSnapshot().newModification();
 540         mod.merge(CarsModel.CAR_LIST_PATH, Builders.mapBuilder()
 541             .withNodeIdentifier(new NodeIdentifier(CarsModel.CAR_QNAME))
 542             .withChild(createCar("three", BigInteger.TEN))
 543             .build());
 544         mod.ready();
 545         dataTree.validate(mod);
 546         final DataTreeCandidate third = dataTree.prepare(mod);
 547         dataTree.commit(third);
 548
 549         // Apply first candidate as a snapshot
 550         shardDataTree.applyRecoverySnapshot(
 551             new ShardSnapshotState(new MetadataShardDataTreeSnapshot(first.getRootNode().getDataAfter().get()), true));
 552         // Apply the other two snapshots as transactions
 553         shardDataTree.applyRecoveryPayload(CommitTransactionPayload.create(nextTransactionId(), second,
 554             PayloadVersion.SODIUM_SR1));
 555         shardDataTree.applyRecoveryPayload(CommitTransactionPayload.create(nextTransactionId(), third,
 556             PayloadVersion.SODIUM_SR1));
 557
 558         // Verify uint translation
 559         final DataTreeSnapshot snapshot = shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 560         final NormalizedNode<?, ?> cars = snapshot.readNode(CarsModel.CAR_LIST_PATH).get();
 561
 562         assertEquals(Builders.mapBuilder()
 563             .withNodeIdentifier(new NodeIdentifier(CarsModel.CAR_QNAME))
 564             // Note: Uint64
 565             .withChild(createCar("one", Uint64.ONE))
 566             .withChild(createCar("two", Uint64.TWO))
 567             .withChild(createCar("three", Uint64.TEN))
 568             .build(), cars);
 569     }
 570
 571     private void assertCarsUint64() {
 572         final DataTreeSnapshot snapshot = shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 573         final NormalizedNode<?, ?> cars = snapshot.readNode(CarsModel.CAR_LIST_PATH).get();
 574
 575         assertEquals(Builders.mapBuilder()
 576             .withNodeIdentifier(new NodeIdentifier(CarsModel.CAR_QNAME))
 577             // Note: Uint64
 578             .withChild(createCar("foo", Uint64.ONE))
 579             .build(), cars);
 580     }
 581
 582     private static ContainerNode bigIntegerRoot() {
 583         return Builders.containerBuilder()
 584                 .withNodeIdentifier(new NodeIdentifier(SchemaContext.NAME))
 585                 .withChild(Builders.containerBuilder()
 586                     .withNodeIdentifier(new NodeIdentifier(CarsModel.CARS_QNAME))
 587                     .withChild(Builders.mapBuilder()
 588                         .withNodeIdentifier(new NodeIdentifier(CarsModel.CAR_QNAME))
 589                         // Note: BigInteger
 590                         .withChild(createCar("foo", BigInteger.ONE))
 591                         .build())
 592                     .build())
 593                 .build();
 594     }
 595
 596     private static MapEntryNode createCar(final String name, final Object value) {
 597         return Builders.mapEntryBuilder()
 598                 .withNodeIdentifier(NodeIdentifierWithPredicates.of(CarsModel.CAR_QNAME,CarsModel.CAR_NAME_QNAME, name))
 599                 .withChild(ImmutableNodes.leafNode(CarsModel.CAR_NAME_QNAME, name))
 600                 // Note: old BigInteger
 601                 .withChild(ImmutableNodes.leafNode(CarsModel.CAR_PRICE_QNAME, value))
 602                 .build();
 603     }
 604
 605     private ShardDataTreeCohort newShardDataTreeCohort(final DataTreeOperation operation) {
 606         final ReadWriteShardDataTreeTransaction transaction =
 607                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 608         final DataTreeModification snapshot = transaction.getSnapshot();
 609         operation.execute(snapshot);
 610         return shardDataTree.finishTransaction(transaction, Optional.empty());
 611     }
 612
 613     @SuppressWarnings({ "rawtypes", "unchecked" })
 614     private static void verifyOnDataTreeChanged(final DOMDataTreeChangeListener listener,
 615             final Consumer<DataTreeCandidate> callback) {
 616         ArgumentCaptor<Collection> changes = ArgumentCaptor.forClass(Collection.class);
 617         verify(listener, atLeastOnce()).onDataTreeChanged(changes.capture());
 618         for (Collection list : changes.getAllValues()) {
 619             for (Object dtc : list) {
 620                 callback.accept((DataTreeCandidate)dtc);
 621             }
 622         }
 623
 624         reset(listener);
 625     }
 626
 627     private static NormalizedNode<?, ?> getCars(final ShardDataTree shardDataTree) {
 628         final ReadOnlyShardDataTreeTransaction readOnlyShardDataTreeTransaction =
 629                 shardDataTree.newReadOnlyTransaction(nextTransactionId());
 630         final DataTreeSnapshot snapshot1 = readOnlyShardDataTreeTransaction.getSnapshot();
 631
 632         final Optional<NormalizedNode<?, ?>> optional = snapshot1.readNode(CarsModel.BASE_PATH);
 633
 634         assertTrue(optional.isPresent());
 635
 636         return optional.get();
 637     }
 638
 639     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree) {
 640         return addCar(shardDataTree, "altima");
 641     }
 642
 643     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree, final String name) {
 644         return doTransaction(shardDataTree, snapshot -> {
 645             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 646             snapshot.merge(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 647             snapshot.write(CarsModel.newCarPath(name), CarsModel.newCarEntry(name, Uint64.valueOf(100)));
 648         });
 649     }
 650
 651     private static DataTreeCandidate removeCar(final ShardDataTree shardDataTree) {
 652         return doTransaction(shardDataTree, snapshot -> snapshot.delete(CarsModel.newCarPath("altima")));
 653     }
 654
 655     @FunctionalInterface
 656     private interface DataTreeOperation {
 657         void execute(DataTreeModification snapshot);
 658     }
 659
 660     private static DataTreeCandidate doTransaction(final ShardDataTree shardDataTree,
 661             final DataTreeOperation operation) {
 662         final ReadWriteShardDataTreeTransaction transaction =
 663                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 664         final DataTreeModification snapshot = transaction.getSnapshot();
 665         operation.execute(snapshot);
 666         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction, Optional.empty());
 667
 668         immediateCanCommit(cohort);
 669         immediatePreCommit(cohort);
 670         final DataTreeCandidate candidate = cohort.getCandidate();
 671         immediateCommit(cohort);
 672
 673         return candidate;
 674     }
 675
 676     private static DataTreeCandidate applyCandidates(final ShardDataTree shardDataTree,
 677             final List<DataTreeCandidate> candidates) {
 678         final ReadWriteShardDataTreeTransaction transaction =
 679                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 680         final DataTreeModification snapshot = transaction.getSnapshot();
 681         for (final DataTreeCandidate candidateTip : candidates) {
 682             DataTreeCandidates.applyToModification(snapshot, candidateTip);
 683         }
 684         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction, Optional.empty());
 685
 686         immediateCanCommit(cohort);
 687         immediatePreCommit(cohort);
 688         final DataTreeCandidate candidate = cohort.getCandidate();
 689         immediateCommit(cohort);
 690
 691         return candidate;
 692     }
 693 }