import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
-import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
import org.opendaylight.controller.cluster.datastore.messages.ReadData;
import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
-import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
-import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshot;
+import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListenerReply;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
-import org.opendaylight.controller.cluster.raft.Snapshot;
import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
-import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
+import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
shard.tell(new UpdateSchemaContext(SchemaContextHelper.full()), ActorRef.noSender());
final MockDataChangeListener listener = new MockDataChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener),
- "testRegisterChangeListener-DataChangeListener");
+ final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener,
+ TestModel.TEST_PATH), "testRegisterChangeListener-DataChangeListener");
shard.tell(new RegisterChangeListener(TestModel.TEST_PATH, dclActor,
AsyncDataBroker.DataChangeScope.BASE, true), getRef());
setupInMemorySnapshotStore();
+ final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataChangeListener listener = new MockDataChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener, path),
"testRegisterChangeListenerWhenNotLeaderInitially-DataChangeListener");
final TestActorRef<Shard> shard = actorFactory.createTestActor(
new ShardTestKit(getSystem()) {
{
- final YangInstanceIdentifier path = TestModel.TEST_PATH;
-
// Wait until the shard receives the first ElectionTimeout
// message.
assertEquals("Got first ElectionTimeout", true, onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
shard.tell(new UpdateSchemaContext(SchemaContextHelper.full()), ActorRef.noSender());
final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener),
- "testRegisterDataTreeChangeListener-DataTreeChangeListener");
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener,
+ TestModel.TEST_PATH), "testRegisterDataTreeChangeListener-DataTreeChangeListener");
shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, false), getRef());
setupInMemorySnapshotStore();
+ final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener, path),
"testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration-DataChangeListener");
final TestActorRef<Shard> shard = actorFactory.createTestActor(
Props.create(new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()),
"testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration");
- final YangInstanceIdentifier path = TestModel.TEST_PATH;
-
new ShardTestKit(getSystem()) {
{
assertEquals("Got first ElectionTimeout", true, onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
final YangInstanceIdentifier root = YangInstanceIdentifier.EMPTY;
final NormalizedNode<?,?> expected = readStore(store, root);
- final Snapshot snapshot = Snapshot.create(new MetadataShardDataTreeSnapshot(expected).serialize(),
- Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4);
+ final Snapshot snapshot = Snapshot.create(
+ new ShardSnapshotState(new MetadataShardDataTreeSnapshot(expected)),
+ Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4, -1, null, null);
shard.tell(new ApplySnapshot(snapshot), ActorRef.noSender());
final DataTree store = InMemoryDataTreeFactory.getInstance().create(TreeType.OPERATIONAL);
store.setSchemaContext(SCHEMA_CONTEXT);
- writeToStore(store, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
-
- final NormalizedNode<?, ?> root = readStore(store, YangInstanceIdentifier.EMPTY);
- final Snapshot snapshot = Snapshot.create(new MetadataShardDataTreeSnapshot(root).serialize(),
- Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4);
-
- shard.tell(new ApplySnapshot(snapshot), ActorRef.noSender());
final DataTreeModification writeMod = store.takeSnapshot().newModification();
final ContainerNode node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
writeMod.ready();
final TransactionIdentifier tx = nextTransactionId();
- final ApplyState applyState = new ApplyState(null, tx,
- new SimpleReplicatedLogEntry(1, 2, payloadForModification(store, writeMod, tx)));
-
- shard.tell(applyState, shard);
+ shard.underlyingActor().applyState(null, null, payloadForModification(store, writeMod, tx));
final Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
};
}
- @Test
- public void testReadWriteCommitWhenTransactionHasNoModifications() {
- testCommitWhenTransactionHasNoModifications(true);
- }
-
- @Test
- public void testWriteOnlyCommitWhenTransactionHasNoModifications() {
- testCommitWhenTransactionHasNoModifications(false);
- }
-
- private void testCommitWhenTransactionHasNoModifications(final boolean readWrite) {
- // Note that persistence is enabled which would normally result in the
- // entry getting written to the journal
- // but here that need not happen
- new ShardTestKit(getSystem()) {
- {
- final TestActorRef<Shard> shard = actorFactory.createTestActor(
- newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
- "testCommitWhenTransactionHasNoModifications-" + readWrite);
-
- waitUntilLeader(shard);
-
- final TransactionIdentifier transactionID = nextTransactionId();
-
- final FiniteDuration duration = duration("5 seconds");
-
- if (readWrite) {
- final ReadWriteShardDataTreeTransaction rwTx = shard.underlyingActor().getDataStore()
- .newReadWriteTransaction(transactionID);
- shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION, rwTx, false), getRef());
- } else {
- shard.tell(prepareBatchedModifications(transactionID, new MutableCompositeModification()),
- getRef());
- }
-
- expectMsgClass(duration, ReadyTransactionReply.class);
-
- // Send the CanCommitTransaction message.
-
- shard.tell(new CanCommitTransaction(transactionID, CURRENT_VERSION).toSerializable(), getRef());
- final CanCommitTransactionReply canCommitReply = CanCommitTransactionReply
- .fromSerializable(expectMsgClass(duration, CanCommitTransactionReply.class));
- assertEquals("Can commit", true, canCommitReply.getCanCommit());
-
- shard.tell(new CommitTransaction(transactionID, CURRENT_VERSION).toSerializable(), getRef());
- expectMsgClass(duration, CommitTransactionReply.class);
-
- shard.tell(Shard.GET_SHARD_MBEAN_MESSAGE, getRef());
- final ShardStats shardStats = expectMsgClass(duration, ShardStats.class);
-
- // Use MBean for verification
- // Committed transaction count should increase as usual
- assertEquals(1, shardStats.getCommittedTransactionsCount());
-
- // Commit index should not advance because this does not go into
- // the journal
- assertEquals(-1, shardStats.getCommitIndex());
- }
- };
- }
-
@Test
public void testReadWriteCommitWhenTransactionHasModifications() throws Exception {
testCommitWhenTransactionHasModifications(true);
private void verifySnapshot(final Snapshot snapshot, final NormalizedNode<?, ?> expectedRoot)
throws IOException {
- final NormalizedNode<?, ?> actual = ShardDataTreeSnapshot.deserialize(snapshot.getState()).getRootNode()
- .get();
+ final NormalizedNode<?, ?> actual = ((ShardSnapshotState)snapshot.getState()).getSnapshot()
+ .getRootNode().get();
assertEquals("Root node", expectedRoot, actual);
}
};
dataStoreContextBuilder.shardElectionTimeoutFactor(1000)
.customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
+ final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataChangeListener listener = new MockDataChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener, path),
actorFactory.generateActorId(testName + "-DataChangeListener"));
setupInMemorySnapshotStore();
waitUntilNoLeader(shard);
- final YangInstanceIdentifier path = TestModel.TEST_PATH;
-
shard.tell(new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.BASE, true),
getRef());
final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataChangeListener listener = new MockDataChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener, path),
actorFactory.generateActorId(testName + "-DataChangeListener"));
followerShard.tell(
.customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener),
- actorFactory.generateActorId(testName + "-DataTreeChangeListener"));
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener,
+ TestModel.TEST_PATH), actorFactory.generateActorId(testName + "-DataTreeChangeListener"));
setupInMemorySnapshotStore();
final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener, path),
actorFactory.generateActorId(testName + "-DataTreeChangeListener"));
followerShard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), getRef());