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.fail;
  14 import static org.mockito.Matchers.any;
  15 import static org.mockito.Matchers.anyBoolean;
  16 import static org.mockito.Matchers.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.collect.Maps;
  37 import com.google.common.primitives.UnsignedLong;
  38 import com.google.common.util.concurrent.FutureCallback;
  39 import java.math.BigInteger;
  40 import java.util.ArrayList;
  41 import java.util.Collection;
  42 import java.util.List;
  43 import java.util.Map;
  44 import java.util.Optional;
  45 import java.util.concurrent.ExecutionException;
  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
  71     private final Shard mockShard = Mockito.mock(Shard.class);
  72     private ShardDataTree shardDataTree;
  73     private SchemaContext fullSchema;
  74
  75     @Before
  76     public void setUp() {
  77         doReturn(Ticker.systemTicker()).when(mockShard).ticker();
  78         doReturn(Mockito.mock(ShardStats.class)).when(mockShard).getShardMBean();
  79
  80         fullSchema = SchemaContextHelper.full();
  81
  82         shardDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
  83     }
  84
  85     @Test
  86     public void testWrite() throws ExecutionException, InterruptedException {
  87         modify(false, true, true);
  88     }
  89
  90     @Test
  91     public void testMerge() throws ExecutionException, InterruptedException {
  92         modify(true, true, true);
  93     }
  94
  95     private void modify(final boolean merge, final boolean expectedCarsPresent, final boolean expectedPeoplePresent)
  96             throws ExecutionException, InterruptedException {
  97         immediatePayloadReplication(shardDataTree, mockShard);
  98
  99         assertEquals(fullSchema, shardDataTree.getSchemaContext());
 100
 101         final ReadWriteShardDataTreeTransaction transaction =
 102                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 103
 104         final DataTreeModification snapshot = transaction.getSnapshot();
 105
 106         assertNotNull(snapshot);
 107
 108         if (merge) {
 109             snapshot.merge(CarsModel.BASE_PATH, CarsModel.create());
 110             snapshot.merge(PeopleModel.BASE_PATH, PeopleModel.create());
 111         } else {
 112             snapshot.write(CarsModel.BASE_PATH, CarsModel.create());
 113             snapshot.write(PeopleModel.BASE_PATH, PeopleModel.create());
 114         }
 115
 116         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction);
 117
 118         immediateCanCommit(cohort);
 119         immediatePreCommit(cohort);
 120         immediateCommit(cohort);
 121
 122         final ReadOnlyShardDataTreeTransaction readOnlyShardDataTreeTransaction =
 123                 shardDataTree.newReadOnlyTransaction(nextTransactionId());
 124
 125         final DataTreeSnapshot snapshot1 = readOnlyShardDataTreeTransaction.getSnapshot();
 126
 127         final Optional<NormalizedNode<?, ?>> optional = snapshot1.readNode(CarsModel.BASE_PATH);
 128
 129         assertEquals(expectedCarsPresent, optional.isPresent());
 130
 131         final Optional<NormalizedNode<?, ?>> optional1 = snapshot1.readNode(PeopleModel.BASE_PATH);
 132
 133         assertEquals(expectedPeoplePresent, optional1.isPresent());
 134     }
 135
 136     @Test
 137     public void bug4359AddRemoveCarOnce() throws ExecutionException, InterruptedException {
 138         immediatePayloadReplication(shardDataTree, mockShard);
 139
 140         final List<DataTreeCandidate> candidates = new ArrayList<>();
 141         candidates.add(addCar(shardDataTree));
 142         candidates.add(removeCar(shardDataTree));
 143
 144         final NormalizedNode<?, ?> expected = getCars(shardDataTree);
 145
 146         applyCandidates(shardDataTree, candidates);
 147
 148         final NormalizedNode<?, ?> actual = getCars(shardDataTree);
 149
 150         assertEquals(expected, actual);
 151     }
 152
 153     @Test
 154     public void bug4359AddRemoveCarTwice() throws ExecutionException, InterruptedException {
 155         immediatePayloadReplication(shardDataTree, mockShard);
 156
 157         final List<DataTreeCandidate> candidates = new ArrayList<>();
 158         candidates.add(addCar(shardDataTree));
 159         candidates.add(removeCar(shardDataTree));
 160         candidates.add(addCar(shardDataTree));
 161         candidates.add(removeCar(shardDataTree));
 162
 163         final NormalizedNode<?, ?> expected = getCars(shardDataTree);
 164
 165         applyCandidates(shardDataTree, candidates);
 166
 167         final NormalizedNode<?, ?> actual = getCars(shardDataTree);
 168
 169         assertEquals(expected, actual);
 170     }
 171
 172     @Test
 173     public void testListenerNotifiedOnApplySnapshot() throws Exception {
 174         immediatePayloadReplication(shardDataTree, mockShard);
 175
 176         DOMDataTreeChangeListener listener = mock(DOMDataTreeChangeListener.class);
 177         shardDataTree.registerTreeChangeListener(CarsModel.CAR_LIST_PATH.node(CarsModel.CAR_QNAME), listener,
 178             com.google.common.base.Optional.absent(), noop -> { });
 179
 180         addCar(shardDataTree, "optima");
 181
 182         verifyOnDataTreeChanged(listener, dtc -> {
 183             assertEquals("getModificationType", ModificationType.WRITE, dtc.getRootNode().getModificationType());
 184             assertEquals("getRootPath", CarsModel.newCarPath("optima"), dtc.getRootPath());
 185         });
 186
 187         addCar(shardDataTree, "sportage");
 188
 189         verifyOnDataTreeChanged(listener, dtc -> {
 190             assertEquals("getModificationType", ModificationType.WRITE, dtc.getRootNode().getModificationType());
 191             assertEquals("getRootPath", CarsModel.newCarPath("sportage"), dtc.getRootPath());
 192         });
 193
 194         ShardDataTree newDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
 195         immediatePayloadReplication(newDataTree, mockShard);
 196         addCar(newDataTree, "optima");
 197         addCar(newDataTree, "murano");
 198
 199         shardDataTree.applySnapshot(newDataTree.takeStateSnapshot());
 200
 201         Map<YangInstanceIdentifier, ModificationType> expChanges = Maps.newHashMap();
 202         expChanges.put(CarsModel.newCarPath("optima"), ModificationType.WRITE);
 203         expChanges.put(CarsModel.newCarPath("murano"), ModificationType.WRITE);
 204         expChanges.put(CarsModel.newCarPath("sportage"), ModificationType.DELETE);
 205         verifyOnDataTreeChanged(listener, dtc -> {
 206             ModificationType expType = expChanges.remove(dtc.getRootPath());
 207             assertNotNull("Got unexpected change for " + dtc.getRootPath(), expType);
 208             assertEquals("getModificationType", expType, dtc.getRootNode().getModificationType());
 209         });
 210
 211         if (!expChanges.isEmpty()) {
 212             fail("Missing change notifications: " + expChanges);
 213         }
 214     }
 215
 216     @Test
 217     public void testPipelinedTransactionsWithCoordinatedCommits() throws Exception {
 218         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 219             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 220
 221         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 222             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 223
 224         NormalizedNode<?, ?> peopleNode = PeopleModel.create();
 225         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 226             snapshot.write(PeopleModel.BASE_PATH, peopleNode));
 227
 228         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 229         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
 230         final ShardDataTreeCohort cohort4 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 231
 232         immediateCanCommit(cohort1);
 233         final FutureCallback<Void> canCommitCallback2 = coordinatedCanCommit(cohort2);
 234         final FutureCallback<Void> canCommitCallback3 = coordinatedCanCommit(cohort3);
 235         final FutureCallback<Void> canCommitCallback4 = coordinatedCanCommit(cohort4);
 236
 237         final FutureCallback<DataTreeCandidate> preCommitCallback1 = coordinatedPreCommit(cohort1);
 238         verify(preCommitCallback1).onSuccess(cohort1.getCandidate());
 239         verify(canCommitCallback2).onSuccess(null);
 240
 241         final FutureCallback<DataTreeCandidate> preCommitCallback2 = coordinatedPreCommit(cohort2);
 242         verify(preCommitCallback2).onSuccess(cohort2.getCandidate());
 243         verify(canCommitCallback3).onSuccess(null);
 244
 245         final FutureCallback<DataTreeCandidate> preCommitCallback3 = coordinatedPreCommit(cohort3);
 246         verify(preCommitCallback3).onSuccess(cohort3.getCandidate());
 247         verify(canCommitCallback4).onSuccess(null);
 248
 249         final FutureCallback<DataTreeCandidate> preCommitCallback4 = coordinatedPreCommit(cohort4);
 250         verify(preCommitCallback4).onSuccess(cohort4.getCandidate());
 251
 252         final FutureCallback<UnsignedLong> commitCallback2 = coordinatedCommit(cohort2);
 253         verify(mockShard, never()).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 254                 anyBoolean());
 255         verifyNoMoreInteractions(commitCallback2);
 256
 257         final FutureCallback<UnsignedLong> commitCallback4 = coordinatedCommit(cohort4);
 258         verify(mockShard, never()).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 259                 anyBoolean());
 260         verifyNoMoreInteractions(commitCallback4);
 261
 262         final FutureCallback<UnsignedLong> commitCallback1 = coordinatedCommit(cohort1);
 263         InOrder inOrder = inOrder(mockShard);
 264         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 265                 eq(true));
 266         inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
 267                 eq(false));
 268         verifyNoMoreInteractions(commitCallback1);
 269         verifyNoMoreInteractions(commitCallback2);
 270
 271         final FutureCallback<UnsignedLong> commitCallback3 = coordinatedCommit(cohort3);
 272         inOrder = inOrder(mockShard);
 273         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 274                 eq(true));
 275         inOrder.verify(mockShard).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 276                 eq(false));
 277         verifyNoMoreInteractions(commitCallback3);
 278         verifyNoMoreInteractions(commitCallback4);
 279
 280         final ShardDataTreeCohort cohort5 = newShardDataTreeCohort(snapshot ->
 281             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 282         final FutureCallback<Void> canCommitCallback5 = coordinatedCanCommit(cohort5);
 283
 284         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 285         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 286                 cohort1.getCandidate());
 287         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 288         shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
 289         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 290         shardDataTree.applyReplicatedPayload(cohort4.getIdentifier(), mockPayload);
 291
 292         inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3, commitCallback4);
 293         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 294         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 295         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 296         inOrder.verify(commitCallback4).onSuccess(any(UnsignedLong.class));
 297
 298         verify(canCommitCallback5).onSuccess(null);
 299
 300         final DataTreeSnapshot snapshot =
 301                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 302         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 303         assertEquals("Car node present", true, optional.isPresent());
 304         assertEquals("Car node", carNode, optional.get());
 305
 306         optional = snapshot.readNode(PeopleModel.BASE_PATH);
 307         assertEquals("People node present", true, optional.isPresent());
 308         assertEquals("People node", peopleNode, optional.get());
 309     }
 310
 311     @Test
 312     public void testPipelinedTransactionsWithImmediateCommits() throws Exception {
 313         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 314             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 315
 316         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 317             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 318
 319         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 320         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
 321         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 322
 323         final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
 324         final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
 325         final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
 326
 327         InOrder inOrder = inOrder(mockShard);
 328         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 329                 eq(true));
 330         inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
 331                 eq(true));
 332         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 333                 eq(false));
 334
 335         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 336         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 337                 cohort1.getCandidate());
 338         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 339         shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
 340         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 341
 342         inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
 343         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 344         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 345         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 346
 347         final DataTreeSnapshot snapshot =
 348                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 349         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 350         assertEquals("Car node present", true, optional.isPresent());
 351         assertEquals("Car node", carNode, optional.get());
 352     }
 353
 354     @Test
 355     public void testPipelinedTransactionsWithImmediateReplication() throws Exception {
 356         immediatePayloadReplication(shardDataTree, mockShard);
 357
 358         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 359             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 360
 361         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 362             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 363
 364         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 365         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
 366         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 367
 368         final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
 369         final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
 370         final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
 371
 372         InOrder inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
 373         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
 374         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
 375         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
 376
 377         final DataTreeSnapshot snapshot =
 378                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 379         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(CarsModel.BASE_PATH);
 380         assertEquals("Car node present", true, 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         assertEquals("Car node present", true, optional.isPresent());
 437         assertEquals("Car node", carNode, optional.get());
 438     }
 439
 440     @SuppressWarnings("unchecked")
 441     @Test
 442     public void testAbortWithFailedRebase() throws Exception {
 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         assertEquals("People node present", true, 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);
 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         assertEquals(true, optional.isPresent());
 510
 511         return optional.get();
 512     }
 513
 514     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree)
 515             throws ExecutionException, InterruptedException {
 516         return addCar(shardDataTree, "altima");
 517     }
 518
 519     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree, final String name)
 520             throws ExecutionException, InterruptedException {
 521         return doTransaction(shardDataTree, snapshot -> {
 522             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 523             snapshot.merge(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 524             snapshot.write(CarsModel.newCarPath(name), CarsModel.newCarEntry(name, new BigInteger("100")));
 525         });
 526     }
 527
 528     private static DataTreeCandidate removeCar(final ShardDataTree shardDataTree)
 529             throws ExecutionException, InterruptedException {
 530         return doTransaction(shardDataTree, snapshot -> snapshot.delete(CarsModel.newCarPath("altima")));
 531     }
 532
 533     @FunctionalInterface
 534     private interface DataTreeOperation {
 535         void execute(DataTreeModification snapshot);
 536     }
 537
 538     private static DataTreeCandidate doTransaction(final ShardDataTree shardDataTree,
 539             final DataTreeOperation operation) throws ExecutionException, InterruptedException {
 540         final ReadWriteShardDataTreeTransaction transaction =
 541                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 542         final DataTreeModification snapshot = transaction.getSnapshot();
 543         operation.execute(snapshot);
 544         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction);
 545
 546         immediateCanCommit(cohort);
 547         immediatePreCommit(cohort);
 548         final DataTreeCandidate candidate = cohort.getCandidate();
 549         immediateCommit(cohort);
 550
 551         return candidate;
 552     }
 553
 554     private static DataTreeCandidate applyCandidates(final ShardDataTree shardDataTree,
 555             final List<DataTreeCandidate> candidates) throws ExecutionException, InterruptedException {
 556         final ReadWriteShardDataTreeTransaction transaction =
 557                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 558         final DataTreeModification snapshot = transaction.getSnapshot();
 559         for (final DataTreeCandidate candidateTip : candidates) {
 560             DataTreeCandidates.applyToModification(snapshot, candidateTip);
 561         }
 562         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction);
 563
 564         immediateCanCommit(cohort);
 565         immediatePreCommit(cohort);
 566         final DataTreeCandidate candidate = cohort.getCandidate();
 567         immediateCommit(cohort);
 568
 569         return candidate;
 570     }
 571 }