+ private void checkReady(){
+ if (isReadyWithLeaderId()) {
+ LOG.info("{}: All Shards are ready - data store {} is ready, available count is {}",
+ persistenceId(), type, waitTillReadyCountdownLatch.getCount());
+
+ waitTillReadyCountdownLatch.countDown();
+ }
+ }
+
+ private void onLeaderStateChanged(ShardLeaderStateChanged leaderStateChanged) {
+ LOG.info("{}: Received LeaderStateChanged message: {}", persistenceId(), leaderStateChanged);
+
+ ShardInformation shardInformation = findShardInformation(leaderStateChanged.getMemberId());
+ if(shardInformation != null) {
+ shardInformation.setLocalDataTree(leaderStateChanged.getLocalShardDataTree());
+ shardInformation.setLeaderVersion(leaderStateChanged.getLeaderPayloadVersion());
+ if(shardInformation.setLeaderId(leaderStateChanged.getLeaderId())) {
+ primaryShardInfoCache.remove(shardInformation.getShardName());
+ }
+
+ checkReady();
+ } else {
+ LOG.debug("No shard found with member Id {}", leaderStateChanged.getMemberId());
+ }
+ }
+
+ private void onShardNotInitializedTimeout(ShardNotInitializedTimeout message) {
+ ShardInformation shardInfo = message.getShardInfo();
+
+ LOG.debug("{}: Received ShardNotInitializedTimeout message for shard {}", persistenceId(),
+ shardInfo.getShardName());
+
+ shardInfo.removeOnShardInitialized(message.getOnShardInitialized());
+
+ if(!shardInfo.isShardInitialized()) {
+ LOG.debug("{}: Returning NotInitializedException for shard {}", persistenceId(), shardInfo.getShardName());
+ message.getSender().tell(createNotInitializedException(shardInfo.shardId), getSelf());
+ } else {
+ LOG.debug("{}: Returning NoShardLeaderException for shard {}", persistenceId(), shardInfo.getShardName());
+ message.getSender().tell(createNoShardLeaderException(shardInfo.shardId), getSelf());
+ }
+ }
+
+ private void onFollowerInitialSyncStatus(FollowerInitialSyncUpStatus status) {
+ LOG.info("{} Received follower initial sync status for {} status sync done {}", persistenceId(),
+ status.getName(), status.isInitialSyncDone());
+
+ ShardInformation shardInformation = findShardInformation(status.getName());
+
+ if(shardInformation != null) {
+ shardInformation.setFollowerSyncStatus(status.isInitialSyncDone());
+
+ mBean.setSyncStatus(isInSync());
+ }
+
+ }
+
+ private void onRoleChangeNotification(RoleChangeNotification roleChanged) {
+ LOG.info("{}: Received role changed for {} from {} to {}", persistenceId(), roleChanged.getMemberId(),
+ roleChanged.getOldRole(), roleChanged.getNewRole());
+
+ ShardInformation shardInformation = findShardInformation(roleChanged.getMemberId());
+ if(shardInformation != null) {
+ shardInformation.setRole(roleChanged.getNewRole());
+ checkReady();
+ mBean.setSyncStatus(isInSync());
+ }
+ }
+
+
+ private ShardInformation findShardInformation(String memberId) {
+ for(ShardInformation info : localShards.values()){
+ if(info.getShardId().toString().equals(memberId)){
+ return info;
+ }
+ }
+
+ return null;
+ }
+
+ private boolean isReadyWithLeaderId() {
+ boolean isReady = true;
+ for (ShardInformation info : localShards.values()) {
+ if(!info.isShardReadyWithLeaderId()){
+ isReady = false;
+ break;
+ }
+ }
+ return isReady;
+ }
+
+ private boolean isInSync(){
+ for (ShardInformation info : localShards.values()) {
+ if(!info.isInSync()){
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private void onActorInitialized(Object message) {
+ final ActorRef sender = getSender();
+
+ if (sender == null) {
+ return; //why is a non-actor sending this message? Just ignore.
+ }
+
+ String actorName = sender.path().name();
+ //find shard name from actor name; actor name is stringified shardId
+ ShardIdentifier shardId = ShardIdentifier.builder().fromShardIdString(actorName).build();
+
+ if (shardId.getShardName() == null) {
+ return;
+ }
+
+ markShardAsInitialized(shardId.getShardName());
+ }
+
+ private void markShardAsInitialized(String shardName) {
+ LOG.debug("{}: Initializing shard [{}]", persistenceId(), shardName);
+
+ ShardInformation shardInformation = localShards.get(shardName);
+ if (shardInformation != null) {
+ shardInformation.setActorInitialized();
+
+ shardInformation.getActor().tell(new RegisterRoleChangeListener(), self());
+ }
+ }
+
+ @Override
+ protected void handleRecover(Object message) throws Exception {
+ if (message instanceof RecoveryCompleted) {
+ LOG.info("Recovery complete : {}", persistenceId());
+
+ // We no longer persist SchemaContext modules so delete all the prior messages from the akka
+ // journal on upgrade from Helium.
+ deleteMessages(lastSequenceNr());
+ }
+ }
+