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.util.ArrayList;
  39 import java.util.Collection;
  40 import java.util.HashMap;
  41 import java.util.List;
  42 import java.util.Map;
  43 import java.util.Optional;
  44 import java.util.function.Consumer;
  45 import org.junit.Before;
  46 import org.junit.Test;
  47 import org.mockito.ArgumentCaptor;
  48 import org.mockito.InOrder;
  49 import org.mockito.Mockito;
  50 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
  51 import org.opendaylight.controller.cluster.datastore.persisted.CommitTransactionPayload;
  52 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  53 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
  54 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  55 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
  56 import org.opendaylight.yangtools.yang.common.Uint64;
  57 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  58 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  59 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  60 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  61 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
  62 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  63 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
  64 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
  65 import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
  66 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  67
  68 public class ShardDataTreeTest extends AbstractTest {
  69     private static final DatastoreContext DATASTORE_CONTEXT = DatastoreContext.newBuilder().build();
  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(mock(ShardStats.class)).when(mockShard).getShardMBean();
  79         doReturn(DATASTORE_CONTEXT).when(mockShard).getDatastoreContext();
  80
  81         fullSchema = SchemaContextHelper.full();
  82
  83         shardDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
  84     }
  85
  86     @Test
  87     public void testWrite() {
  88         modify(false, true, true);
  89     }
  90
  91     @Test
  92     public void testMerge() {
  93         modify(true, true, true);
  94     }
  95
  96     private void modify(final boolean merge, final boolean expectedCarsPresent, final boolean expectedPeoplePresent) {
  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, Optional.empty());
 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() {
 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() {
 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             Optional.empty(), 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 = new HashMap<>();
 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", Uint64.valueOf(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         assertTrue("Car node present", optional.isPresent());
 304         assertEquals("Car node", carNode, optional.get());
 305
 306         optional = snapshot.readNode(PeopleModel.BASE_PATH);
 307         assertTrue("People node present", 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", Uint64.valueOf(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         assertTrue("Car node present", optional.isPresent());
 351         assertEquals("Car node", carNode, optional.get());
 352     }
 353
 354     @Test
 355     public void testPipelinedTransactionsWithImmediateReplication() {
 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", Uint64.valueOf(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 = shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 378         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(CarsModel.BASE_PATH);
 379         assertTrue("Car node present", optional.isPresent());
 380     }
 381
 382     @SuppressWarnings("unchecked")
 383     @Test
 384     public void testAbortWithPendingCommits() throws Exception {
 385         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 386             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 387
 388         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 389             snapshot.write(PeopleModel.BASE_PATH, PeopleModel.create()));
 390
 391         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 392             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 393
 394         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 395         MapEntryNode carNode = CarsModel.newCarEntry("optima", Uint64.valueOf(100));
 396         final ShardDataTreeCohort cohort4 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
 397
 398         coordinatedCanCommit(cohort2);
 399         immediateCanCommit(cohort1);
 400         coordinatedCanCommit(cohort3);
 401         coordinatedCanCommit(cohort4);
 402
 403         coordinatedPreCommit(cohort1);
 404         coordinatedPreCommit(cohort2);
 405         coordinatedPreCommit(cohort3);
 406
 407         FutureCallback<Void> mockAbortCallback = mock(FutureCallback.class);
 408         doNothing().when(mockAbortCallback).onSuccess(null);
 409         cohort2.abort(mockAbortCallback);
 410         verify(mockAbortCallback).onSuccess(null);
 411
 412         coordinatedPreCommit(cohort4);
 413         coordinatedCommit(cohort1);
 414         coordinatedCommit(cohort3);
 415         coordinatedCommit(cohort4);
 416
 417         InOrder inOrder = inOrder(mockShard);
 418         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
 419                 eq(false));
 420         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
 421                 eq(false));
 422         inOrder.verify(mockShard).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
 423                 eq(false));
 424
 425         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
 426         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
 427                 cohort1.getCandidate());
 428         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
 429         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
 430         shardDataTree.applyReplicatedPayload(cohort4.getIdentifier(), mockPayload);
 431
 432         final DataTreeSnapshot snapshot =
 433                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 434         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
 435         assertTrue("Car node present", optional.isPresent());
 436         assertEquals("Car node", carNode, optional.get());
 437     }
 438
 439     @SuppressWarnings("unchecked")
 440     @Test
 441     public void testAbortWithFailedRebase() {
 442         immediatePayloadReplication(shardDataTree, mockShard);
 443
 444         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
 445             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
 446
 447         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
 448             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
 449
 450         NormalizedNode<?, ?> peopleNode = PeopleModel.create();
 451         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
 452             snapshot.write(PeopleModel.BASE_PATH, peopleNode));
 453
 454         immediateCanCommit(cohort1);
 455         FutureCallback<Void> canCommitCallback2 = coordinatedCanCommit(cohort2);
 456
 457         coordinatedPreCommit(cohort1);
 458         verify(canCommitCallback2).onSuccess(null);
 459
 460         FutureCallback<Void> mockAbortCallback = mock(FutureCallback.class);
 461         doNothing().when(mockAbortCallback).onSuccess(null);
 462         cohort1.abort(mockAbortCallback);
 463         verify(mockAbortCallback).onSuccess(null);
 464
 465         FutureCallback<DataTreeCandidate> preCommitCallback2 = coordinatedPreCommit(cohort2);
 466         verify(preCommitCallback2).onFailure(any(Throwable.class));
 467
 468         immediateCanCommit(cohort3);
 469         immediatePreCommit(cohort3);
 470         immediateCommit(cohort3);
 471
 472         final DataTreeSnapshot snapshot =
 473                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
 474         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(PeopleModel.BASE_PATH);
 475         assertTrue("People node present", optional.isPresent());
 476         assertEquals("People node", peopleNode, optional.get());
 477     }
 478
 479     private ShardDataTreeCohort newShardDataTreeCohort(final DataTreeOperation operation) {
 480         final ReadWriteShardDataTreeTransaction transaction =
 481                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 482         final DataTreeModification snapshot = transaction.getSnapshot();
 483         operation.execute(snapshot);
 484         return shardDataTree.finishTransaction(transaction, Optional.empty());
 485     }
 486
 487     @SuppressWarnings({ "rawtypes", "unchecked" })
 488     private static void verifyOnDataTreeChanged(final DOMDataTreeChangeListener listener,
 489             final Consumer<DataTreeCandidate> callback) {
 490         ArgumentCaptor<Collection> changes = ArgumentCaptor.forClass(Collection.class);
 491         verify(listener, atLeastOnce()).onDataTreeChanged(changes.capture());
 492         for (Collection list : changes.getAllValues()) {
 493             for (Object dtc : list) {
 494                 callback.accept((DataTreeCandidate)dtc);
 495             }
 496         }
 497
 498         reset(listener);
 499     }
 500
 501     private static NormalizedNode<?, ?> getCars(final ShardDataTree shardDataTree) {
 502         final ReadOnlyShardDataTreeTransaction readOnlyShardDataTreeTransaction =
 503                 shardDataTree.newReadOnlyTransaction(nextTransactionId());
 504         final DataTreeSnapshot snapshot1 = readOnlyShardDataTreeTransaction.getSnapshot();
 505
 506         final Optional<NormalizedNode<?, ?>> optional = snapshot1.readNode(CarsModel.BASE_PATH);
 507
 508         assertTrue(optional.isPresent());
 509
 510         return optional.get();
 511     }
 512
 513     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree) {
 514         return addCar(shardDataTree, "altima");
 515     }
 516
 517     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree, final String name) {
 518         return doTransaction(shardDataTree, snapshot -> {
 519             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 520             snapshot.merge(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 521             snapshot.write(CarsModel.newCarPath(name), CarsModel.newCarEntry(name, Uint64.valueOf(100)));
 522         });
 523     }
 524
 525     private static DataTreeCandidate removeCar(final ShardDataTree shardDataTree) {
 526         return doTransaction(shardDataTree, snapshot -> snapshot.delete(CarsModel.newCarPath("altima")));
 527     }
 528
 529     @FunctionalInterface
 530     private interface DataTreeOperation {
 531         void execute(DataTreeModification snapshot);
 532     }
 533
 534     private static DataTreeCandidate doTransaction(final ShardDataTree shardDataTree,
 535             final DataTreeOperation operation) {
 536         final ReadWriteShardDataTreeTransaction transaction =
 537                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 538         final DataTreeModification snapshot = transaction.getSnapshot();
 539         operation.execute(snapshot);
 540         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction, Optional.empty());
 541
 542         immediateCanCommit(cohort);
 543         immediatePreCommit(cohort);
 544         final DataTreeCandidate candidate = cohort.getCandidate();
 545         immediateCommit(cohort);
 546
 547         return candidate;
 548     }
 549
 550     private static DataTreeCandidate applyCandidates(final ShardDataTree shardDataTree,
 551             final List<DataTreeCandidate> candidates) {
 552         final ReadWriteShardDataTreeTransaction transaction =
 553                 shardDataTree.newReadWriteTransaction(nextTransactionId());
 554         final DataTreeModification snapshot = transaction.getSnapshot();
 555         for (final DataTreeCandidate candidateTip : candidates) {
 556             DataTreeCandidates.applyToModification(snapshot, candidateTip);
 557         }
 558         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction, Optional.empty());
 559
 560         immediateCanCommit(cohort);
 561         immediatePreCommit(cohort);
 562         final DataTreeCandidate candidate = cohort.getCandidate();
 563         immediateCommit(cohort);
 564
 565         return candidate;
 566     }
 567 }