ClusterAdminRpcService implements ClusterAdminService { @Override public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards( final FlipMemberVotingStatesForAllShardsInput input) { final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>(); Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new; LOG.info("Flip member voting states for all shards"); sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier); sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier); return waitForShardResults(shardResultData, shardResults -> new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(), "Failed to change member voting states"); } ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
            final DistributedDataStoreInterface operDataStore,
            final BindingNormalizedNodeSerializer serializer); @Override ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input); @Override ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
            final RemoveShardReplicaInput input); @Override ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input); @Override ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input); @Override ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
            final AddPrefixShardReplicaInput input); @Override ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
            final RemovePrefixShardReplicaInput input); @Override ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
            final AddReplicasForAllShardsInput input); @Override ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
            final RemoveAllShardReplicasInput input); @Override ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
            final ChangeMemberVotingStatesForShardInput input); @Override ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
            final ChangeMemberVotingStatesForAllShardsInput input); @Override ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
            final FlipMemberVotingStatesForAllShardsInput input); @Override ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input); @Override ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
            final GetPrefixShardRoleInput input); @Override ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input); @Override ListenableFuture<RpcResult<GetKnownClientsForAllShardsOutput>> getKnownClientsForAllShards(
            final GetKnownClientsForAllShardsInput input);  }
