import akka.cluster.Cluster; //导入依赖的package包/类
public void waitForMembersUp(final String... otherMembers) {
    Set<String> otherMembersSet = Sets.newHashSet(otherMembers);
    Stopwatch sw = Stopwatch.createStarted();
    while (sw.elapsed(TimeUnit.SECONDS) <= 10) {
        CurrentClusterState state = Cluster.get(getSystem()).state();
        for (Member m: state.getMembers()) {
            if (m.status() == MemberStatus.up() && otherMembersSet.remove(m.getRoles().iterator().next())
                    && otherMembersSet.isEmpty()) {
                return;
            }
        }

        Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
    }

    fail("Member(s) " + otherMembersSet + " are not Up");
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Before
public void setUp() {
    InMemoryJournal.clear();
    InMemorySnapshotStore.clear();

    leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
    Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);

    followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
    Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);

    leaderSystemProvider = Mockito.mock(ActorSystemProvider.class);
    doReturn(leaderSystem).when(leaderSystemProvider).getActorSystem();

    followerSystemProvider = Mockito.mock(ActorSystemProvider.class);
    doReturn(followerSystem).when(followerSystemProvider).getActorSystem();

}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Override
public void preStart() {
    ActorRefProvider provider = getContext().provider();
    selfAddress = provider.getDefaultAddress();

    bucketStore = new BucketStoreAccess(getContext(), config.getAskDuration());

    if (provider instanceof ClusterActorRefProvider) {
        cluster = Cluster.get(getContext().system());
        cluster.subscribe(getSelf(),
                ClusterEvent.initialStateAsEvents(),
                ClusterEvent.MemberEvent.class,
                ClusterEvent.ReachableMember.class,
                ClusterEvent.UnreachableMember.class);
    }

    if (autoStartGossipTicks) {
        gossipTask = getContext().system().scheduler().schedule(
                new FiniteDuration(1, TimeUnit.SECONDS),        //initial delay
                config.getGossipTickInterval(),                 //interval
                getSelf(),                                      //target
                GOSSIP_TICK,                                    //message
                getContext().dispatcher(),                      //execution context
                getSelf()                                       //sender
        );
    }
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@BeforeClass
public static void staticSetup() throws InterruptedException {
    AkkaConfigurationReader reader = ConfigFactory::load;

    RemoteRpcProviderConfig config1 = new RemoteRpcProviderConfig.Builder("memberA").gossipTickInterval("200ms")
            .withConfigReader(reader).build();
    RemoteRpcProviderConfig config2 = new RemoteRpcProviderConfig.Builder("memberB").gossipTickInterval("200ms")
            .withConfigReader(reader).build();
    RemoteRpcProviderConfig config3 = new RemoteRpcProviderConfig.Builder("memberC").gossipTickInterval("200ms")
            .withConfigReader(reader).build();
    node1 = ActorSystem.create("opendaylight-rpc", config1.get());
    node2 = ActorSystem.create("opendaylight-rpc", config2.get());
    node3 = ActorSystem.create("opendaylight-rpc", config3.get());

    waitForMembersUp(node1, Cluster.get(node2).selfUniqueAddress(), Cluster.get(node3).selfUniqueAddress());
    waitForMembersUp(node2, Cluster.get(node1).selfUniqueAddress(), Cluster.get(node3).selfUniqueAddress());
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
static void waitForMembersUp(final ActorSystem node, final UniqueAddress... addresses) {
    Set<UniqueAddress> otherMembersSet = Sets.newHashSet(addresses);
    Stopwatch sw = Stopwatch.createStarted();
    while (sw.elapsed(TimeUnit.SECONDS) <= 10) {
        CurrentClusterState state = Cluster.get(node).state();
        for (Member m : state.getMembers()) {
            if (m.status() == MemberStatus.up() && otherMembersSet.remove(m.uniqueAddress())
                    && otherMembersSet.isEmpty()) {
                return;
            }
        }

        Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
    }

    fail("Member(s) " + otherMembersSet + " are not Up");
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
public static void main(String[] args) throws IOException {
  ActorSystem actorSystem = ActorSystem.create(CLUSTER_NAME);
  actorSystem.actorOf(SimpleClusterListener.props());
  final ActorMaterializer materializer = ActorMaterializer.create(actorSystem);

  Cluster cluster = Cluster.get(actorSystem);
  List<Address> addresses = Arrays.asList(System.getenv().get("SEED_NODES").split(","))
      .stream()
      .map(ip -> new Address("akka.tcp", CLUSTER_NAME, ip, 2551))
      .collect(Collectors.toList());
  cluster.joinSeedNodes(addresses);
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Override
@SuppressWarnings("checkstyle:IllegalCatch")
public Optional<Cluster> getCluster() {
    if (cluster == null) {
        try {
            cluster = Optional.of(Cluster.get(getActorSystem()));
        } catch (Exception e) {
            // An exception means there's no cluster configured. This will only happen in unit tests.
            log.debug("{}: Could not obtain Cluster: {}", getId(), e);
            cluster = Optional.empty();
        }
    }

    return cluster;
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
public ClusterWrapperImpl(ActorSystem actorSystem) {
    Preconditions.checkNotNull(actorSystem, "actorSystem should not be null");

    cluster = Cluster.get(actorSystem);

    Preconditions.checkState(cluster.getSelfRoles().size() > 0,
        "No akka roles were specified.\n"
        + "One way to specify the member name is to pass a property on the command line like so\n"
        + "   -Dakka.cluster.roles.0=member-3\n"
        + "member-3 here would be the name of the member");

    currentMemberName = MemberName.forName(cluster.getSelfRoles().iterator().next());
    selfAddress = cluster.selfAddress();
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Before
public void setUp() throws IOException {
    InMemorySnapshotStore.clear();
    InMemoryJournal.clear();
    system = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
    Address member1Address = AddressFromURIString.parse("akka://[email protected]:2558");
    Cluster.get(system).join(member1Address);
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Before
public void setUp() {
    InMemoryJournal.clear();
    InMemorySnapshotStore.clear();

    leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
    Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);

    followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
    Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);

    follower2System = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member3"));
    Cluster.get(follower2System).join(MEMBER_1_ADDRESS);
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Test
public void testTransactionWithCreateTxFailureDueToNoLeader() throws Exception {
    followerDatastoreContextBuilder.frontendRequestTimeoutInSeconds(2);
    initDatastoresWithCars("testTransactionWithCreateTxFailureDueToNoLeader");

    // Do an initial read to get the primary shard info cached.

    final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
    readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS);

    // Shutdown the leader and try to create a new tx.

    JavaTestKit.shutdownActorSystem(leaderSystem, null, true);

    Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);

    Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);

    sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
            .operationTimeoutInMillis(10).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));

    final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();

    rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());

    try {
        followerTestKit.doCommit(rwTx.ready());
        fail("Exception expected");
    } catch (final ExecutionException e) {
        final String msg = "Unexpected exception: " + Throwables.getStackTraceAsString(e.getCause());
        if (DistributedDataStore.class.equals(testParameter)) {
            assertTrue(msg, Throwables.getRootCause(e) instanceof NoShardLeaderException);
        } else {
            assertTrue(msg, Throwables.getRootCause(e) instanceof RequestTimeoutException);
        }
    }
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Test
public void testTransactionRetryWithInitialAskTimeoutExOnCreateTx() throws Exception {
    followerDatastoreContextBuilder.backendAlivenessTimerIntervalInSeconds(2);
    String testName = "testTransactionRetryWithInitialAskTimeoutExOnCreateTx";
    initDatastores(testName, MODULE_SHARDS_CARS_1_2_3, CARS);

    final DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder()
            .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10);
    final IntegrationTestKit follower2TestKit = new IntegrationTestKit(
            follower2System, follower2DatastoreContextBuilder, commitTimeout);

    try (AbstractDataStore ds =
            follower2TestKit.setupAbstractDataStore(
                    testParameter, testName, MODULE_SHARDS_CARS_1_2_3, false, CARS)) {

        followerTestKit.waitForMembersUp("member-1", "member-3");
        follower2TestKit.waitForMembersUp("member-1", "member-2");

        // Do an initial read to get the primary shard info cached.

        final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
        readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS);

        // Shutdown the leader and try to create a new tx.

        JavaTestKit.shutdownActorSystem(leaderSystem, null, true);

        Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);

        sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
            .operationTimeoutInMillis(500).shardElectionTimeoutFactor(5).customRaftPolicyImplementation(null));

        final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();

        rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());

        followerTestKit.doCommit(rwTx.ready());
    }
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Before
public void setUp() {
    MockitoAnnotations.initMocks(this);

    InMemoryJournal.clear();
    InMemorySnapshotStore.clear();

    leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
    Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);

    leaderSystemProvider = Mockito.mock(ActorSystemProvider.class);
    doReturn(leaderSystem).when(leaderSystemProvider).getActorSystem();
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Override
public String persistenceId() {
    for (String role : JavaConversions.asJavaIterable((Cluster.get(getContext().system()).selfRoles()))) {
        if (role.startsWith("backend-")) {
            return role + "-master";
        }
    }
    return "master";

}
 

