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
   9 package org.opendaylight.controller.cluster.datastore;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertNotNull;
  13 import static org.junit.Assert.assertTrue;
  14 import static org.junit.Assert.fail;
  15 import static org.mockito.ArgumentMatchers.any;
  16 import static org.mockito.ArgumentMatchers.anyBoolean;
  17 import static org.mockito.ArgumentMatchers.eq;
  18 import static org.mockito.Mockito.atLeastOnce;
  19 import static org.mockito.Mockito.doNothing;
  20 import static org.mockito.Mockito.doReturn;
  21 import static org.mockito.Mockito.inOrder;
  22 import static org.mockito.Mockito.mock;
  23 import static org.mockito.Mockito.never;
  24 import static org.mockito.Mockito.reset;
  25 import static org.mockito.Mockito.verify;
  26 import static org.mockito.Mockito.verifyNoMoreInteractions;
  27 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedCanCommit;
  28 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedCommit;
  29 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedPreCommit;
  30 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediate3PhaseCommit;
  31 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediateCanCommit;
  32 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediateCommit;
  33 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediatePayloadReplication;
  34 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediatePreCommit;
  35
  36 import com.google.common.base.Ticker;
  37 import com.google.common.collect.Maps;
  38 import com.google.common.primitives.UnsignedLong;
  39 import com.google.common.util.concurrent.FutureCallback;
  40 import java.math.BigInteger;
  41 import java.util.ArrayList;
  42 import java.util.Collection;
  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.md.cluster.datastore.model.CarsModel;
  55 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
  56 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  57 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
  58 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  59 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  60 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  61 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  62 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
  63 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  64 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
  65 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
  66 import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
  67 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  68
  69 public class ShardDataTreeTest extends AbstractTest {
  70     private static final DatastoreContext DATASTORE_CONTEXT = DatastoreContext.newBuilder().build();
  71
  72     private final Shard mockShard = Mockito.mock(Shard.class);
  73     private ShardDataTree shardDataTree;
  74     private SchemaContext fullSchema;
  75
  76     @Before
  77     public void setUp() {
  78         doReturn(Ticker.systemTicker()).when(mockShard).ticker();
  79         doReturn(mock(ShardStats.class)).when(mockShard).getShardMBean();
  80         doReturn(DATASTORE_CONTEXT).when(mockShard).getDatastoreContext();
  81
  82         fullSchema = SchemaContextHelper.full();
  83
  84         shardDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
  85     }
  86
  87     @Test
  88     public void testWrite() {
  89         modify(false, true, true);
  90     }
  91
  92     @Test
  93     public void testMerge() {
  94         modify(true, true, true);
  95     }
  96
  97     private void modify(final boolean merge, final boolean expectedCarsPresent, final boolean expectedPeoplePresent) {
  98         immediatePayloadReplication(shardDataTree, mockShard);
  99
 100         assertEquals(fullSchema, shardDataTree.getSchemaContext());
 101
 102         final ReadWriteShardDataTreeTransaction transaction =
 103                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 104
 105         final DataTreeModification snapshot = transaction.getSnapshot();
 106
 107         assertNotNull(snapshot);
 108
 109         if (merge) {
 110             snapshot.merge(CarsModel.BASE_PATH, CarsModel.create());
 111             snapshot.merge(PeopleModel.BASE_PATH, PeopleModel.create());
 112         } else {
 113             snapshot.write(CarsModel.BASE_PATH, CarsModel.create());
 114             snapshot.write(PeopleModel.BASE_PATH, PeopleModel.create());
 115         }
 116
 117         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction, Optional.empty());
 118
 119         immediateCanCommit(cohort);
 120         immediatePreCommit(cohort);
 121         immediateCommit(cohort);
 122
 123         final ReadOnlyShardDataTreeTransaction readOnlyShardDataTreeTransaction =
 124                 shardDataTree.newReadOnlyTransaction(nextTransactionId());
 125
 126         final DataTreeSnapshot snapshot1 = readOnlyShardDataTreeTransaction.getSnapshot();
 127
 128         final Optional<NormalizedNode<?, ?>> optional = snapshot1.readNode(CarsModel.BASE_PATH);
 129
 130         assertEquals(expectedCarsPresent, optional.isPresent());
 131
 132         final Optional<NormalizedNode<?, ?>> optional1 = snapshot1.readNode(PeopleModel.BASE_PATH);
 133
 134         assertEquals(expectedPeoplePresent, optional1.isPresent());
 135     }
 136
 137     @Test
 138     public void bug4359AddRemoveCarOnce() {
 139         immediatePayloadReplication(shardDataTree, mockShard);
 140
 141         final List<DataTreeCandidate> candidates = new ArrayList<>();
 142         candidates.add(addCar(shardDataTree));
 143         candidates.add(removeCar(shardDataTree));
 144
 145         final NormalizedNode<?, ?> expected = getCars(shardDataTree);
 146
 147         applyCandidates(shardDataTree, candidates);
 148
 149         final NormalizedNode<?, ?> actual = getCars(shardDataTree);
 150
 151         assertEquals(expected, actual);
 152     }
 153
 154     @Test
 155     public void bug4359AddRemoveCarTwice() {
 156         immediatePayloadReplication(shardDataTree, mockShard);
 157
 158         final List<DataTreeCandidate> candidates = new ArrayList<>();
 159         candidates.add(addCar(shardDataTree));
 160         candidates.add(removeCar(shardDataTree));
 161         candidates.add(addCar(shardDataTree));
 162         candidates.add(removeCar(shardDataTree));
 163
 164         final NormalizedNode<?, ?> expected = getCars(shardDataTree);
 165
 166         applyCandidates(shardDataTree, candidates);
 167
 168         final NormalizedNode<?, ?> actual = getCars(shardDataTree);
 169
 170         assertEquals(expected, actual);
 171     }
 172
 173     @Test
 174     public void testListenerNotifiedOnApplySnapshot() throws Exception {
 175         immediatePayloadReplication(shardDataTree, mockShard);
 176
 177         DOMDataTreeChangeListener listener = mock(DOMDataTreeChangeListener.class);
 178         shardDataTree.registerTreeChangeListener(CarsModel.CAR_LIST_PATH.node(CarsModel.CAR_QNAME), listener,
 179             com.google.common.base.Optional.absent(), noop -> { });
 180
 181         addCar(shardDataTree, "optima");
 182
 183         verifyOnDataTreeChanged(listener, dtc -> {
 184             assertEquals("getModificationType", ModificationType.WRITE, dtc.getRootNode().getModificationType());
 185             assertEquals("getRootPath", CarsModel.newCarPath("optima"), dtc.getRootPath());
 186         });
 187
 188         addCar(shardDataTree, "sportage");
 189
 190         verifyOnDataTreeChanged(listener, dtc -> {
 191             assertEquals("getModificationType", ModificationType.WRITE, dtc.getRootNode().getModificationType());
 192             assertEquals("getRootPath", CarsModel.newCarPath("sportage"), dtc.getRootPath());
 193         });
 194
 195         ShardDataTree newDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
 196         immediatePayloadReplication(newDataTree, mockShard);
 197         addCar(newDataTree, "optima");
 198         addCar(newDataTree, "murano");
 199
 200         shardDataTree.applySnapshot(newDataTree.takeStateSnapshot());
 201
 202         Map<YangInstanceIdentifier, ModificationType> expChanges = Maps.newHashMap();
 203         expChanges.put(CarsModel.newCarPath("optima"), ModificationType.WRITE);
 204         expChanges.put(CarsModel.newCarPath("murano"), ModificationType.WRITE);
 205         expChanges.put(CarsModel.newCarPath("sportage"), ModificationType.DELETE);
 206         verifyOnDataTreeChanged(listener, dtc -> {
 207             ModificationType expType = expChanges.remove(dtc.getRootPath());
 208             assertNotNull("Got unexpected change for " + dtc.getRootPath(), expType);
 209             assertEquals("getModificationType", expType, dtc.getRootNode().getModificationType());
 210         });
 211
 212         if (!expChanges.isEmpty()) {
 213             fail("Missing change notifications: " + expChanges);
 214         }
 215     }
 216
 217     @Test
 218     public void testPipelinedTransactionsWithCoordinatedCommits() throws Exception {
 219         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 220             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 221
 222         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 223             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 224
 225         NormalizedNode<?, ?> peopleNode = PeopleModel.create();
 226         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 227             snapshot.write(PeopleModel.BASE_PATH, peopleNode));
 228
 229         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 230         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
 231         final ShardDataTreeCohort cohort4 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 232
 233         immediateCanCommit(cohort1);
 234         final FutureCallback<Void> canCommitCallback2 = coordinatedCanCommit(cohort2);
 235         final FutureCallback<Void> canCommitCallback3 = coordinatedCanCommit(cohort3);
 236         final FutureCallback<Void> canCommitCallback4 = coordinatedCanCommit(cohort4);
 237
 238         final FutureCallback<DataTreeCandidate> preCommitCallback1 = coordinatedPreCommit(cohort1);
 239         verify(preCommitCallback1).onSuccess(cohort1.getCandidate());
 240         verify(canCommitCallback2).onSuccess(null);
 241
 242         final FutureCallback<DataTreeCandidate> preCommitCallback2 = coordinatedPreCommit(cohort2);
 243         verify(preCommitCallback2).onSuccess(cohort2.getCandidate());
 244         verify(canCommitCallback3).onSuccess(null);
 245
 246         final FutureCallback<DataTreeCandidate> preCommitCallback3 = coordinatedPreCommit(cohort3);
 247         verify(preCommitCallback3).onSuccess(cohort3.getCandidate());
 248         verify(canCommitCallback4).onSuccess(null);
 249
 250         final FutureCallback<DataTreeCandidate> preCommitCallback4 = coordinatedPreCommit(cohort4);
 251         verify(preCommitCallback4).onSuccess(cohort4.getCandidate());
 252
 253         final FutureCallback<UnsignedLong> commitCallback2 = coordinatedCommit(cohort2);
 254         verify(mockShard, never()).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 255                 anyBoolean());
 256         verifyNoMoreInteractions(commitCallback2);
 257
 258         final FutureCallback<UnsignedLong> commitCallback4 = coordinatedCommit(cohort4);
 259         verify(mockShard, never()).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 260                 anyBoolean());
 261         verifyNoMoreInteractions(commitCallback4);
 262
 263         final FutureCallback<UnsignedLong> commitCallback1 = coordinatedCommit(cohort1);
 264         InOrder inOrder = inOrder(mockShard);
 265         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 266                 eq(true));
 267         inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
 268                 eq(false));
 269         verifyNoMoreInteractions(commitCallback1);
 270         verifyNoMoreInteractions(commitCallback2);
 271
 272         final FutureCallback<UnsignedLong> commitCallback3 = coordinatedCommit(cohort3);
 273         inOrder = inOrder(mockShard);
 274         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 275                 eq(true));
 276         inOrder.verify(mockShard).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 277                 eq(false));
 278         verifyNoMoreInteractions(commitCallback3);
 279         verifyNoMoreInteractions(commitCallback4);
 280
 281         final ShardDataTreeCohort cohort5 = newShardDataTreeCohort(snapshot ->
 282             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 283         final FutureCallback<Void> canCommitCallback5 = coordinatedCanCommit(cohort5);
 284
 285         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 286         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 287                 cohort1.getCandidate());
 288         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 289         shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
 290         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 291         shardDataTree.applyReplicatedPayload(cohort4.getIdentifier(), mockPayload);
 292
 293         inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3, commitCallback4);
 294         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 295         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 296         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 297         inOrder.verify(commitCallback4).onSuccess(any(UnsignedLong.class));
 298
 299         verify(canCommitCallback5).onSuccess(null);
 300
 301         final DataTreeSnapshot snapshot =
 302                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 303         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 304         assertTrue("Car node present", optional.isPresent());
 305         assertEquals("Car node", carNode, optional.get());
 306
 307         optional = snapshot.readNode(PeopleModel.BASE_PATH);
 308         assertTrue("People node present", optional.isPresent());
 309         assertEquals("People node", peopleNode, optional.get());
 310     }
 311
 312     @Test
 313     public void testPipelinedTransactionsWithImmediateCommits() throws Exception {
 314         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 315             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 316
 317         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 318             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 319
 320         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 321         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
 322         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 323
 324         final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
 325         final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
 326         final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
 327
 328         InOrder inOrder = inOrder(mockShard);
 329         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 330                 eq(true));
 331         inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
 332                 eq(true));
 333         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 334                 eq(false));
 335
 336         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 337         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 338                 cohort1.getCandidate());
 339         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 340         shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
 341         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 342
 343         inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
 344         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 345         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 346         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 347
 348         final DataTreeSnapshot snapshot =
 349                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 350         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 351         assertTrue("Car node present", optional.isPresent());
 352         assertEquals("Car node", carNode, optional.get());
 353     }
 354
 355     @Test
 356     public void testPipelinedTransactionsWithImmediateReplication() {
 357         immediatePayloadReplication(shardDataTree, mockShard);
 358
 359         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 360             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 361
 362         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 363             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 364
 365         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 366         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
 367         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 368
 369         final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
 370         final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
 371         final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
 372
 373         InOrder inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
 374         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 375         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 376         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 377
 378         final DataTreeSnapshot snapshot = shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 379         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(CarsModel.BASE_PATH);
 380         assertTrue("Car node present", optional.isPresent());
 381     }
 382
 383     @SuppressWarnings("unchecked")
 384     @Test
 385     public void testAbortWithPendingCommits() throws Exception {
 386         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 387             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 388
 389         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 390             snapshot.write(PeopleModel.BASE_PATH, PeopleModel.create()));
 391
 392         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 393             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 394
 395         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 396         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
 397         final ShardDataTreeCohort cohort4 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 398
 399         coordinatedCanCommit(cohort2);
 400         immediateCanCommit(cohort1);
 401         coordinatedCanCommit(cohort3);
 402         coordinatedCanCommit(cohort4);
 403
 404         coordinatedPreCommit(cohort1);
 405         coordinatedPreCommit(cohort2);
 406         coordinatedPreCommit(cohort3);
 407
 408         FutureCallback<Void> mockAbortCallback = mock(FutureCallback.class);
 409         doNothing().when(mockAbortCallback).onSuccess(null);
 410         cohort2.abort(mockAbortCallback);
 411         verify(mockAbortCallback).onSuccess(null);
 412
 413         coordinatedPreCommit(cohort4);
 414         coordinatedCommit(cohort1);
 415         coordinatedCommit(cohort3);
 416         coordinatedCommit(cohort4);
 417
 418         InOrder inOrder = inOrder(mockShard);
 419         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 420                 eq(false));
 421         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 422                 eq(false));
 423         inOrder.verify(mockShard).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 424                 eq(false));
 425
 426         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 427         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 428                 cohort1.getCandidate());
 429         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 430         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 431         shardDataTree.applyReplicatedPayload(cohort4.getIdentifier(), mockPayload);
 432
 433         final DataTreeSnapshot snapshot =
 434                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 435         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 436         assertTrue("Car node present", optional.isPresent());
 437         assertEquals("Car node", carNode, optional.get());
 438     }
 439
 440     @SuppressWarnings("unchecked")
 441     @Test
 442     public void testAbortWithFailedRebase() {
 443         immediatePayloadReplication(shardDataTree, mockShard);
 444
 445         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 446             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 447
 448         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 449             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 450
 451         NormalizedNode<?, ?> peopleNode = PeopleModel.create();
 452         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 453             snapshot.write(PeopleModel.BASE_PATH, peopleNode));
 454
 455         immediateCanCommit(cohort1);
 456         FutureCallback<Void> canCommitCallback2 = coordinatedCanCommit(cohort2);
 457
 458         coordinatedPreCommit(cohort1);
 459         verify(canCommitCallback2).onSuccess(null);
 460
 461         FutureCallback<Void> mockAbortCallback = mock(FutureCallback.class);
 462         doNothing().when(mockAbortCallback).onSuccess(null);
 463         cohort1.abort(mockAbortCallback);
 464         verify(mockAbortCallback).onSuccess(null);
 465
 466         FutureCallback<DataTreeCandidate> preCommitCallback2 = coordinatedPreCommit(cohort2);
 467         verify(preCommitCallback2).onFailure(any(Throwable.class));
 468
 469         immediateCanCommit(cohort3);
 470         immediatePreCommit(cohort3);
 471         immediateCommit(cohort3);
 472
 473         final DataTreeSnapshot snapshot =
 474                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 475         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(PeopleModel.BASE_PATH);
 476         assertTrue("People node present", optional.isPresent());
 477         assertEquals("People node", peopleNode, optional.get());
 478     }
 479
 480     private ShardDataTreeCohort newShardDataTreeCohort(final DataTreeOperation operation) {
 481         final ReadWriteShardDataTreeTransaction transaction =
 482                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 483         final DataTreeModification snapshot = transaction.getSnapshot();
 484         operation.execute(snapshot);
 485         return shardDataTree.finishTransaction(transaction, Optional.empty());
 486     }
 487
 488     @SuppressWarnings({ "rawtypes", "unchecked" })
 489     private static void verifyOnDataTreeChanged(final DOMDataTreeChangeListener listener,
 490             final Consumer<DataTreeCandidate> callback) {
 491         ArgumentCaptor<Collection> changes = ArgumentCaptor.forClass(Collection.class);
 492         verify(listener, atLeastOnce()).onDataTreeChanged(changes.capture());
 493         for (Collection list : changes.getAllValues()) {
 494             for (Object dtc : list) {
 495                 callback.accept((DataTreeCandidate)dtc);
 496             }
 497         }
 498
 499         reset(listener);
 500     }
 501
 502     private static NormalizedNode<?, ?> getCars(final ShardDataTree shardDataTree) {
 503         final ReadOnlyShardDataTreeTransaction readOnlyShardDataTreeTransaction =
 504                 shardDataTree.newReadOnlyTransaction(nextTransactionId());
 505         final DataTreeSnapshot snapshot1 = readOnlyShardDataTreeTransaction.getSnapshot();
 506
 507         final Optional<NormalizedNode<?, ?>> optional = snapshot1.readNode(CarsModel.BASE_PATH);
 508
 509         assertTrue(optional.isPresent());
 510
 511         return optional.get();
 512     }
 513
 514     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree) {
 515         return addCar(shardDataTree, "altima");
 516     }
 517
 518     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree, final String name) {
 519         return doTransaction(shardDataTree, snapshot -> {
 520             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 521             snapshot.merge(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 522             snapshot.write(CarsModel.newCarPath(name), CarsModel.newCarEntry(name, new BigInteger("100")));
 523         });
 524     }
 525
 526     private static DataTreeCandidate removeCar(final ShardDataTree shardDataTree) {
 527         return doTransaction(shardDataTree, snapshot -> snapshot.delete(CarsModel.newCarPath("altima")));
 528     }
 529
 530     @FunctionalInterface
 531     private interface DataTreeOperation {
 532         void execute(DataTreeModification snapshot);
 533     }
 534
 535     private static DataTreeCandidate doTransaction(final ShardDataTree shardDataTree,
 536             final DataTreeOperation operation) {
 537         final ReadWriteShardDataTreeTransaction transaction =
 538                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 539         final DataTreeModification snapshot = transaction.getSnapshot();
 540         operation.execute(snapshot);
 541         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction, Optional.empty());
 542
 543         immediateCanCommit(cohort);
 544         immediatePreCommit(cohort);
 545         final DataTreeCandidate candidate = cohort.getCandidate();
 546         immediateCommit(cohort);
 547
 548         return candidate;
 549     }
 550
 551     private static DataTreeCandidate applyCandidates(final ShardDataTree shardDataTree,
 552             final List<DataTreeCandidate> candidates) {
 553         final ReadWriteShardDataTreeTransaction transaction =
 554                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 555         final DataTreeModification snapshot = transaction.getSnapshot();
 556         for (final DataTreeCandidate candidateTip : candidates) {
 557             DataTreeCandidates.applyToModification(snapshot, candidateTip);
 558         }
 559         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction, Optional.empty());
 560
 561         immediateCanCommit(cohort);
 562         immediatePreCommit(cohort);
 563         final DataTreeCandidate candidate = cohort.getCandidate();
 564         immediateCommit(cohort);
 565
 566         return candidate;
 567     }
 568 }