@Test public void testFlipMemberVotingStates() throws Exception { String name = "testFlipMemberVotingStates"; ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList( new ServerInfo("member-1", true), new ServerInfo("member-2", true), new ServerInfo("member-3", false))); setupPersistedServerConfigPayload(persistedServerConfig, "member-1", name, "cars", "people"); setupPersistedServerConfigPayload(persistedServerConfig, "member-2", name, "cars", "people"); setupPersistedServerConfigPayload(persistedServerConfig, "member-3", name, "cars", "people"); String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf"; final MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder(DatastoreContext.newBuilder() .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10)) .build(); final MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); final MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); leaderNode1.configDataStore().waitTillReady(); leaderNode1.operDataStore().waitTillReady(); replicaNode3.configDataStore().waitTillReady(); replicaNode3.operDataStore().waitTillReady(); verifyVotingStates(leaderNode1.configDataStore(), "cars", new SimpleEntry<>("member-1", TRUE), new SimpleEntry<>("member-2", TRUE), new SimpleEntry<>("member-3", FALSE)); ClusterAdminRpcService service3 = new ClusterAdminRpcService(replicaNode3.configDataStore(), replicaNode3.operDataStore(), null); RpcResult<FlipMemberVotingStatesForAllShardsOutput> rpcResult = service3.flipMemberVotingStatesForAllShards( new FlipMemberVotingStatesForAllShardsInputBuilder().build()).get(10, TimeUnit.SECONDS); FlipMemberVotingStatesForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult); verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config), successShardResult("people", DataStoreType.Config), successShardResult("cars", DataStoreType.Operational), successShardResult("people", DataStoreType.Operational)); verifyVotingStates(new AbstractDataStore[]{leaderNode1.configDataStore(), leaderNode1.operDataStore(), replicaNode2.configDataStore(), replicaNode2.operDataStore(), replicaNode3.configDataStore(), replicaNode3.operDataStore()}, new String[]{"cars", "people"}, new SimpleEntry<>("member-1", FALSE), new SimpleEntry<>("member-2", FALSE), new SimpleEntry<>("member-3", TRUE)); verifyRaftState(leaderNode1.configDataStore(), "cars", raftState -> { assertNotNull("Expected non-null leader Id", raftState.getLeader()); assertTrue("Expected leader member-3. Actual: " + raftState.getLeader(), raftState.getLeader().contains("member-3")); }); verifyRaftState(leaderNode1.operDataStore(), "cars", raftState -> { assertNotNull("Expected non-null leader Id", raftState.getLeader()); assertTrue("Expected leader member-3. Actual: " + raftState.getLeader(), raftState.getLeader().contains("member-3")); }); rpcResult = service3.flipMemberVotingStatesForAllShards( new FlipMemberVotingStatesForAllShardsInputBuilder().build()).get(10, TimeUnit.SECONDS); result = verifySuccessfulRpcResult(rpcResult); verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config), successShardResult("people", DataStoreType.Config), successShardResult("cars", DataStoreType.Operational), successShardResult("people", DataStoreType.Operational)); verifyVotingStates(new AbstractDataStore[]{leaderNode1.configDataStore(), leaderNode1.operDataStore(), replicaNode2.configDataStore(), replicaNode2.operDataStore(), replicaNode3.configDataStore(), replicaNode3.operDataStore()}, new String[]{"cars", "people"}, new SimpleEntry<>("member-1", TRUE), new SimpleEntry<>("member-2", TRUE), new SimpleEntry<>("member-3", FALSE)); verifyRaftState(leaderNode1.configDataStore(), "cars", raftState -> { assertNotNull("Expected non-null leader Id", raftState.getLeader()); assertTrue("Expected leader member-1 or member-2. Actual: " + raftState.getLeader(), raftState.getLeader().contains("member-1") || raftState.getLeader().contains("member-2")); }); }
@Test public void testFlipMemberVotingStatesWithNoInitialLeader() throws Exception { String name = "testFlipMemberVotingStatesWithNoInitialLeader"; ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList( new ServerInfo("member-1", false), new ServerInfo("member-2", false), new ServerInfo("member-3", false), new ServerInfo("member-4", true), new ServerInfo("member-5", true), new ServerInfo("member-6", true))); setupPersistedServerConfigPayload(persistedServerConfig, "member-1", name, "cars", "people"); setupPersistedServerConfigPayload(persistedServerConfig, "member-2", name, "cars", "people"); setupPersistedServerConfigPayload(persistedServerConfig, "member-3", name, "cars", "people"); String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf"; final MemberNode replicaNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder( DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)) .build(); final MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); final MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); replicaNode1.waitForMembersUp("member-2", "member-3"); verifyVotingStates(replicaNode1.configDataStore(), "cars", new SimpleEntry<>("member-1", FALSE), new SimpleEntry<>("member-2", FALSE), new SimpleEntry<>("member-3", FALSE), new SimpleEntry<>("member-4", TRUE), new SimpleEntry<>("member-5", TRUE), new SimpleEntry<>("member-6", TRUE)); verifyRaftState(replicaNode1.configDataStore(), "cars", raftState -> assertEquals("Expected raft state", RaftState.Follower.toString(), raftState.getRaftState())); ClusterAdminRpcService service1 = new ClusterAdminRpcService(replicaNode1.configDataStore(), replicaNode1.operDataStore(), null); RpcResult<FlipMemberVotingStatesForAllShardsOutput> rpcResult = service1.flipMemberVotingStatesForAllShards( new FlipMemberVotingStatesForAllShardsInputBuilder().build()).get(10, TimeUnit.SECONDS); FlipMemberVotingStatesForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult); verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config), successShardResult("people", DataStoreType.Config), successShardResult("cars", DataStoreType.Operational), successShardResult("people", DataStoreType.Operational)); verifyVotingStates(new AbstractDataStore[]{replicaNode1.configDataStore(), replicaNode1.operDataStore(), replicaNode2.configDataStore(), replicaNode2.operDataStore(), replicaNode3.configDataStore(), replicaNode3.operDataStore()}, new String[]{"cars", "people"}, new SimpleEntry<>("member-1", TRUE), new SimpleEntry<>("member-2", TRUE), new SimpleEntry<>("member-3", TRUE), new SimpleEntry<>("member-4", FALSE), new SimpleEntry<>("member-5", FALSE), new SimpleEntry<>("member-6", FALSE)); verifyRaftState(replicaNode1.configDataStore(), "cars", raftState -> { assertNotNull("Expected non-null leader Id", raftState.getLeader()); assertTrue("Expected leader member-1. Actual: " + raftState.getLeader(), raftState.getLeader().contains("member-1")); }); verifyRaftState(replicaNode1.operDataStore(), "cars", raftState -> { assertNotNull("Expected non-null leader Id", raftState.getLeader()); assertTrue("Expected leader member-1. Actual: " + raftState.getLeader(), raftState.getLeader().contains("member-1")); }); }
@Test public void testFlipMemberVotingStatesWithVotingMembersDown() throws Exception { String name = "testFlipMemberVotingStatesWithVotingMembersDown"; ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList( new ServerInfo("member-1", true), new ServerInfo("member-2", true), new ServerInfo("member-3", true), new ServerInfo("member-4", false), new ServerInfo("member-5", false), new ServerInfo("member-6", false))); setupPersistedServerConfigPayload(persistedServerConfig, "member-1", name, "cars", "people"); setupPersistedServerConfigPayload(persistedServerConfig, "member-2", name, "cars", "people"); setupPersistedServerConfigPayload(persistedServerConfig, "member-3", name, "cars", "people"); String moduleShardsConfig = "module-shards-member1-and-2-and-3.conf"; final MemberNode leaderNode1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name) .moduleShardsConfig(moduleShardsConfig).datastoreContextBuilder( DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(300).shardElectionTimeoutFactor(1)) .build(); final MemberNode replicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); final MemberNode replicaNode3 = MemberNode.builder(memberNodes).akkaConfig("Member3").testName(name) .moduleShardsConfig(moduleShardsConfig).build(); leaderNode1.configDataStore().waitTillReady(); leaderNode1.operDataStore().waitTillReady(); verifyVotingStates(leaderNode1.configDataStore(), "cars", new SimpleEntry<>("member-1", TRUE), new SimpleEntry<>("member-2", TRUE), new SimpleEntry<>("member-3", TRUE), new SimpleEntry<>("member-4", FALSE), new SimpleEntry<>("member-5", FALSE), new SimpleEntry<>("member-6", FALSE)); ClusterAdminRpcService service1 = new ClusterAdminRpcService(leaderNode1.configDataStore(), leaderNode1.operDataStore(), null); RpcResult<FlipMemberVotingStatesForAllShardsOutput> rpcResult = service1.flipMemberVotingStatesForAllShards( new FlipMemberVotingStatesForAllShardsInputBuilder().build()).get(10, TimeUnit.SECONDS); FlipMemberVotingStatesForAllShardsOutput result = verifySuccessfulRpcResult(rpcResult); verifyShardResults(result.getShardResult(), successShardResult("cars", DataStoreType.Config), successShardResult("people", DataStoreType.Config), successShardResult("cars", DataStoreType.Operational), successShardResult("people", DataStoreType.Operational)); verifyVotingStates(new AbstractDataStore[]{leaderNode1.configDataStore(), leaderNode1.operDataStore(), replicaNode2.configDataStore(), replicaNode2.operDataStore(), replicaNode3.configDataStore(), replicaNode3.operDataStore()}, new String[]{"cars", "people"}, new SimpleEntry<>("member-1", FALSE), new SimpleEntry<>("member-2", FALSE), new SimpleEntry<>("member-3", FALSE), new SimpleEntry<>("member-4", TRUE), new SimpleEntry<>("member-5", TRUE), new SimpleEntry<>("member-6", TRUE)); verifyRaftState(leaderNode1.configDataStore(), "cars", raftState -> { assertNotNull("Expected non-null leader Id", raftState.getLeader()); assertTrue("Expected leader member-1", raftState.getLeader().contains("member-1")); }); }