import akka.cluster.Cluster; //导入依赖的package包/类
/**
 * Api to make a node leave the cluster.
 * @param host hostname/ip of the node.
 * @param port port
 */
@POST
@Path("/cluster/leave")
public Response leaveCluster(@QueryParam("host") String host, @QueryParam("port") Integer port) {
    if(StringUtils.isEmpty(host) || port == null) {
        return Response.status(Response.Status.BAD_REQUEST.getStatusCode()).entity("empty hostname or port").build();
    }

    Address akkAddress = new Address("akka.tcp", actorSystemManager.retrieveActorSystem().name(), host, port);
    Cluster cluster = Cluster.get(actorSystemManager.retrieveActorSystem());
    cluster.leave(akkAddress);

    return Response.status(Response.Status.OK.getStatusCode()).build();
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Override
public String persistenceId() {
  for (String role : JavaConversions.asJavaIterable((Cluster.get(getContext().system()).selfRoles()))) {
    if (role.startsWith("backend-")) {
      return role + "-master";
    }
  }
  return "master";

}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@Override
public void leaveCluster(){

    Cluster cluster = Cluster.get(clusterSystem);
    logger.info("Downing self: {} from cluster: {}", cluster.selfAddress(), clusterSystem.name());
    cluster.leave(cluster.selfAddress());
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
@BeforeClass
public static void setUpClass() throws IOException {
    system = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
    final Address member1Address = AddressFromURIString.parse("akka://[email protected]:2558");
    Cluster.get(system).join(member1Address);
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
public MemberNode build() throws Exception {
    Preconditions.checkNotNull(moduleShardsConfig, "moduleShardsConfig must be specified");
    Preconditions.checkNotNull(akkaConfig, "akkaConfig must be specified");
    Preconditions.checkNotNull(testName, "testName must be specified");

    if (schemaContext == null) {
        schemaContext = SchemaContextHelper.full();
    }

    MemberNode node = new MemberNode();
    node.datastoreContextBuilder = datastoreContextBuilder;

    Config baseConfig = ConfigFactory.load();
    Config config;
    if (useAkkaArtery) {
        config = baseConfig.getConfig(akkaConfig);
    } else {
        config = baseConfig.getConfig(akkaConfig + "-without-artery")
                .withFallback(baseConfig.getConfig(akkaConfig));
    }

    ActorSystem system = ActorSystem.create("cluster-test", config);
    String member1Address = useAkkaArtery ? MEMBER_1_ADDRESS : MEMBER_1_ADDRESS.replace("akka", "akka.tcp");
    Cluster.get(system).join(AddressFromURIString.parse(member1Address));

    node.kit = new IntegrationTestKit(system, datastoreContextBuilder);

    String memberName = new ClusterWrapperImpl(system).getCurrentMemberName().getName();
    node.kit.getDatastoreContextBuilder().shardManagerPersistenceId("shard-manager-config-" + memberName);
    node.configDataStore = node.kit.setupAbstractDataStore(DistributedDataStore.class,
            "config_" + testName, moduleShardsConfig, true, schemaContext, waitForshardLeader);

    if (createOperDatastore) {
        node.kit.getDatastoreContextBuilder().shardManagerPersistenceId("shard-manager-oper-" + memberName);
        node.operDataStore = node.kit.setupAbstractDataStore(DistributedDataStore.class,
                "oper_" + testName, moduleShardsConfig, true, schemaContext, waitForshardLeader);
    }

    members.add(node);
    return node;
}
 

import akka.cluster.Cluster; //导入依赖的package包/类
/**
 * Tests bootstrapping the entity-ownership shard when there's no shards initially configured for local
 * member. The entity-ownership shard is initially created as inactive (ie remains a follower), requiring
 * an AddShardReplica request to join it to an existing leader.
 */
@Test
public void testEntityOwnershipShardBootstrapping() throws Exception {
    String name = "testEntityOwnershipShardBootstrapping";
    String moduleShardsConfig = MODULE_SHARDS_MEMBER_1_CONFIG;
    MemberNode leaderNode = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name)
            .moduleShardsConfig(moduleShardsConfig).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false)
            .datastoreContextBuilder(leaderDatastoreContextBuilder).build();

    AbstractDataStore leaderDistributedDataStore = leaderNode.configDataStore();
    final DOMEntityOwnershipService leaderEntityOwnershipService = newOwnershipService(leaderDistributedDataStore);

    leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);

    MemberNode follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name)
            .moduleShardsConfig(moduleShardsConfig).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false)
            .datastoreContextBuilder(followerDatastoreContextBuilder).build();

    AbstractDataStore follower1DistributedDataStore = follower1Node.configDataStore();
    follower1DistributedDataStore.waitTillReady();

    leaderNode.waitForMembersUp("member-2");
    follower1Node.waitForMembersUp("member-1");

    DOMEntityOwnershipService follower1EntityOwnershipService = newOwnershipService(follower1DistributedDataStore);

    leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);

    // Register a candidate for follower1 - should get queued since follower1 has no leader
    final DOMEntityOwnershipCandidateRegistration candidateReg =
            follower1EntityOwnershipService.registerCandidate(ENTITY1);
    Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
    verify(leaderMockListener, never()).ownershipChanged(ownershipChange(ENTITY1));

    // Add replica in follower1
    AddShardReplica addReplica = new AddShardReplica(ENTITY_OWNERSHIP_SHARD_NAME);
    follower1DistributedDataStore.getActorContext().getShardManager().tell(addReplica,
            follower1Node.kit().getRef());
    Object reply = follower1Node.kit().expectMsgAnyClassOf(JavaTestKit.duration("5 sec"),
            Success.class, Failure.class);
    if (reply instanceof Failure) {
        throw new AssertionError("AddShardReplica failed", ((Failure)reply).cause());
    }

    // The queued candidate registration should proceed
    verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));
    reset(leaderMockListener);

    candidateReg.close();
    verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, false));
    reset(leaderMockListener);

    // Restart follower1 and verify the entity ownership shard is re-instated by registering.
    Cluster.get(leaderNode.kit().getSystem()).down(Cluster.get(follower1Node.kit().getSystem()).selfAddress());
    follower1Node.cleanup();

    follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name)
            .moduleShardsConfig(moduleShardsConfig).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false)
            .datastoreContextBuilder(followerDatastoreContextBuilder).build();
    follower1EntityOwnershipService = newOwnershipService(follower1Node.configDataStore());

    follower1EntityOwnershipService.registerCandidate(ENTITY1);
    verify(leaderMockListener, timeout(20000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));

    verifyRaftState(follower1Node.configDataStore(), ENTITY_OWNERSHIP_SHARD_NAME, raftState -> {
        assertNull("Custom RaftPolicy class name", raftState.getCustomRaftPolicyClassName());
        assertEquals("Peer count", 1, raftState.getPeerAddresses().keySet().size());
        assertThat("Peer Id", Iterables.<String>getLast(raftState.getPeerAddresses().keySet()),
                org.hamcrest.CoreMatchers.containsString("member-1"));
    });
}