import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeNotificationListenerRegistration;
+import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeNotificationListenerRegistrationReply;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
-import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
-import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListenerReply;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeNotificationListenerReply;
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.md.cluster.datastore.model.TestModel;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
+import org.opendaylight.yangtools.concepts.Identifier;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
shard.tell(new RegisterChangeListener(TestModel.TEST_PATH, dclActor,
AsyncDataBroker.DataChangeScope.BASE, true), getRef());
- final RegisterChangeListenerReply reply = expectMsgClass(duration("3 seconds"),
- RegisterChangeListenerReply.class);
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("3 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
final String replyPath = reply.getListenerRegistrationPath().toString();
assertTrue("Incorrect reply path: " + replyPath,
replyPath.matches("akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
// original ElectionTimeout message to proceed with the election.
firstElectionTimeout = false;
final ActorRef self = getSelf();
- new Thread() {
- @Override
- public void run() {
- Uninterruptibles.awaitUninterruptibly(
- onChangeListenerRegistered, 5, TimeUnit.SECONDS);
- self.tell(message, self);
- }
- }.start();
+ new Thread(() -> {
+ Uninterruptibles.awaitUninterruptibly(
+ onChangeListenerRegistered, 5, TimeUnit.SECONDS);
+ self.tell(message, self);
+ }).start();
onFirstElectionTimeout.countDown();
} else {
shard.tell(new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.SUBTREE, false),
getRef());
- final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
- RegisterChangeListenerReply.class);
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());
// Sanity check - verify the shard is not the leader yet.
shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, false), getRef());
- final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("3 seconds"),
- RegisterDataTreeChangeListenerReply.class);
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("3 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
final String replyPath = reply.getListenerRegistrationPath().toString();
assertTrue("Incorrect reply path: " + replyPath,
replyPath.matches("akka:\\/\\/test\\/user\\/testRegisterDataTreeChangeListener\\/\\$.*"));
if (message instanceof ElectionTimeout && firstElectionTimeout) {
firstElectionTimeout = false;
final ActorRef self = getSelf();
- new Thread() {
- @Override
- public void run() {
- Uninterruptibles.awaitUninterruptibly(
- onChangeListenerRegistered, 5, TimeUnit.SECONDS);
- self.tell(message, self);
- }
- }.start();
+ new Thread(() -> {
+ Uninterruptibles.awaitUninterruptibly(
+ onChangeListenerRegistered, 5, TimeUnit.SECONDS);
+ self.tell(message, self);
+ }).start();
onFirstElectionTimeout.countDown();
} else {
assertEquals("Got first ElectionTimeout", true, onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
shard.tell(new RegisterDataTreeChangeListener(path, dclActor, false), getRef());
- final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
- RegisterDataTreeChangeListenerReply.class);
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath());
shard.tell(FindLeader.INSTANCE, getRef());
CreateTransactionReply.class);
final String path = reply.getTransactionPath().toString();
- assertTrue("Unexpected transaction path " + path, path
- .startsWith("akka://test/user/testCreateTransaction/shard-member-1:ShardTransactionTest@0:"));
+ assertTrue("Unexpected transaction path " + path, path.startsWith(String.format(
+ "akka://test/user/testCreateTransaction/shard-%s-%s:ShardTransactionTest@0:",
+ shardID.getShardName(), shardID.getMemberName().getName())));
}
};
}
CreateTransactionReply.class);
final String path = reply.getTransactionPath().toString();
- assertTrue("Unexpected transaction path " + path, path.startsWith(
- "akka://test/user/testCreateTransactionOnChain/shard-member-1:ShardTransactionTest@0:"));
+ assertTrue("Unexpected transaction path " + path, path.startsWith(String.format(
+ "akka://test/user/testCreateTransactionOnChain/shard-%s-%s:ShardTransactionTest@0:",
+ shardID.getShardName(), shardID.getMemberName().getName())));
}
};
}
final ReadyTransactionReply readyReply = ReadyTransactionReply
.fromSerializable(expectMsgClass(duration, ReadyTransactionReply.class));
assertEquals("Cohort path", shard.path().toString(), readyReply.getCohortPath());
-
// Send the CanCommitTransaction message for the first Tx.
shard.tell(new CanCommitTransaction(transactionID1, CURRENT_VERSION).toSerializable(), getRef());
final boolean done = commitLatch.await(timeoutSec, TimeUnit.SECONDS);
- if (caughtEx.get() != null) {
- Throwables.propagateIfInstanceOf(caughtEx.get(), Exception.class);
- Throwables.propagate(caughtEx.get());
+ final Throwable t = caughtEx.get();
+ if (t != null) {
+ Throwables.propagateIfPossible(t, Exception.class);
+ throw new RuntimeException(t);
}
assertEquals("Commits complete", true, done);
verifyOuterListEntry(shard, 1);
- verifyLastApplied(shard, 2);
+ verifyLastApplied(shard, 5);
}
};
}
final Failure failure = expectMsgClass(duration("5 seconds"), akka.actor.Status.Failure.class);
if (failure != null) {
- Throwables.propagateIfInstanceOf(failure.cause(), Exception.class);
- Throwables.propagate(failure.cause());
+ Throwables.propagateIfPossible(failure.cause(), Exception.class);
+ throw new RuntimeException(failure.cause());
}
}
};
inOrder.verify(dataTree).prepare(any(DataTreeModification.class));
inOrder.verify(dataTree).commit(any(DataTreeCandidate.class));
+ // Purge request is scheduled as asynchronous, wait for two heartbeats to let it propagate into
+ // the journal
+ Thread.sleep(HEARTBEAT_MILLIS * 2);
+
shard.tell(Shard.GET_SHARD_MBEAN_MESSAGE, getRef());
final ShardStats shardStats = expectMsgClass(duration, ShardStats.class);
// Commit index should advance as we do not have an empty
// modification
- assertEquals(0, shardStats.getCommitIndex());
+ assertEquals(1, shardStats.getCommitIndex());
}
};
}
{
final Creator<Shard> creator = () -> new Shard(newShardBuilder()) {
@Override
- void persistPayload(final TransactionIdentifier transactionId, final Payload payload,
- boolean batchHint) {
+ void persistPayload(final Identifier id, final Payload payload,
+ final boolean batchHint) {
// Simulate an AbortTransaction message occurring during
// replication, after
// persisting and before finishing the commit to the
// in-memory store.
- doAbortTransaction(transactionId, null);
- super.persistPayload(transactionId, payload, batchHint);
+ doAbortTransaction(id, null);
+ super.persistPayload(id, payload, batchHint);
}
};
.shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(true).build();
final Props persistentProps = Shard.builder().id(shardID).datastoreContext(persistentContext)
- .schemaContext(SCHEMA_CONTEXT).props();
+ .schemaContextProvider(() -> SCHEMA_CONTEXT).props();
final DatastoreContext nonPersistentContext = DatastoreContext.newBuilder()
.shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(false).build();
final Props nonPersistentProps = Shard.builder().id(shardID).datastoreContext(nonPersistentContext)
- .schemaContext(SCHEMA_CONTEXT).props();
+ .schemaContextProvider(() -> SCHEMA_CONTEXT).props();
new ShardTestKit(getSystem()) {
{
waitUntilLeader(shard);
- final TestActorRef<MessageCollectorActor> listener =
- TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class));
+ final ActorRef listener = getSystem().actorOf(MessageCollectorActor.props());
shard.tell(new RegisterRoleChangeListener(), listener);
}
@Test
- public void testClusteredDataChangeListenerDelayedRegistration() throws Exception {
+ public void testClusteredDataChangeListenerWithDelayedRegistration() throws Exception {
new ShardTestKit(getSystem()) {
{
- final String testName = "testClusteredDataChangeListenerDelayedRegistration";
+ final String testName = "testClusteredDataChangeListenerWithDelayedRegistration";
dataStoreContextBuilder.shardElectionTimeoutFactor(1000)
.customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
shard.tell(new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.BASE, true),
getRef());
- final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
- RegisterChangeListenerReply.class);
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());
shard.tell(DatastoreContext.newBuilderFrom(dataStoreContextBuilder.build())
.datastoreContext(dataStoreContextBuilder.shardElectionTimeoutFactor(1000).build())
.peerAddresses(Collections.singletonMap(leaderShardID.toString(),
"akka://test/user/" + leaderShardID.toString()))
- .schemaContext(SCHEMA_CONTEXT).props()
+ .schemaContextProvider(() -> SCHEMA_CONTEXT).props()
.withDispatcher(Dispatchers.DefaultDispatcherId()), followerShardID.toString());
final TestActorRef<Shard> leaderShard = actorFactory
.createTestActor(Shard.builder().id(leaderShardID).datastoreContext(newDatastoreContext())
.peerAddresses(Collections.singletonMap(followerShardID.toString(),
"akka://test/user/" + followerShardID.toString()))
- .schemaContext(SCHEMA_CONTEXT).props()
+ .schemaContextProvider(() -> SCHEMA_CONTEXT).props()
.withDispatcher(Dispatchers.DefaultDispatcherId()), leaderShardID.toString());
leaderShard.tell(TimeoutNow.INSTANCE, ActorRef.noSender());
followerShard.tell(
new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.BASE, true),
getRef());
- final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
- RegisterChangeListenerReply.class);
- assertNotNull("getListenerRegistratioznPath", reply.getListenerRegistrationPath());
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
+ assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());
writeToStore(followerShard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
}
@Test
- public void testClusteredDataTreeChangeListenerDelayedRegistration() throws Exception {
+ public void testClusteredDataTreeChangeListenerWithDelayedRegistration() throws Exception {
new ShardTestKit(getSystem()) {
{
- final String testName = "testClusteredDataTreeChangeListenerDelayedRegistration";
+ final String testName = "testClusteredDataTreeChangeListenerWithDelayedRegistration";
dataStoreContextBuilder.shardElectionTimeoutFactor(1000)
.customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
waitUntilNoLeader(shard);
shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), getRef());
- final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
- RegisterDataTreeChangeListenerReply.class);
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());
shard.tell(DatastoreContext.newBuilderFrom(dataStoreContextBuilder.build())
};
}
+ @Test
+ public void testClusteredDataTreeChangeListenerWithDelayedRegistrationClosed() throws Exception {
+ new ShardTestKit(getSystem()) {
+ {
+ final String testName = "testClusteredDataTreeChangeListenerWithDelayedRegistrationClosed";
+ dataStoreContextBuilder.shardElectionTimeoutFactor(1000)
+ .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
+
+ final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(0);
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener,
+ TestModel.TEST_PATH), actorFactory.generateActorId(testName + "-DataTreeChangeListener"));
+
+ setupInMemorySnapshotStore();
+
+ final TestActorRef<Shard> shard = actorFactory.createTestActor(
+ newShardBuilder().props().withDispatcher(Dispatchers.DefaultDispatcherId()),
+ actorFactory.generateActorId(testName + "-shard"));
+
+ waitUntilNoLeader(shard);
+
+ shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), getRef());
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
+ assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());
+
+ final ActorSelection regActor = getSystem().actorSelection(reply.getListenerRegistrationPath());
+ regActor.tell(CloseDataTreeNotificationListenerRegistration.getInstance(), getRef());
+ expectMsgClass(CloseDataTreeNotificationListenerRegistrationReply.class);
+
+ shard.tell(DatastoreContext.newBuilderFrom(dataStoreContextBuilder.build())
+ .customRaftPolicyImplementation(null).build(), ActorRef.noSender());
+
+ listener.expectNoMoreChanges("Received unexpected change after close");
+ }
+ };
+ }
+
@Test
public void testClusteredDataTreeChangeListenerRegistration() throws Exception {
new ShardTestKit(getSystem()) {
.datastoreContext(dataStoreContextBuilder.shardElectionTimeoutFactor(1000).build())
.peerAddresses(Collections.singletonMap(leaderShardID.toString(),
"akka://test/user/" + leaderShardID.toString()))
- .schemaContext(SCHEMA_CONTEXT).props()
+ .schemaContextProvider(() -> SCHEMA_CONTEXT).props()
.withDispatcher(Dispatchers.DefaultDispatcherId()), followerShardID.toString());
final TestActorRef<Shard> leaderShard = actorFactory
.createTestActor(Shard.builder().id(leaderShardID).datastoreContext(newDatastoreContext())
.peerAddresses(Collections.singletonMap(followerShardID.toString(),
"akka://test/user/" + followerShardID.toString()))
- .schemaContext(SCHEMA_CONTEXT).props()
+ .schemaContextProvider(() -> SCHEMA_CONTEXT).props()
.withDispatcher(Dispatchers.DefaultDispatcherId()), leaderShardID.toString());
leaderShard.tell(TimeoutNow.INSTANCE, ActorRef.noSender());
actorFactory.generateActorId(testName + "-DataTreeChangeListener"));
followerShard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), getRef());
- final RegisterDataTreeChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
- RegisterDataTreeChangeListenerReply.class);
+ final RegisterDataTreeNotificationListenerReply reply = expectMsgClass(duration("5 seconds"),
+ RegisterDataTreeNotificationListenerReply.class);
assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());
writeToStore(followerShard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
@Test
public void testServerRemoved() throws Exception {
- final TestActorRef<MessageCollectorActor> parent = actorFactory.createTestActor(MessageCollectorActor.props());
+ final TestActorRef<MessageCollectorActor> parent = actorFactory.createTestActor(MessageCollectorActor.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()));
final ActorRef shard = parent.underlyingActor().context().actorOf(
newShardBuilder().props().withDispatcher(Dispatchers.DefaultDispatcherId()),
Code Review / controller.git / blobdiff