+ @SuppressWarnings("checkstyle:IllegalCatch")
+ private void testTransactionCommitFailureWithNoShardLeader(final boolean writeOnly, final String testName)
+ throws Exception {
+ new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
+ {
+ String shardName = "default";
+
+ // We don't want the shard to become the leader so prevent shard
+ // elections.
+ datastoreContextBuilder.customRaftPolicyImplementation(
+ "org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy");
+
+ // The ShardManager uses the election timeout for FindPrimary so
+ // reset it low so it will timeout quickly.
+ datastoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1)
+ .shardInitializationTimeout(200, TimeUnit.MILLISECONDS);
+
+ try (AbstractDataStore dataStore = setupDistributedDataStore(testName, false, shardName)) {
+
+ Object result = dataStore.getActorContext().executeOperation(
+ dataStore.getActorContext().getShardManager(), new FindLocalShard(shardName, true));
+ assertTrue("Expected LocalShardFound. Actual: " + result, result instanceof LocalShardFound);
+
+ // Create the write Tx.
+
+ try (DOMStoreWriteTransaction writeTx = writeOnly ? dataStore.newWriteOnlyTransaction()
+ : dataStore.newReadWriteTransaction()) {
+ assertNotNull("newReadWriteTransaction returned null", writeTx);
+
+ // Do some modifications and ready the Tx on a separate
+ // thread.
+
+ final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
+ final AtomicReference<Exception> caughtEx = new AtomicReference<>();
+ final CountDownLatch txReady = new CountDownLatch(1);
+ Thread txThread = new Thread() {
+ @Override
+ public void run() {
+ try {
+ writeTx.write(TestModel.JUNK_PATH,
+ ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
+
+ txCohort.set(writeTx.ready());
+ } catch (Exception e) {
+ caughtEx.set(e);
+ return;
+ } finally {
+ txReady.countDown();
+ }
+ }
+ };
+
+ txThread.start();
+
+ // Wait for the Tx operations to complete.
+
+ boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
+ if (caughtEx.get() != null) {
+ throw caughtEx.get();
+ }
+
+ assertEquals("Tx ready", true, done);
+
+ // Wait for the commit to complete. Since no shard
+ // leader was elected in time, the Tx
+ // should have timed out and throw an appropriate
+ // exception cause.
+
+ try {
+ txCohort.get().canCommit().get(5, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ Throwables.propagateIfInstanceOf(e.getCause(), Exception.class);
+ Throwables.propagate(e.getCause());
+ }
+ }
+ }
+ }
+ };
+ }
+
+ @Test(expected = NoShardLeaderException.class)
+ public void testWriteOnlyTransactionCommitFailureWithNoShardLeader() throws Exception {