ShardManager extends AbstractUntypedPersistentActorWithMetering { @Override public void handleCommand(final Object message) throws Exception { if (message instanceof FindPrimary) { findPrimary((FindPrimary)message); } else if (message instanceof FindLocalShard) { findLocalShard((FindLocalShard) message); } else if (message instanceof UpdateSchemaContext) { updateSchemaContext(message); } else if (message instanceof ActorInitialized) { onActorInitialized(message); } else if (message instanceof ClusterEvent.MemberUp) { memberUp((ClusterEvent.MemberUp) message); } else if (message instanceof ClusterEvent.MemberWeaklyUp) { memberWeaklyUp((ClusterEvent.MemberWeaklyUp) message); } else if (message instanceof ClusterEvent.MemberExited) { memberExited((ClusterEvent.MemberExited) message); } else if (message instanceof ClusterEvent.MemberRemoved) { memberRemoved((ClusterEvent.MemberRemoved) message); } else if (message instanceof ClusterEvent.UnreachableMember) { memberUnreachable((ClusterEvent.UnreachableMember) message); } else if (message instanceof ClusterEvent.ReachableMember) { memberReachable((ClusterEvent.ReachableMember) message); } else if (message instanceof DatastoreContextFactory) { onDatastoreContextFactory((DatastoreContextFactory) message); } else if (message instanceof RoleChangeNotification) { onRoleChangeNotification((RoleChangeNotification) message); } else if (message instanceof FollowerInitialSyncUpStatus) { onFollowerInitialSyncStatus((FollowerInitialSyncUpStatus) message); } else if (message instanceof ShardNotInitializedTimeout) { onShardNotInitializedTimeout((ShardNotInitializedTimeout) message); } else if (message instanceof ShardLeaderStateChanged) { onLeaderStateChanged((ShardLeaderStateChanged) message); } else if (message instanceof SwitchShardBehavior) { onSwitchShardBehavior((SwitchShardBehavior) message); } else if (message instanceof CreateShard) { onCreateShard((CreateShard)message); } else if (message instanceof AddShardReplica) { onAddShardReplica((AddShardReplica) message); } else if (message instanceof AddPrefixShardReplica) { onAddPrefixShardReplica((AddPrefixShardReplica) message); } else if (message instanceof PrefixShardCreated) { onPrefixShardCreated((PrefixShardCreated) message); } else if (message instanceof PrefixShardRemoved) { onPrefixShardRemoved((PrefixShardRemoved) message); } else if (message instanceof InitConfigListener) { onInitConfigListener(); } else if (message instanceof ForwardedAddServerReply) { ForwardedAddServerReply msg = (ForwardedAddServerReply)message; onAddServerReply(msg.shardInfo, msg.addServerReply, getSender(), msg.leaderPath, msg.removeShardOnFailure); } else if (message instanceof ForwardedAddServerFailure) { ForwardedAddServerFailure msg = (ForwardedAddServerFailure)message; onAddServerFailure(msg.shardName, msg.failureMessage, msg.failure, getSender(), msg.removeShardOnFailure); } else if (message instanceof RemoveShardReplica) { onRemoveShardReplica((RemoveShardReplica) message); } else if (message instanceof RemovePrefixShardReplica) { onRemovePrefixShardReplica((RemovePrefixShardReplica) message); } else if (message instanceof WrappedShardResponse) { onWrappedShardResponse((WrappedShardResponse) message); } else if (message instanceof GetSnapshot) { onGetSnapshot((GetSnapshot) message); } else if (message instanceof ServerRemoved) { onShardReplicaRemoved((ServerRemoved) message); } else if (message instanceof ChangeShardMembersVotingStatus) { onChangeShardServersVotingStatus((ChangeShardMembersVotingStatus) message); } else if (message instanceof FlipShardMembersVotingStatus) { onFlipShardMembersVotingStatus((FlipShardMembersVotingStatus) message); } else if (message instanceof SaveSnapshotSuccess) { onSaveSnapshotSuccess((SaveSnapshotSuccess) message); } else if (message instanceof SaveSnapshotFailure) { LOG.error("{}: SaveSnapshotFailure received for saving snapshot of shards", persistenceId(), ((SaveSnapshotFailure) message).cause()); } else if (message instanceof Shutdown) { onShutDown(); } else if (message instanceof GetLocalShardIds) { onGetLocalShardIds(); } else if (message instanceof GetShardRole) { onGetShardRole((GetShardRole) message); } else if (message instanceof RunnableMessage) { ((RunnableMessage)message).run(); } else if (message instanceof RegisterForShardAvailabilityChanges) { onRegisterForShardAvailabilityChanges((RegisterForShardAvailabilityChanges)message); } else if (message instanceof DeleteSnapshotsFailure) { LOG.warn("{}: Failed to delete prior snapshots", persistenceId(), ((DeleteSnapshotsFailure) message).cause()); } else if (message instanceof DeleteSnapshotsSuccess) { LOG.debug("{}: Successfully deleted prior snapshots", persistenceId()); } else if (message instanceof RegisterRoleChangeListenerReply) { LOG.trace("{}: Received RegisterRoleChangeListenerReply", persistenceId()); } else if (message instanceof ClusterEvent.MemberEvent) { LOG.trace("{}: Received other ClusterEvent.MemberEvent: {}", persistenceId(), message); } else { unknownMessage(message); } } ShardManager(final AbstractShardManagerCreator<?> builder); @Override void preStart(); @Override void postStop(); @Override void handleCommand(final Object message); @Override SupervisorStrategy supervisorStrategy(); @Override String persistenceId();  }
@Test public void testRoleChangeNotificationAndShardLeaderStateChangedReleaseReady() throws Exception { TestShardManager shardManager = newTestShardManager(); String memberId = "member-1-shard-default-" + shardMrgIDSuffix; shardManager.handleCommand(new RoleChangeNotification( memberId, RaftState.Candidate.name(), RaftState.Leader.name())); assertFalse(ready.isDone()); shardManager.handleCommand(new ShardLeaderStateChanged(memberId, memberId, mock(DataTree.class), DataStoreVersions.CURRENT_VERSION)); assertTrue(ready.isDone()); }
@Test public void testRoleChangeNotificationToFollowerWithShardLeaderStateChangedReleaseReady() throws Exception { final TestKit kit = new TestKit(getSystem()); TestShardManager shardManager = newTestShardManager(); String memberId = "member-1-shard-default-" + shardMrgIDSuffix; shardManager.handleCommand(new RoleChangeNotification(memberId, null, RaftState.Follower.name())); assertFalse(ready.isDone()); shardManager.handleCommand(MockClusterWrapper.createMemberUp("member-2", kit.getRef().path().toString())); shardManager.handleCommand( new ShardLeaderStateChanged(memberId, "member-2-shard-default-" + shardMrgIDSuffix, mock(DataTree.class), DataStoreVersions.CURRENT_VERSION)); assertTrue(ready.isDone()); }
@Test public void testReadyCountDownForMemberUpAfterLeaderStateChanged() throws Exception { final TestKit kit = new TestKit(getSystem()); TestShardManager shardManager = newTestShardManager(); String memberId = "member-1-shard-default-" + shardMrgIDSuffix; shardManager.handleCommand(new RoleChangeNotification(memberId, null, RaftState.Follower.name())); assertFalse(ready.isDone()); shardManager.handleCommand( new ShardLeaderStateChanged(memberId, "member-2-shard-default-" + shardMrgIDSuffix, mock(DataTree.class), DataStoreVersions.CURRENT_VERSION)); shardManager.handleCommand(MockClusterWrapper.createMemberUp("member-2", kit.getRef().path().toString())); assertTrue(ready.isDone()); }
@Test public void testRoleChangeNotificationDoNothingForUnknownShard() throws Exception { TestShardManager shardManager = newTestShardManager(); shardManager.handleCommand(new RoleChangeNotification("unknown", RaftState.Candidate.name(), RaftState.Leader.name())); assertFalse(ready.isDone()); }