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

   1 /*
   2  * Copyright (c) 2015 Brocade Communications 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.fail;
  13 import static org.mockito.Matchers.any;
  14 import static org.mockito.Matchers.eq;
  15 import static org.mockito.Mockito.timeout;
  16 import static org.mockito.Mockito.verify;
  17 import akka.actor.ActorRef;
  18 import akka.actor.ActorSystem;
  19 import akka.actor.Address;
  20 import akka.actor.AddressFromURIString;
  21 import akka.cluster.Cluster;
  22 import akka.testkit.JavaTestKit;
  23 import com.google.common.base.Optional;
  24 import com.google.common.collect.ImmutableMap;
  25 import com.google.common.util.concurrent.MoreExecutors;
  26 import com.typesafe.config.ConfigFactory;
  27 import java.math.BigInteger;
  28 import java.util.concurrent.TimeUnit;
  29 import org.junit.After;
  30 import org.junit.Before;
  31 import org.junit.Test;
  32 import org.mockito.Mockito;
  33 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  34 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
  35 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  36 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  37 import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
  38 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  39 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  40 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
  41 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  42 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  43 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  44 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
  45 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  46 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  47 import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
  48 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
  49 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  50 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  51 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  52 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  53 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  54 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  55 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  56 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  57 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  58 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  59 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  60 import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTree;
  61 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  62 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder;
  63 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  64 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
  65
  66 /**
  67  * End-to-end distributed data store tests that exercise remote shards and transactions.
  68  *
  69  * @author Thomas Pantelis
  70  */
  71 public class DistributedDataStoreRemotingIntegrationTest {
  72
  73     private static final String[] SHARD_NAMES = {"cars", "people"};
  74
  75     private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
  76     private static final Address MEMBER_2_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2559");
  77
  78     private static final String MODULE_SHARDS_CONFIG = "module-shards-member1-and-2.conf";
  79
  80     private ActorSystem leaderSystem;
  81     private ActorSystem followerSystem;
  82
  83     private final DatastoreContext.Builder leaderDatastoreContextBuilder =
  84             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1);
  85
  86     private final DatastoreContext.Builder followerDatastoreContextBuilder =
  87             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
  88
  89     private DistributedDataStore followerDistributedDataStore;
  90     private DistributedDataStore leaderDistributedDataStore;
  91     private IntegrationTestKit followerTestKit;
  92     private IntegrationTestKit leaderTestKit;
  93
  94     @Before
  95     public void setUpClass() {
  96         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
  97         Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
  98
  99         followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
 100         Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);
 101     }
 102
 103     @After
 104     public void tearDownClass() {
 105         JavaTestKit.shutdownActorSystem(leaderSystem);
 106         JavaTestKit.shutdownActorSystem(followerSystem);
 107     }
 108
 109     private void initDatastores(String type) {
 110         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
 111
 112         followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
 113         followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, MODULE_SHARDS_CONFIG, false, SHARD_NAMES);
 114
 115         leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(type, MODULE_SHARDS_CONFIG, false, SHARD_NAMES);
 116
 117         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), SHARD_NAMES);
 118     }
 119
 120     private void verifyCars(DOMStoreReadTransaction readTx, MapEntryNode... entries) throws Exception {
 121         Optional<NormalizedNode<?, ?>> optional = readTx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
 122         assertEquals("isPresent", true, optional.isPresent());
 123
 124         CollectionNodeBuilder<MapEntryNode, MapNode> listBuilder = ImmutableNodes.mapNodeBuilder(CarsModel.CAR_QNAME);
 125         for(NormalizedNode<?, ?> entry: entries) {
 126             listBuilder.withChild((MapEntryNode) entry);
 127         }
 128
 129         assertEquals("Car list node", listBuilder.build(), optional.get());
 130     }
 131
 132     private void verifyNode(DOMStoreReadTransaction readTx, YangInstanceIdentifier path, NormalizedNode<?, ?> expNode)
 133             throws Exception {
 134         Optional<NormalizedNode<?, ?>> optional = readTx.read(path).get(5, TimeUnit.SECONDS);
 135         assertEquals("isPresent", true, optional.isPresent());
 136         assertEquals("Data node", expNode, optional.get());
 137     }
 138
 139     private void verifyExists(DOMStoreReadTransaction readTx, YangInstanceIdentifier path) throws Exception {
 140         Boolean exists = readTx.exists(path).get(5, TimeUnit.SECONDS);
 141         assertEquals("exists", true, exists);
 142     }
 143
 144     @Test
 145     public void testWriteTransactionWithSingleShard() throws Exception {
 146         String testName = "testWriteTransactionWithSingleShard";
 147         initDatastores(testName);
 148
 149         String followerCarShardName = "member-2-shard-cars-" + testName;
 150         InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 2, ApplyJournalEntries.class );
 151
 152         DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 153         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 154
 155         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 156         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 157
 158         MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
 159         YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
 160         writeTx.merge(car1Path, car1);
 161
 162         MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(25000));
 163         YangInstanceIdentifier car2Path = CarsModel.newCarPath("sportage");
 164         writeTx.merge(car2Path, car2);
 165
 166         followerTestKit.doCommit(writeTx.ready());
 167
 168         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1, car2);
 169
 170         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2);
 171
 172         // Test delete
 173
 174         writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 175
 176         writeTx.delete(car1Path);
 177
 178         followerTestKit.doCommit(writeTx.ready());
 179
 180         verifyExists(followerDistributedDataStore.newReadOnlyTransaction(), car2Path);
 181
 182         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car2);
 183
 184         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car2);
 185
 186         // Re-instate the follower member 2 as a single-node to verify replication and recovery.
 187
 188         InMemoryJournal.waitForWriteMessagesComplete(followerCarShardName);
 189
 190         JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
 191         JavaTestKit.shutdownActorSystem(followerSystem, null, true);
 192
 193         ActorSystem newSystem = ActorSystem.create("reinstated-member2", ConfigFactory.load().getConfig("Member2"));
 194
 195         DistributedDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder).
 196                 setupDistributedDataStore(testName, "module-shards-member2", true, SHARD_NAMES);
 197
 198         verifyCars(member2Datastore.newReadOnlyTransaction(), car2);
 199
 200         JavaTestKit.shutdownActorSystem(newSystem);
 201     }
 202
 203     @Test
 204     public void testReadWriteTransactionWithSingleShard() throws Exception {
 205         initDatastores("testReadWriteTransactionWithSingleShard");
 206
 207         DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
 208         assertNotNull("newReadWriteTransaction returned null", rwTx);
 209
 210         rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 211         rwTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 212
 213         MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
 214         rwTx.merge(CarsModel.newCarPath("optima"), car1);
 215
 216         verifyCars(rwTx, car1);
 217
 218         MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(25000));
 219         YangInstanceIdentifier car2Path = CarsModel.newCarPath("sportage");
 220         rwTx.merge(car2Path, car2);
 221
 222         verifyExists(rwTx, car2Path);
 223
 224         followerTestKit.doCommit(rwTx.ready());
 225
 226         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1, car2);
 227     }
 228
 229     @Test
 230     public void testWriteTransactionWithMultipleShards() throws Exception {
 231         initDatastores("testWriteTransactionWithMultipleShards");
 232
 233         DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 234         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 235
 236         YangInstanceIdentifier carsPath = CarsModel.BASE_PATH;
 237         NormalizedNode<?, ?> carsNode = CarsModel.emptyContainer();
 238         writeTx.write(carsPath, carsNode);
 239
 240         YangInstanceIdentifier peoplePath = PeopleModel.BASE_PATH;
 241         NormalizedNode<?, ?> peopleNode = PeopleModel.emptyContainer();
 242         writeTx.write(peoplePath, peopleNode);
 243
 244         followerTestKit.doCommit(writeTx.ready());
 245
 246         DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
 247
 248         verifyNode(readTx, carsPath, carsNode);
 249         verifyNode(readTx, peoplePath, peopleNode);
 250     }
 251
 252     @Test
 253     public void testReadWriteTransactionWithMultipleShards() throws Exception {
 254         initDatastores("testReadWriteTransactionWithMultipleShards");
 255
 256         DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
 257         assertNotNull("newReadWriteTransaction returned null", rwTx);
 258
 259         YangInstanceIdentifier carsPath = CarsModel.BASE_PATH;
 260         NormalizedNode<?, ?> carsNode = CarsModel.emptyContainer();
 261         rwTx.write(carsPath, carsNode);
 262
 263         YangInstanceIdentifier peoplePath = PeopleModel.BASE_PATH;
 264         NormalizedNode<?, ?> peopleNode = PeopleModel.emptyContainer();
 265         rwTx.write(peoplePath, peopleNode);
 266
 267         followerTestKit.doCommit(rwTx.ready());
 268
 269         DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
 270
 271         verifyNode(readTx, carsPath, carsNode);
 272         verifyNode(readTx, peoplePath, peopleNode);
 273     }
 274
 275     @Test
 276     public void testTransactionChainWithSingleShard() throws Exception {
 277         initDatastores("testTransactionChainWithSingleShard");
 278
 279         DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
 280
 281         // Add the top-level cars container with write-only.
 282
 283         DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 284         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 285
 286         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 287
 288         writeTx.ready();
 289
 290         // Verify the top-level cars container with read-only.
 291
 292         verifyNode(txChain.newReadOnlyTransaction(), CarsModel.BASE_PATH, CarsModel.emptyContainer());
 293
 294         // Perform car operations with read-write.
 295
 296         DOMStoreReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
 297
 298         verifyNode(rwTx, CarsModel.BASE_PATH, CarsModel.emptyContainer());
 299
 300         rwTx.merge(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 301
 302         MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
 303         YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
 304         rwTx.write(car1Path, car1);
 305
 306         verifyExists(rwTx, car1Path);
 307
 308         verifyCars(rwTx, car1);
 309
 310         MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(25000));
 311         rwTx.merge(CarsModel.newCarPath("sportage"), car2);
 312
 313         rwTx.delete(car1Path);
 314
 315         followerTestKit.doCommit(rwTx.ready());
 316
 317         txChain.close();
 318
 319         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car2);
 320     }
 321
 322     @Test
 323     public void testTransactionChainWithMultipleShards() throws Exception{
 324         initDatastores("testTransactionChainWithMultipleShards");
 325
 326         DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
 327
 328         DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 329         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 330
 331         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 332         writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 333
 334         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 335         writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
 336
 337         followerTestKit.doCommit(writeTx.ready());
 338
 339         DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction();
 340
 341         MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
 342         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 343         readWriteTx.write(carPath, car);
 344
 345         MapEntryNode person = PeopleModel.newPersonEntry("jack");
 346         YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
 347         readWriteTx.merge(personPath, person);
 348
 349         Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
 350         assertEquals("isPresent", true, optional.isPresent());
 351         assertEquals("Data node", car, optional.get());
 352
 353         optional = readWriteTx.read(personPath).get(5, TimeUnit.SECONDS);
 354         assertEquals("isPresent", true, optional.isPresent());
 355         assertEquals("Data node", person, optional.get());
 356
 357         DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready();
 358
 359         writeTx = txChain.newWriteOnlyTransaction();
 360
 361         writeTx.delete(personPath);
 362
 363         DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
 364
 365         followerTestKit.doCommit(cohort2);
 366         followerTestKit.doCommit(cohort3);
 367
 368         txChain.close();
 369
 370         DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
 371         verifyCars(readTx, car);
 372
 373         optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
 374         assertEquals("isPresent", false, optional.isPresent());
 375     }
 376
 377     @Test
 378     public void testChainedTransactionFailureWithSingleShard() throws Exception {
 379         initDatastores("testChainedTransactionFailureWithSingleShard");
 380
 381         ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
 382                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder().put(
 383                         LogicalDatastoreType.CONFIGURATION, followerDistributedDataStore).build(),
 384                         MoreExecutors.directExecutor());
 385
 386         TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
 387         DOMTransactionChain txChain = broker.createTransactionChain(listener);
 388
 389         DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 390
 391         ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
 392                 new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)).
 393                     withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
 394
 395         writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
 396
 397         try {
 398             writeTx.submit().checkedGet(5, TimeUnit.SECONDS);
 399             fail("Expected TransactionCommitFailedException");
 400         } catch (TransactionCommitFailedException e) {
 401             // Expected
 402         }
 403
 404         verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx), any(Throwable.class));
 405
 406         txChain.close();
 407         broker.close();
 408     }
 409
 410     @Test
 411     public void testChainedTransactionFailureWithMultipleShards() throws Exception {
 412         initDatastores("testChainedTransactionFailureWithMultipleShards");
 413
 414         ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
 415                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder().put(
 416                         LogicalDatastoreType.CONFIGURATION, followerDistributedDataStore).build(),
 417                         MoreExecutors.directExecutor());
 418
 419         TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
 420         DOMTransactionChain txChain = broker.createTransactionChain(listener);
 421
 422         DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 423
 424         writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 425
 426         ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
 427                 new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)).
 428                     withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
 429
 430         // Note that merge will validate the data and fail but put succeeds b/c deep validation is not
 431         // done for put for performance reasons.
 432         writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
 433
 434         try {
 435             writeTx.submit().checkedGet(5, TimeUnit.SECONDS);
 436             fail("Expected TransactionCommitFailedException");
 437         } catch (TransactionCommitFailedException e) {
 438             // Expected
 439         }
 440
 441         verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx), any(Throwable.class));
 442
 443         txChain.close();
 444         broker.close();
 445     }
 446
 447     @Test
 448     public void testSingleShardTransactionsWithLeaderChanges() throws Exception {
 449         String testName = "testSingleShardTransactionsWithLeaderChanges";
 450         initDatastores(testName);
 451
 452         String followerCarShardName = "member-2-shard-cars-" + testName;
 453         InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 1, ApplyJournalEntries.class );
 454
 455         // Write top-level car container from the follower so it uses a remote Tx.
 456
 457         DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 458
 459         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 460         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 461
 462         followerTestKit.doCommit(writeTx.ready());
 463
 464         InMemoryJournal.waitForWriteMessagesComplete(followerCarShardName);
 465
 466         // Switch the leader to the follower
 467
 468         followerDatastoreContextBuilder.shardElectionTimeoutFactor(1);
 469         followerDistributedDataStore.onDatastoreContextUpdated(followerDatastoreContextBuilder.build());
 470
 471         JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
 472
 473         followerTestKit.waitUntilNoLeader(followerDistributedDataStore.getActorContext(), SHARD_NAMES);
 474
 475         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
 476         Cluster.get(leaderSystem).join(MEMBER_2_ADDRESS);
 477
 478         DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder().
 479                 shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
 480         IntegrationTestKit newMember1TestKit = new IntegrationTestKit(leaderSystem, newMember1Builder);
 481         DistributedDataStore newMember1Datastore = newMember1TestKit.
 482                     setupDistributedDataStore(testName, MODULE_SHARDS_CONFIG, false, SHARD_NAMES);
 483
 484         followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), SHARD_NAMES);
 485
 486         // Write a car entry to the new leader - should switch to local Tx
 487
 488         writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 489
 490         MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
 491         YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
 492         writeTx.merge(car1Path, car1);
 493
 494         followerTestKit.doCommit(writeTx.ready());
 495
 496         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1);
 497     }
 498
 499     @Test
 500     public void testReadyLocalTransactionForwardedToLeader() throws Exception {
 501         initDatastores("testReadyLocalTransactionForwardedToLeader");
 502
 503         Optional<ActorRef> carsFollowerShard = followerDistributedDataStore.getActorContext().findLocalShard("cars");
 504         assertEquals("Cars follower shard found", true, carsFollowerShard.isPresent());
 505
 506         TipProducingDataTree dataTree = InMemoryDataTreeFactory.getInstance().create();
 507         dataTree.setSchemaContext(SchemaContextHelper.full());
 508         DataTreeModification modification = dataTree.takeSnapshot().newModification();
 509
 510         new WriteModification(CarsModel.BASE_PATH, CarsModel.emptyContainer()).apply(modification);
 511         new MergeModification(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()).apply(modification);
 512
 513         MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
 514         new WriteModification(CarsModel.newCarPath("optima"), car).apply(modification);
 515
 516         String transactionID = "tx-1";
 517         ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(transactionID , modification, true);
 518
 519         carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef());
 520         followerTestKit.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS);
 521
 522         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car);
 523     }
 524 }