2 * Copyright (c) 2014 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.datastore.compat;
10 import akka.actor.ActorRef;
11 import akka.actor.ActorSelection;
12 import akka.actor.PoisonPill;
13 import akka.actor.Props;
14 import akka.dispatch.Dispatchers;
15 import akka.testkit.TestActorRef;
16 import java.util.Collections;
17 import org.junit.Assert;
18 import org.junit.Test;
19 import org.opendaylight.controller.cluster.datastore.AbstractActorTest;
20 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
21 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
22 import org.opendaylight.controller.cluster.datastore.Shard;
23 import org.opendaylight.controller.cluster.datastore.ShardTest;
24 import org.opendaylight.controller.cluster.datastore.ShardTestKit;
25 import org.opendaylight.controller.cluster.datastore.TransactionProxy;
26 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
27 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
28 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
29 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
30 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
31 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
32 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
33 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
34 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
35 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
36 import org.opendaylight.controller.protobuff.messages.cohort3pc.ThreePhaseCommitCohortMessages;
37 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages;
38 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
39 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
40 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
41 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
42 import scala.concurrent.duration.FiniteDuration;
45 * Tests backwards compatibility support from Helium-1 to Helium.
47 * In Helium-1, the 3-phase commit support was moved from the ThreePhaseCommitCohort actor to the
48 * Shard. As a consequence, a new transactionId field was added to the CanCommitTransaction,
49 * CommitTransaction and AbortTransaction messages. With a base Helium version node, these messages
50 * would be sans transactionId so this test verifies the Shard handles that properly.
52 * @author Thomas Pantelis
54 public class ShardTransactionHeliumBackwardsCompatibilityTest extends AbstractActorTest {
57 public void testTransactionCommit() throws Exception {
58 new ShardTestKit(getSystem()) {{
59 SchemaContext schemaContext = TestModel.createTestContext();
60 Props shardProps = Shard.props(ShardIdentifier.builder().memberName("member-1").
61 shardName("inventory").type("config").build(),
62 Collections.<ShardIdentifier,String>emptyMap(),
63 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).build(),
64 schemaContext).withDispatcher(Dispatchers.DefaultDispatcherId());
66 final TestActorRef<Shard> shard = TestActorRef.create(getSystem(), shardProps,
67 "testTransactionCommit");
69 waitUntilLeader(shard);
71 // Send CreateTransaction message with no messages version
73 String transactionID = "txn-1";
74 shard.tell(ShardTransactionMessages.CreateTransaction.newBuilder()
75 .setTransactionId(transactionID)
76 .setTransactionType(TransactionProxy.TransactionType.WRITE_ONLY.ordinal())
77 .setTransactionChainId("").build(), getRef());
79 final FiniteDuration duration = duration("5 seconds");
81 CreateTransactionReply reply = expectMsgClass(duration, CreateTransactionReply.class);
83 ActorSelection txActor = getSystem().actorSelection(reply.getTransactionActorPath());
85 // Write data to the Tx
87 txActor.tell(new WriteData(TestModel.TEST_PATH,
88 ImmutableNodes.containerNode(TestModel.TEST_QNAME), DataStoreVersions.BASE_HELIUM_VERSION).
89 toSerializable(), getRef());
91 expectMsgClass(duration, ShardTransactionMessages.WriteDataReply.class);
95 txActor.tell(new ReadyTransaction().toSerializable(), getRef());
97 ReadyTransactionReply readyReply = ReadyTransactionReply.fromSerializable(expectMsgClass(
98 duration, ReadyTransactionReply.SERIALIZABLE_CLASS));
100 ActorSelection cohortActor = getSystem().actorSelection(readyReply.getCohortPath());
102 // Send the CanCommitTransaction message with no transactionId.
104 cohortActor.tell(ThreePhaseCommitCohortMessages.CanCommitTransaction.newBuilder().build(),
107 expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS);
109 // Send the PreCommitTransaction message with no transactionId.
111 cohortActor.tell(ThreePhaseCommitCohortMessages.PreCommitTransaction.newBuilder().build(),
114 expectMsgClass(duration, PreCommitTransactionReply.SERIALIZABLE_CLASS);
116 // Send the CommitTransaction message with no transactionId.
118 cohortActor.tell(ThreePhaseCommitCohortMessages.CommitTransaction.newBuilder().build(),
121 expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
123 NormalizedNode<?, ?> node = ShardTest.readStore(shard, TestModel.TEST_PATH);
124 Assert.assertNotNull("Data not found in store", node);
126 shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
131 public void testTransactionAbort() throws Exception {
132 new ShardTestKit(getSystem()) {{
133 SchemaContext schemaContext = TestModel.createTestContext();
134 Props shardProps = Shard.props(ShardIdentifier.builder().memberName("member-1").
135 shardName("inventory").type("config").build(),
136 Collections.<ShardIdentifier,String>emptyMap(),
137 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).build(),
138 schemaContext).withDispatcher(Dispatchers.DefaultDispatcherId());
140 final TestActorRef<Shard> shard = TestActorRef.create(getSystem(), shardProps,
141 "testTransactionAbort");
143 waitUntilLeader(shard);
145 // Send CreateTransaction message with no messages version
147 String transactionID = "txn-1";
148 shard.tell(ShardTransactionMessages.CreateTransaction.newBuilder()
149 .setTransactionId(transactionID)
150 .setTransactionType(TransactionProxy.TransactionType.WRITE_ONLY.ordinal())
151 .setTransactionChainId("").build(), getRef());
153 final FiniteDuration duration = duration("5 seconds");
155 CreateTransactionReply reply = expectMsgClass(duration, CreateTransactionReply.class);
157 ActorSelection txActor = getSystem().actorSelection(reply.getTransactionActorPath());
159 // Write data to the Tx
161 txActor.tell(new WriteData(TestModel.TEST_PATH,
162 ImmutableNodes.containerNode(TestModel.TEST_QNAME),
163 DataStoreVersions.BASE_HELIUM_VERSION).toSerializable(), getRef());
165 expectMsgClass(duration, WriteDataReply.INSTANCE.toSerializable(
166 DataStoreVersions.BASE_HELIUM_VERSION).getClass());
170 txActor.tell(new ReadyTransaction().toSerializable(), getRef());
172 ReadyTransactionReply readyReply = ReadyTransactionReply.fromSerializable(expectMsgClass(
173 duration, ReadyTransactionReply.SERIALIZABLE_CLASS));
175 ActorSelection cohortActor = getSystem().actorSelection(readyReply.getCohortPath());
177 // Send the CanCommitTransaction message with no transactionId.
179 cohortActor.tell(ThreePhaseCommitCohortMessages.CanCommitTransaction.newBuilder().build(),
182 expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS);
184 // Send the AbortTransaction message with no transactionId.
186 cohortActor.tell(ThreePhaseCommitCohortMessages.AbortTransaction.newBuilder().build(),
189 expectMsgClass(duration, AbortTransactionReply.SERIALIZABLE_CLASS);
191 shard.tell(PoisonPill.getInstance(), ActorRef.noSender());