opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/DistributedEntityOwnershipIntegrationTest.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.controller.cluster.datastore.entityownership;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertNull;
  12 import static org.junit.Assert.assertThat;
  13 import static org.junit.Assert.assertTrue;
  14 import static org.mockito.AdditionalMatchers.or;
  15 import static org.mockito.Mockito.doNothing;
  16 import static org.mockito.Mockito.reset;
  17 import static org.mockito.Mockito.timeout;
  18 import static org.mockito.Mockito.verify;
  19 import static org.opendaylight.controller.cluster.datastore.MemberNode.verifyRaftState;
  20 import static org.opendaylight.controller.cluster.datastore.entityownership.AbstractEntityOwnershipTest.ownershipChange;
  21 import static org.opendaylight.controller.cluster.datastore.entityownership.DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME;
  22 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.CANDIDATE_NAME_NODE_ID;
  23 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.entityPath;
  24 import akka.actor.ActorRef;
  25 import akka.actor.Status.Failure;
  26 import akka.actor.Status.Success;
  27 import akka.cluster.Cluster;
  28 import akka.testkit.JavaTestKit;
  29 import com.google.common.base.Function;
  30 import com.google.common.base.Optional;
  31 import com.google.common.base.Stopwatch;
  32 import com.google.common.collect.Iterables;
  33 import com.google.common.util.concurrent.Uninterruptibles;
  34 import java.util.ArrayList;
  35 import java.util.Arrays;
  36 import java.util.Collection;
  37 import java.util.HashSet;
  38 import java.util.List;
  39 import java.util.concurrent.TimeUnit;
  40 import org.junit.After;
  41 import org.junit.Before;
  42 import org.junit.Test;
  43 import org.mockito.ArgumentCaptor;
  44 import org.mockito.Mock;
  45 import org.mockito.Mockito;
  46 import org.mockito.MockitoAnnotations;
  47 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  48 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
  49 import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
  50 import org.opendaylight.controller.cluster.datastore.MemberNode;
  51 import org.opendaylight.controller.cluster.datastore.MemberNode.RaftStateVerifier;
  52 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.EntityOwnerSelectionStrategyConfig;
  53 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
  54 import org.opendaylight.controller.cluster.raft.RaftState;
  55 import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
  56 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
  57 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  58 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  59 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  60 import org.opendaylight.controller.md.sal.common.api.clustering.CandidateAlreadyRegisteredException;
  61 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
  62 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipCandidateRegistration;
  63 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
  64 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
  65 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipService;
  66 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipState;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.clustering.entity.owners.rev150804.entity.owners.entity.type.entity.Candidate;
  68 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  69 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  70 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  71 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  72 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  73
  74 /**
  75  * End-to-end integration tests for the entity ownership functionality.
  76  *
  77  * @author Thomas Pantelis
  78  */
  79 public class DistributedEntityOwnershipIntegrationTest {
  80     private static final String MODULE_SHARDS_CONFIG = "module-shards-default.conf";
  81     private static final String MODULE_SHARDS_MEMBER_1_CONFIG = "module-shards-default-member-1.conf";
  82     private static final String ENTITY_TYPE1 = "entityType1";
  83     private static final String ENTITY_TYPE2 = "entityType2";
  84     private static final Entity ENTITY1 = new Entity(ENTITY_TYPE1, "entity1");
  85     private static final Entity ENTITY1_2 = new Entity(ENTITY_TYPE2, "entity1");
  86     private static final Entity ENTITY2 = new Entity(ENTITY_TYPE1, "entity2");
  87     private static final Entity ENTITY3 = new Entity(ENTITY_TYPE1, "entity3");
  88     private static final Entity ENTITY4 = new Entity(ENTITY_TYPE1, "entity4");
  89     private static final SchemaContext SCHEMA_CONTEXT = SchemaContextHelper.entityOwners();
  90
  91     private final DatastoreContext.Builder leaderDatastoreContextBuilder =
  92             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5).
  93                     shardIsolatedLeaderCheckIntervalInMillis(1000000);
  94
  95     private final DatastoreContext.Builder followerDatastoreContextBuilder =
  96             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10000);
  97
  98     private final List<MemberNode> memberNodes = new ArrayList<>();
  99
 100     @Mock
 101     private EntityOwnershipListener leaderMockListener;
 102
 103     @Mock
 104     private EntityOwnershipListener leaderMockListener2;
 105
 106     @Mock
 107     private EntityOwnershipListener follower1MockListener;
 108
 109     @Mock
 110     private EntityOwnershipListener follower2MockListener;
 111
 112     @Mock
 113     private EntityOwnershipListener member1Listener, member2Listener, member3Listener;
 114
 115     @Before
 116     public void setUp() {
 117         MockitoAnnotations.initMocks(this);
 118         InMemoryJournal.clear();
 119         InMemorySnapshotStore.clear();
 120     }
 121
 122     @After
 123     public void tearDown() {
 124         for(MemberNode m: memberNodes) {
 125             m.cleanup();
 126         }
 127     }
 128
 129     private DistributedEntityOwnershipService newOwnershipService(DistributedDataStore datastore) {
 130         DistributedEntityOwnershipService service = new DistributedEntityOwnershipService(datastore,
 131                 EntityOwnerSelectionStrategyConfig.newBuilder().build());
 132         service.start();
 133         return service;
 134     }
 135
 136     @Test
 137     public void testFunctionalityWithThreeNodes() throws Exception {
 138         String name = "test";
 139         MemberNode leaderNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
 140                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 141                 datastoreContextBuilder(leaderDatastoreContextBuilder).build();
 142
 143         MemberNode follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name ).
 144                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 145                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 146
 147         MemberNode follower2Node = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name ).
 148                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 149                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 150
 151         DistributedDataStore leaderDistributedDataStore = leaderNode.configDataStore();
 152
 153         leaderDistributedDataStore.waitTillReady();
 154         follower1Node.configDataStore().waitTillReady();
 155         follower2Node.configDataStore().waitTillReady();
 156
 157         EntityOwnershipService leaderEntityOwnershipService = newOwnershipService(leaderDistributedDataStore);
 158         EntityOwnershipService follower1EntityOwnershipService = newOwnershipService(follower1Node.configDataStore());
 159         EntityOwnershipService follower2EntityOwnershipService = newOwnershipService(follower2Node.configDataStore());
 160
 161         leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
 162
 163         leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
 164         leaderEntityOwnershipService.registerListener(ENTITY_TYPE2, leaderMockListener2);
 165         follower1EntityOwnershipService.registerListener(ENTITY_TYPE1, follower1MockListener);
 166
 167         // Register leader candidate for entity1 and verify it becomes owner
 168
 169         leaderEntityOwnershipService.registerCandidate(ENTITY1);
 170         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, true, true));
 171         verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));
 172         reset(leaderMockListener, follower1MockListener);
 173
 174         verifyGetOwnershipState(leaderEntityOwnershipService, ENTITY1, true, true);
 175         verifyGetOwnershipState(follower1EntityOwnershipService, ENTITY1, false, true);
 176
 177         // Register leader candidate for entity1_2 (same id, different type) and verify it becomes owner
 178
 179         leaderEntityOwnershipService.registerCandidate(ENTITY1_2);
 180         verify(leaderMockListener2, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1_2, false, true, true));
 181         verify(leaderMockListener, timeout(300).never()).ownershipChanged(ownershipChange(ENTITY1_2));
 182         reset(leaderMockListener2);
 183
 184         // Register follower1 candidate for entity1 and verify it gets added but doesn't become owner
 185
 186         follower1EntityOwnershipService.registerCandidate(ENTITY1);
 187         verifyCandidates(leaderDistributedDataStore, ENTITY1, "member-1", "member-2");
 188         verifyOwner(leaderDistributedDataStore, ENTITY1, "member-1");
 189         verify(leaderMockListener, timeout(300).never()).ownershipChanged(ownershipChange(ENTITY1));
 190         verify(follower1MockListener, timeout(300).never()).ownershipChanged(ownershipChange(ENTITY1));
 191
 192         // Register follower1 candidate for entity2 and verify it becomes owner
 193
 194         EntityOwnershipCandidateRegistration follower1Entity2Reg = follower1EntityOwnershipService.registerCandidate(ENTITY2);
 195         verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, true, true));
 196         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, false, true));
 197         reset(leaderMockListener, follower1MockListener);
 198
 199         // Register follower2 candidate for entity2 and verify it gets added but doesn't become owner
 200
 201         follower2EntityOwnershipService.registerListener(ENTITY_TYPE1, follower2MockListener);
 202         verify(follower2MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, false, true));
 203         verify(follower2MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));
 204
 205         follower2EntityOwnershipService.registerCandidate(ENTITY2);
 206         verifyCandidates(leaderDistributedDataStore, ENTITY2, "member-2", "member-3");
 207         verifyOwner(leaderDistributedDataStore, ENTITY2, "member-2");
 208
 209         // Unregister follower1 candidate for entity2 and verify follower2 becomes owner
 210
 211         follower1Entity2Reg.close();
 212         verifyCandidates(leaderDistributedDataStore, ENTITY2, "member-3");
 213         verifyOwner(leaderDistributedDataStore, ENTITY2, "member-3");
 214         verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, true, false, true));
 215         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, false, true));
 216
 217         // Depending on timing, follower2MockListener could get ownershipChanged with "false, false, true" if
 218         // if the original ownership change with "member-2 is replicated to follower2 after the listener is
 219         // registered.
 220         Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 221         verify(follower2MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, true, true));
 222
 223         // Register follower1 candidate for entity3 and verify it becomes owner
 224
 225         follower1EntityOwnershipService.registerCandidate(ENTITY3);
 226         verifyOwner(leaderDistributedDataStore, ENTITY3, "member-2");
 227         verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY3, false, true, true));
 228         verify(follower2MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY3, false, false, true));
 229         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY3, false, false, true));
 230
 231         // Register follower2 candidate for entity4 and verify it becomes owner
 232
 233         follower2EntityOwnershipService.registerCandidate(ENTITY4);
 234         verifyOwner(leaderDistributedDataStore, ENTITY4, "member-3");
 235         verify(follower2MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY4, false, true, true));
 236         verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY4, false, false, true));
 237         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY4, false, false, true));
 238         reset(follower1MockListener, follower2MockListener);
 239
 240         // Register follower1 candidate for entity4 and verify it gets added but doesn't become owner
 241
 242         follower1EntityOwnershipService.registerCandidate(ENTITY4);
 243         verifyCandidates(leaderDistributedDataStore, ENTITY4, "member-3", "member-2");
 244         verifyOwner(leaderDistributedDataStore, ENTITY4, "member-3");
 245
 246         // Shutdown follower2 and verify it's owned entities (entity 2 & 4) get re-assigned
 247
 248         reset(leaderMockListener, follower1MockListener);
 249         follower2Node.cleanup();
 250
 251         verify(follower1MockListener, timeout(15000).times(2)).ownershipChanged(or(ownershipChange(ENTITY4, false, true, true),
 252                 ownershipChange(ENTITY2, false, false, false)));
 253         verify(leaderMockListener, timeout(15000).times(2)).ownershipChanged(or(ownershipChange(ENTITY4, false, false, true),
 254                 ownershipChange(ENTITY2, false, false, false)));
 255         verifyOwner(leaderDistributedDataStore, ENTITY2, ""); // no other candidate
 256
 257         // Register leader candidate for entity2 and verify it becomes owner
 258
 259         EntityOwnershipCandidateRegistration leaderEntity2Reg = leaderEntityOwnershipService.registerCandidate(ENTITY2);
 260         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, true, true));
 261         verifyOwner(leaderDistributedDataStore, ENTITY2, "member-1");
 262
 263         // Unregister leader candidate for entity2 and verify the owner is cleared
 264
 265         leaderEntity2Reg.close();
 266         verifyOwner(leaderDistributedDataStore, ENTITY2, "");
 267         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, true, false, false));
 268         verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, false, false));
 269     }
 270
 271     @Test
 272     public void testLeaderCandidatesRemovedAfterShutdown() throws Exception {
 273         followerDatastoreContextBuilder.shardElectionTimeoutFactor(5).
 274                     customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
 275
 276         String name = "test";
 277         MemberNode leaderNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
 278                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 279                 datastoreContextBuilder(leaderDatastoreContextBuilder).build();
 280
 281         MemberNode follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name ).
 282                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 283                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 284
 285         MemberNode follower2Node = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name ).
 286                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 287                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 288
 289         DistributedDataStore leaderDistributedDataStore = leaderNode.configDataStore();
 290
 291         leaderDistributedDataStore.waitTillReady();
 292         follower1Node.configDataStore().waitTillReady();
 293         follower2Node.configDataStore().waitTillReady();
 294
 295         EntityOwnershipService leaderEntityOwnershipService = newOwnershipService(leaderDistributedDataStore);
 296         EntityOwnershipService follower1EntityOwnershipService = newOwnershipService(follower1Node.configDataStore());
 297         EntityOwnershipService follower2EntityOwnershipService = newOwnershipService(follower2Node.configDataStore());
 298
 299         leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
 300
 301         // Register follower1 candidate for entity1 and verify it becomes owner
 302
 303         follower1EntityOwnershipService.registerCandidate(ENTITY1);
 304         verifyOwner(leaderDistributedDataStore, ENTITY1, "member-2");
 305
 306         // Register leader candidate for entity1
 307
 308         leaderEntityOwnershipService.registerCandidate(ENTITY1);
 309         verifyCandidates(leaderDistributedDataStore, ENTITY1, "member-2", "member-1");
 310         verifyOwner(leaderDistributedDataStore, ENTITY1, "member-2");
 311
 312         // Register leader candidate for entity2 and verify it becomes owner
 313
 314         leaderEntityOwnershipService.registerCandidate(ENTITY2);
 315         verifyOwner(leaderDistributedDataStore, ENTITY2, "member-1");
 316
 317         // Register follower2 candidate for entity2
 318
 319         follower2EntityOwnershipService.registerCandidate(ENTITY2);
 320         verifyCandidates(leaderDistributedDataStore, ENTITY2, "member-1", "member-3");
 321         verifyOwner(leaderDistributedDataStore, ENTITY2, "member-1");
 322
 323         // Shutdown the leader and verify its removed from the candidate list
 324
 325         leaderNode.cleanup();
 326         follower1Node.waitForMemberDown("member-1");
 327
 328         // Re-enable elections on folower1 so it becomes the leader
 329
 330         ActorRef follower1Shard = IntegrationTestKit.findLocalShard(follower1Node.configDataStore().
 331                 getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
 332         follower1Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build()).
 333                 customRaftPolicyImplementation(null).build(), ActorRef.noSender());
 334
 335         MemberNode.verifyRaftState(follower1Node.configDataStore(), ENTITY_OWNERSHIP_SHARD_NAME, new RaftStateVerifier() {
 336             @Override
 337             public void verify(OnDemandRaftState raftState) {
 338                 assertEquals("Raft state", RaftState.Leader.toString(), raftState.getRaftState());
 339             }
 340         });
 341
 342         // Verify the prior leader's candidates are removed
 343
 344         verifyCandidates(follower1Node.configDataStore(), ENTITY1, "member-2");
 345         verifyCandidates(follower1Node.configDataStore(), ENTITY2, "member-3");
 346         verifyOwner(follower1Node.configDataStore(), ENTITY1, "member-2");
 347         verifyOwner(follower1Node.configDataStore(), ENTITY2, "member-3");
 348     }
 349
 350     /**
 351      * Tests if candidate(s) get unregistered for restarted nodes, in case majority of nodes restart including the node
 352      * that was leader for entity-ownership shard.
 353      * Tests for the bug <a href="https://bugs.opendaylight.org/show_bug.cgi?id=5613>5613</a>
 354      * @throws Exception
 355      */
 356     @Test
 357     public void testCandidatesRemovedOnRestartOfMajorityNodes() throws Exception {
 358         followerDatastoreContextBuilder.shardElectionTimeoutFactor(5).
 359                 customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
 360
 361         String name = "test";
 362         MemberNode member1Node = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name).
 363                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 364                 datastoreContextBuilder(leaderDatastoreContextBuilder).build();
 365
 366         MemberNode member2Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
 367                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 368                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 369
 370         MemberNode member3Node = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
 371                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 372                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 373
 374         DistributedDataStore member1DistributedDataStore = member1Node.configDataStore();
 375
 376         member1DistributedDataStore.waitTillReady();
 377         member2Node.configDataStore().waitTillReady();
 378         member3Node.configDataStore().waitTillReady();
 379
 380         EntityOwnershipService member1EntityOwnershipService = newOwnershipService(member1DistributedDataStore);
 381         EntityOwnershipService member2EntityOwnershipService = newOwnershipService(member2Node.configDataStore());
 382         EntityOwnershipService member3EntityOwnershipService = newOwnershipService(member3Node.configDataStore());
 383         member1EntityOwnershipService.registerListener(ENTITY_TYPE1, member1Listener);
 384         member2EntityOwnershipService.registerListener(ENTITY_TYPE1, member2Listener);
 385         member3EntityOwnershipService.registerListener(ENTITY_TYPE1, member3Listener);
 386
 387
 388         member1Node.kit().waitUntilLeader(member1Node.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
 389
 390         // Register follower1 candidate for entity1 and verify it becomes owner
 391
 392         member2EntityOwnershipService.registerCandidate(ENTITY1);
 393         verifyOwner(member1DistributedDataStore, ENTITY1, "member-2");
 394         verify(member2Listener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, true, true));
 395
 396
 397         // Register leader candidate for entity1
 398
 399         member1EntityOwnershipService.registerCandidate(ENTITY1);
 400         member3EntityOwnershipService.registerCandidate(ENTITY1);
 401         verifyCandidates(member1DistributedDataStore, ENTITY1, "member-2", "member-1", "member-3");
 402         verifyOwner(member1DistributedDataStore, ENTITY1, "member-2");
 403         verify(member2Listener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, true, true));
 404         verify(member1Listener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));
 405         verify(member3Listener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));
 406
 407         // Stop shard leader and a owner of ENTITY1 (majority of nodes)
 408         member1Node.cleanup();
 409         member2Node.cleanup();
 410
 411         InMemoryJournal.get("member-1-shard-entity-ownership-config_test").clear();
 412         InMemoryJournal.get("member-2-shard-entity-ownership-config_test").clear();
 413
 414
 415
 416         // wait for peer downs
 417         member3Node.waitForMemberDown("member-1");
 418         member3Node.waitForMemberDown("member-2");
 419
 420         // this is needed as the peer-up is not received in this integration test framework
 421         member3Node.cleanup();
 422         Mockito.reset(member3Listener);
 423
 424         member3Node = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name).
 425                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 426                 datastoreContextBuilder(leaderDatastoreContextBuilder).build();
 427         member3EntityOwnershipService = newOwnershipService(member3Node.configDataStore());
 428         member3EntityOwnershipService.registerListener(ENTITY_TYPE1, member3Listener);
 429         member3EntityOwnershipService.registerCandidate(ENTITY1);
 430
 431         // re-start shard leader and owner of ENTITY1
 432         member1Node = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name).
 433                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 434                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 435
 436         member2Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name).
 437                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 438                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 439
 440         member1DistributedDataStore = member1Node.configDataStore();
 441         member1EntityOwnershipService = newOwnershipService(member1DistributedDataStore);
 442         member2EntityOwnershipService = newOwnershipService(member2Node.configDataStore());
 443         member1EntityOwnershipService.registerListener(ENTITY_TYPE1, member1Listener);
 444         member2EntityOwnershipService.registerListener(ENTITY_TYPE1, member2Listener);
 445
 446         // wait for peers to come up
 447         member3Node.waitForMembersUp("member-1", "member-2");
 448         member1Node.waitForMembersUp("member-3");
 449
 450         member3Node.kit().waitUntilLeader(member3Node.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
 451
 452         // wait for new leader to consume all modifications
 453         Uninterruptibles.sleepUninterruptibly(5000, TimeUnit.MILLISECONDS);
 454         // confirm that only member-3 is the candidate now (as 1 and 2 are restarted)
 455         verifyCandidates(member3Node.configDataStore(), ENTITY1, "member-3");
 456         Uninterruptibles.sleepUninterruptibly(5000, TimeUnit.MILLISECONDS);
 457
 458         // reset the listeners before registering for candidates again
 459         Mockito.reset(member2Listener);
 460         Mockito.reset(member1Listener);
 461         member1EntityOwnershipService.registerCandidate(ENTITY1);
 462         member2EntityOwnershipService.registerCandidate(ENTITY1);
 463         // confirm that all are candidates again
 464         verifyCandidates(member3Node.configDataStore(), ENTITY1, "member-3", "member-1", "member-2");
 465         // confirm that 1 and 2 get event after they register for candidates again
 466         verify(member2Listener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1));
 467         verify(member1Listener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1));
 468     }
 469
 470     /**
 471      * Reproduces bug <a href="https://bugs.opendaylight.org/show_bug.cgi?id=4554">4554</a>
 472      *
 473      * @throws CandidateAlreadyRegisteredException
 474      */
 475
 476     public void testCloseCandidateRegistrationInQuickSuccession() throws CandidateAlreadyRegisteredException {
 477         String name = "testCloseCandidateRegistrationInQuickSuccession";
 478         MemberNode leaderNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
 479                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 480                 datastoreContextBuilder(leaderDatastoreContextBuilder).build();
 481
 482         MemberNode follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name ).
 483                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 484                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 485
 486         MemberNode follower2Node = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name ).
 487                 moduleShardsConfig(MODULE_SHARDS_CONFIG).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 488                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 489
 490         DistributedDataStore leaderDistributedDataStore = leaderNode.configDataStore();
 491
 492         leaderDistributedDataStore.waitTillReady();
 493         follower1Node.configDataStore().waitTillReady();
 494         follower2Node.configDataStore().waitTillReady();
 495
 496         EntityOwnershipService leaderEntityOwnershipService = newOwnershipService(leaderDistributedDataStore);
 497         EntityOwnershipService follower1EntityOwnershipService = newOwnershipService(follower1Node.configDataStore());
 498         EntityOwnershipService follower2EntityOwnershipService = newOwnershipService(follower2Node.configDataStore());
 499
 500         leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
 501
 502         leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
 503         follower1EntityOwnershipService.registerListener(ENTITY_TYPE1, follower1MockListener);
 504         follower2EntityOwnershipService.registerListener(ENTITY_TYPE1, follower2MockListener);
 505
 506         final EntityOwnershipCandidateRegistration candidate1 = leaderEntityOwnershipService.registerCandidate(ENTITY1);
 507         final EntityOwnershipCandidateRegistration candidate2 = follower1EntityOwnershipService.registerCandidate(ENTITY1);
 508         final EntityOwnershipCandidateRegistration candidate3 = follower2EntityOwnershipService.registerCandidate(ENTITY1);
 509
 510         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, true, true));
 511
 512         Mockito.reset(leaderMockListener);
 513
 514         ArgumentCaptor<EntityOwnershipChange> leaderChangeCaptor = ArgumentCaptor.forClass(EntityOwnershipChange.class);
 515         ArgumentCaptor<EntityOwnershipChange> follower1ChangeCaptor = ArgumentCaptor.forClass(EntityOwnershipChange.class);
 516         ArgumentCaptor<EntityOwnershipChange> follower2ChangeCaptor = ArgumentCaptor.forClass(EntityOwnershipChange.class);
 517         doNothing().when(leaderMockListener).ownershipChanged(leaderChangeCaptor.capture());
 518         doNothing().when(follower1MockListener).ownershipChanged(follower1ChangeCaptor.capture());
 519         doNothing().when(follower2MockListener).ownershipChanged(follower2ChangeCaptor.capture());
 520
 521         candidate1.close();
 522         candidate2.close();
 523         candidate3.close();
 524
 525         boolean passed = false;
 526         for(int i=0;i<100;i++) {
 527             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 528             if(!leaderEntityOwnershipService.getOwnershipState(ENTITY1).isPresent() ||
 529                     !leaderEntityOwnershipService.getOwnershipState(ENTITY1).get().hasOwner() &&
 530                     !follower1EntityOwnershipService.getOwnershipState(ENTITY1).get().hasOwner() &&
 531                     !follower2EntityOwnershipService.getOwnershipState(ENTITY1).get().hasOwner() &&
 532                     leaderChangeCaptor.getAllValues().size() > 0 && !leaderChangeCaptor.getValue().hasOwner() &&
 533                     leaderChangeCaptor.getAllValues().size() > 0 && !follower1ChangeCaptor.getValue().hasOwner() &&
 534                     leaderChangeCaptor.getAllValues().size() > 0 && !follower2ChangeCaptor.getValue().hasOwner()) {
 535                 passed = true;
 536                 break;
 537             }
 538         }
 539
 540         assertTrue("No ownership change message was sent with hasOwner=false", passed);
 541     }
 542
 543     /**
 544      * Tests bootstrapping the entity-ownership shard when there's no shards initially configured for local
 545      * member. The entity-ownership shard is initially created as inactive (ie remains a follower), requiring
 546      * an AddShardReplica request to join it to an existing leader.
 547      */
 548     @Test
 549     public void testEntityOwnershipShardBootstrapping() throws Throwable {
 550         String name = "testEntityOwnershipShardBootstrapping";
 551         String moduleShardsConfig = MODULE_SHARDS_MEMBER_1_CONFIG;
 552         MemberNode leaderNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name ).
 553                 moduleShardsConfig(moduleShardsConfig).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 554                 datastoreContextBuilder(leaderDatastoreContextBuilder).build();
 555
 556         DistributedDataStore leaderDistributedDataStore = leaderNode.configDataStore();
 557         EntityOwnershipService leaderEntityOwnershipService = newOwnershipService(leaderDistributedDataStore);
 558
 559         leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
 560
 561         MemberNode follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name ).
 562                 moduleShardsConfig(moduleShardsConfig).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 563                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 564
 565         DistributedDataStore follower1DistributedDataStore = follower1Node.configDataStore();
 566         follower1DistributedDataStore.waitTillReady();
 567
 568         leaderNode.waitForMembersUp("member-2");
 569         follower1Node.waitForMembersUp("member-1");
 570
 571         EntityOwnershipService follower1EntityOwnershipService = newOwnershipService(follower1DistributedDataStore);
 572
 573         leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
 574
 575         // Register a candidate for follower1 - should get queued since follower1 has no leader
 576         EntityOwnershipCandidateRegistration candidateReg = follower1EntityOwnershipService.registerCandidate(ENTITY1);
 577         verify(leaderMockListener, timeout(300).never()).ownershipChanged(ownershipChange(ENTITY1));
 578
 579         // Add replica in follower1
 580         AddShardReplica addReplica = new AddShardReplica(ENTITY_OWNERSHIP_SHARD_NAME);
 581         follower1DistributedDataStore.getActorContext().getShardManager().tell(addReplica , follower1Node.kit().getRef());
 582         Object reply = follower1Node.kit().expectMsgAnyClassOf(JavaTestKit.duration("5 sec"), Success.class, Failure.class);
 583         if(reply instanceof Failure) {
 584             throw new AssertionError("AddShardReplica failed", ((Failure)reply).cause());
 585         }
 586
 587         // The queued candidate registration should proceed
 588         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));
 589         reset(leaderMockListener);
 590
 591         candidateReg.close();
 592         verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, false));
 593         reset(leaderMockListener);
 594
 595         // Restart follower1 and verify the entity ownership shard is re-instated by registering.
 596         Cluster.get(leaderNode.kit().getSystem()).down(Cluster.get(follower1Node.kit().getSystem()).selfAddress());
 597         follower1Node.cleanup();
 598
 599         follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name ).
 600                 moduleShardsConfig(moduleShardsConfig).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false).
 601                 datastoreContextBuilder(followerDatastoreContextBuilder).build();
 602         follower1EntityOwnershipService = newOwnershipService(follower1Node.configDataStore());
 603
 604         follower1EntityOwnershipService.registerCandidate(ENTITY1);
 605         verify(leaderMockListener, timeout(20000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));
 606
 607         verifyRaftState(follower1Node.configDataStore(), ENTITY_OWNERSHIP_SHARD_NAME, new RaftStateVerifier() {
 608             @Override
 609             public void verify(OnDemandRaftState raftState) {
 610                 assertNull("Custom RaftPolicy class name", raftState.getCustomRaftPolicyClassName());
 611                 assertEquals("Peer count", 1, raftState.getPeerAddresses().keySet().size());
 612                 assertThat("Peer Id", Iterables.<String>getLast(raftState.getPeerAddresses().keySet()),
 613                         org.hamcrest.CoreMatchers.containsString("member-1"));
 614             }
 615         });
 616     }
 617
 618     private static void verifyGetOwnershipState(EntityOwnershipService service, Entity entity,
 619             boolean isOwner, boolean hasOwner) {
 620         Optional<EntityOwnershipState> state = service.getOwnershipState(entity);
 621         assertEquals("getOwnershipState present", true, state.isPresent());
 622         assertEquals("isOwner", isOwner, state.get().isOwner());
 623         assertEquals("hasOwner", hasOwner, state.get().hasOwner());
 624     }
 625
 626     private static void verifyCandidates(DistributedDataStore dataStore, Entity entity, String... expCandidates) throws Exception {
 627         AssertionError lastError = null;
 628         Stopwatch sw = Stopwatch.createStarted();
 629         while(sw.elapsed(TimeUnit.MILLISECONDS) <= 10000) {
 630             Optional<NormalizedNode<?, ?>> possible = dataStore.newReadOnlyTransaction().read(
 631                     entityPath(entity.getType(), entity.getId()).node(Candidate.QNAME)).get(5, TimeUnit.SECONDS);
 632             try {
 633                 assertEquals("Candidates not found for " + entity, true, possible.isPresent());
 634                 Collection<String> actual = new HashSet<>();
 635                 for(MapEntryNode candidate: ((MapNode)possible.get()).getValue()) {
 636                     actual.add(candidate.getChild(CANDIDATE_NAME_NODE_ID).get().getValue().toString());
 637                 }
 638
 639                 assertEquals("Candidates for " + entity, new HashSet(Arrays.asList(expCandidates)), actual);
 640                 return;
 641             } catch (AssertionError e) {
 642                 lastError = e;
 643                 Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
 644             }
 645         }
 646
 647         throw lastError;
 648     }
 649
 650     private static void verifyOwner(final DistributedDataStore dataStore, Entity entity, String expOwner) {
 651         AbstractEntityOwnershipTest.verifyOwner(expOwner, entity.getType(), entity.getId(),
 652                 new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
 653                     @Override
 654                     public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
 655                         try {
 656                             return dataStore.newReadOnlyTransaction().read(path).get(5, TimeUnit.SECONDS).get();
 657                         } catch (Exception e) {
 658                             return null;
 659                         }
 660                     }
 661                 });
 662     }
 663 }