CDS: Retry remote front-end transactions on AskTimeoutException

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

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java
index 672d8f3b9997cc866d8320c11da3a7d55aa61e49..90600f952926de70e615377d7a97c61b9a9ec2eb 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java
@@ -9,6 +9,7 @@ package org.opendaylight.controller.cluster.datastore;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
@@ -19,17 +20,22 @@ import akka.actor.ActorSystem;
 import akka.actor.Address;
 import akka.actor.AddressFromURIString;
 import akka.cluster.Cluster;
+import akka.pattern.AskTimeoutException;
 import akka.testkit.JavaTestKit;
 import com.google.common.base.Optional;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.Uninterruptibles;
 import com.typesafe.config.ConfigFactory;
 import java.math.BigInteger;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
+import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
+import org.opendaylight.controller.cluster.datastore.exceptions.ShardLeaderNotRespondingException;
 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
@@ -75,10 +81,12 @@ public class DistributedDataStoreRemotingIntegrationTest {
     private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
     private static final Address MEMBER_2_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2559");
 
-    private static final String MODULE_SHARDS_CONFIG = "module-shards-member1-and-2.conf";
+    private static final String MODULE_SHARDS_CONFIG_2 = "module-shards-member1-and-2.conf";
+    private static final String MODULE_SHARDS_CONFIG_3 = "module-shards-member1-and-2-and-3.conf";
 
     private ActorSystem leaderSystem;
     private ActorSystem followerSystem;
+    private ActorSystem follower2System;
 
     private final DatastoreContext.Builder leaderDatastoreContextBuilder =
             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1);
@@ -92,27 +100,35 @@ public class DistributedDataStoreRemotingIntegrationTest {
     private IntegrationTestKit leaderTestKit;
 
     @Before
-    public void setUpClass() {
+    public void setUp() {
         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
         Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
 
         followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
         Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);
+
+        follower2System = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member3"));
+        Cluster.get(follower2System).join(MEMBER_1_ADDRESS);
     }
 
     @After
-    public void tearDownClass() {
+    public void tearDown() {
         JavaTestKit.shutdownActorSystem(leaderSystem);
         JavaTestKit.shutdownActorSystem(followerSystem);
+        JavaTestKit.shutdownActorSystem(follower2System);
     }
 
     private void initDatastores(String type) {
+        initDatastores(type, MODULE_SHARDS_CONFIG_2);
+    }
+
+    private void initDatastores(String type, String moduleShardsConfig) {
         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
 
-        followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
-        followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, MODULE_SHARDS_CONFIG, false, SHARD_NAMES);
+        leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, SHARD_NAMES);
 
-        leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(type, MODULE_SHARDS_CONFIG, false, SHARD_NAMES);
+        followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
+        followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, SHARD_NAMES);
 
         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), SHARD_NAMES);
     }
@@ -478,8 +494,7 @@ public class DistributedDataStoreRemotingIntegrationTest {
         DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder().
                 shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
         IntegrationTestKit newMember1TestKit = new IntegrationTestKit(leaderSystem, newMember1Builder);
-        DistributedDataStore newMember1Datastore = newMember1TestKit.
-                    setupDistributedDataStore(testName, MODULE_SHARDS_CONFIG, false, SHARD_NAMES);
+        newMember1TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CONFIG_2, false, SHARD_NAMES);
 
         followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), SHARD_NAMES);
 
@@ -521,4 +536,115 @@ public class DistributedDataStoreRemotingIntegrationTest {
 
         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car);
     }
+
+    @Test(expected=NoShardLeaderException.class)
+    public void testTransactionWithIsolatedLeader() throws Throwable {
+        leaderDatastoreContextBuilder.shardIsolatedLeaderCheckIntervalInMillis(300);
+        String testName = "testTransactionWithIsolatedLeader";
+        initDatastores(testName);
+
+        JavaTestKit.shutdownActorSystem(followerSystem, null, true);
+
+        Uninterruptibles.sleepUninterruptibly(leaderDistributedDataStore.getActorContext().getDatastoreContext()
+                .getShardRaftConfig().getElectionTimeOutInterval().toMillis() * 3, TimeUnit.MILLISECONDS);
+
+        DOMStoreWriteTransaction writeTx = leaderDistributedDataStore.newWriteOnlyTransaction();
+        writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+        try {
+            followerTestKit.doCommit(writeTx.ready());
+        } catch (ExecutionException e) {
+            throw e.getCause();
+        }
+    }
+
+    @Test(expected=AskTimeoutException.class)
+    public void testTransactionWithShardLeaderNotResponding() throws Throwable {
+        followerDatastoreContextBuilder.shardElectionTimeoutFactor(30);
+        initDatastores("testTransactionWithShardLeaderNotResponding");
+
+        // Do an initial read to get the primary shard info cached.
+
+        DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
+        readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS);
+
+        // Shutdown the leader and try to create a new tx.
+
+        JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
+
+        followerDatastoreContextBuilder.operationTimeoutInMillis(50).shardElectionTimeoutFactor(1);
+        followerDistributedDataStore.onDatastoreContextUpdated(followerDatastoreContextBuilder.build());
+
+        DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+
+        rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+        try {
+            followerTestKit.doCommit(rwTx.ready());
+        } catch (ExecutionException e) {
+            assertTrue("Expected ShardLeaderNotRespondingException cause. Actual: " + e.getCause(),
+                    e.getCause() instanceof ShardLeaderNotRespondingException);
+            assertNotNull("Expected a nested cause", e.getCause().getCause());
+            throw e.getCause().getCause();
+        }
+    }
+
+    @Test(expected=NoShardLeaderException.class)
+    public void testTransactionWithCreateTxFailureDueToNoLeader() throws Throwable {
+        initDatastores("testTransactionWithCreateTxFailureDueToNoLeader");
+
+        // Do an initial read to get the primary shard info cached.
+
+        DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
+        readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS);
+
+        // Shutdown the leader and try to create a new tx.
+
+        JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
+
+        Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+        followerDatastoreContextBuilder.operationTimeoutInMillis(10).shardElectionTimeoutFactor(1);
+        followerDistributedDataStore.onDatastoreContextUpdated(followerDatastoreContextBuilder.build());
+
+        DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+
+        rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+        try {
+            followerTestKit.doCommit(rwTx.ready());
+        } catch (ExecutionException e) {
+            throw e.getCause();
+        }
+    }
+
+    @Test
+    public void testTransactionRetryWithInitialAskTimeoutExOnCreateTx() throws Exception {
+        followerDatastoreContextBuilder.shardElectionTimeoutFactor(30);
+        String testName = "testTransactionRetryWithInitialAskTimeoutExOnCreateTx";
+        initDatastores(testName, MODULE_SHARDS_CONFIG_3);
+
+        DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder().
+                shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
+        IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, follower2DatastoreContextBuilder);
+        follower2TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CONFIG_3, false, SHARD_NAMES);
+
+        // Do an initial read to get the primary shard info cached.
+
+        DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
+        readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS);
+
+        // Shutdown the leader and try to create a new tx.
+
+        JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
+
+        followerDatastoreContextBuilder.operationTimeoutInMillis(500);
+        followerDistributedDataStore.onDatastoreContextUpdated(followerDatastoreContextBuilder.build());
+
+        DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+
+        rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+        followerTestKit.doCommit(rwTx.ready());
+    }
 }