X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Fshardmanager%2FShardManager.java;fp=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Fshardmanager%2FShardManager.java;h=dadbeeb4309703f80a823ca8ac3774e9b0e7a7e7;hp=38922df30c0235d8357edb2c91c44ce3b4d222b8;hb=0a2ed6b43f45b92f09c291e99e2e66a7fa18085f;hpb=03e868508b52c1e59bdba364deccf69679307b01 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java index 38922df30c..dadbeeb430 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java @@ -178,7 +178,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { private ListenerRegistration configListenerReg = null; private PrefixedShardConfigUpdateHandler configUpdateHandler; - ShardManager(AbstractShardManagerCreator builder) { + ShardManager(final AbstractShardManagerCreator builder) { this.cluster = builder.getCluster(); this.configuration = builder.getConfiguration(); this.datastoreContextFactory = builder.getDatastoreContextFactory(); @@ -223,7 +223,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @Override - public void handleCommand(Object message) throws Exception { + public void handleCommand(final Object message) throws Exception { if (message instanceof FindPrimary) { findPrimary((FindPrimary)message); } else if (message instanceof FindLocalShard) { @@ -368,7 +368,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { combinedFutures.onComplete(new OnComplete>() { @Override - public void onComplete(Throwable failure, Iterable results) { + public void onComplete(final Throwable failure, final Iterable results) { LOG.debug("{}: All shards shutdown - sending PoisonPill to self", persistenceId()); self().tell(PoisonPill.getInstance(), self()); @@ -391,15 +391,15 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { }, dispatcher); } - private void onWrappedShardResponse(WrappedShardResponse message) { + private void onWrappedShardResponse(final WrappedShardResponse message) { if (message.getResponse() instanceof RemoveServerReply) { onRemoveServerReply(getSender(), message.getShardId(), (RemoveServerReply) message.getResponse(), message.getLeaderPath()); } } - private void onRemoveServerReply(ActorRef originalSender, ShardIdentifier shardId, RemoveServerReply replyMsg, - String leaderPath) { + private void onRemoveServerReply(final ActorRef originalSender, final ShardIdentifier shardId, + final RemoveServerReply replyMsg, final String leaderPath) { shardReplicaOperationsInProgress.remove(shardId.getShardName()); LOG.debug("{}: Received {} for shard {}", persistenceId(), replyMsg, shardId.getShardName()); @@ -439,7 +439,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { futureObj.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Object response) { + public void onComplete(final Throwable failure, final Object response) { if (failure != null) { shardReplicaOperationsInProgress.remove(shardName); String msg = String.format("RemoveServer request to leader %s for shard %s failed", @@ -457,8 +457,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client)); } - private void removeShardReplica(RemoveShardReplica contextMessage, final String shardName, final String primaryPath, - final ActorRef sender) { + private void removeShardReplica(final RemoveShardReplica contextMessage, final String shardName, + final String primaryPath, final ActorRef sender) { if (isShardReplicaOperationInProgress(shardName, sender)) { return; } @@ -479,7 +479,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { futureObj.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Object response) { + public void onComplete(final Throwable failure, final Object response) { if (failure != null) { shardReplicaOperationsInProgress.remove(shardName); String msg = String.format("RemoveServer request to leader %s for shard %s failed", @@ -497,7 +497,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client)); } - private void onShardReplicaRemoved(ServerRemoved message) { + private void onShardReplicaRemoved(final ServerRemoved message) { removeShard(new ShardIdentifier.Builder().fromShardIdString(message.getServerId()).build()); } @@ -576,7 +576,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @SuppressWarnings("checkstyle:IllegalCatch") - private void onCreateShard(CreateShard createShard) { + private void onCreateShard(final CreateShard createShard) { LOG.debug("{}: onCreateShard: {}", persistenceId(), createShard); Object reply; @@ -636,7 +636,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { final ActorRef sender = getSender(); stopOnComplete.addOnComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Boolean result) { + public void onComplete(final Throwable failure, final Boolean result) { LOG.debug("{} : Stop complete for shard {} - re-queing {}", persistenceId(), shardName, messageToDefer); self().tell(messageToDefer, sender); } @@ -645,7 +645,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { return true; } - private void doCreatePrefixShard(PrefixShardConfiguration config, ShardIdentifier shardId, String shardName) { + private void doCreatePrefixShard(final PrefixShardConfiguration config, final ShardIdentifier shardId, + final String shardName) { configuration.addPrefixShardConfiguration(config); final Builder builder = newShardDatastoreContextBuilder(shardName); @@ -665,7 +666,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { localShards.put(info.getShardName(), info); if (schemaContext != null) { - info.setActor(newShardActor(schemaContext, info)); + info.setSchemaContext(schemaContext); + info.setActor(newShardActor(info)); } } @@ -726,16 +728,17 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { localShards.put(info.getShardName(), info); if (schemaContext != null) { - info.setActor(newShardActor(schemaContext, info)); + info.setSchemaContext(schemaContext); + info.setActor(newShardActor(info)); } } - private DatastoreContext.Builder newShardDatastoreContextBuilder(String shardName) { + private DatastoreContext.Builder newShardDatastoreContextBuilder(final String shardName) { return DatastoreContext.newBuilderFrom(datastoreContextFactory.getShardDatastoreContext(shardName)) .shardPeerAddressResolver(peerAddressResolver); } - private DatastoreContext newShardDatastoreContext(String shardName) { + private DatastoreContext newShardDatastoreContext(final String shardName) { return newShardDatastoreContextBuilder(shardName).build(); } @@ -748,7 +751,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private void onLeaderStateChanged(ShardLeaderStateChanged leaderStateChanged) { + private void onLeaderStateChanged(final ShardLeaderStateChanged leaderStateChanged) { LOG.info("{}: Received LeaderStateChanged message: {}", persistenceId(), leaderStateChanged); ShardInformation shardInformation = findShardInformation(leaderStateChanged.getMemberId()); @@ -765,7 +768,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private void onShardNotInitializedTimeout(ShardNotInitializedTimeout message) { + private void onShardNotInitializedTimeout(final ShardNotInitializedTimeout message) { ShardInformation shardInfo = message.getShardInfo(); LOG.debug("{}: Received ShardNotInitializedTimeout message for shard {}", persistenceId(), @@ -782,7 +785,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private void onFollowerInitialSyncStatus(FollowerInitialSyncUpStatus status) { + private void onFollowerInitialSyncStatus(final FollowerInitialSyncUpStatus status) { LOG.info("{} Received follower initial sync status for {} status sync done {}", persistenceId(), status.getName(), status.isInitialSyncDone()); @@ -796,7 +799,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } - private void onRoleChangeNotification(RoleChangeNotification roleChanged) { + private void onRoleChangeNotification(final RoleChangeNotification roleChanged) { LOG.info("{}: Received role changed for {} from {} to {}", persistenceId(), roleChanged.getMemberId(), roleChanged.getOldRole(), roleChanged.getNewRole()); @@ -809,7 +812,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } - private ShardInformation findShardInformation(String memberId) { + private ShardInformation findShardInformation(final String memberId) { for (ShardInformation info : localShards.values()) { if (info.getShardId().toString().equals(memberId)) { return info; @@ -839,7 +842,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { return true; } - private void onActorInitialized(Object message) { + private void onActorInitialized(final Object message) { final ActorRef sender = getSender(); if (sender == null) { @@ -860,7 +863,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { markShardAsInitialized(shardId.getShardName()); } - private void markShardAsInitialized(String shardName) { + private void markShardAsInitialized(final String shardName) { LOG.debug("{}: Initializing shard [{}]", persistenceId(), shardName); ShardInformation shardInformation = localShards.get(shardName); @@ -872,7 +875,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @Override - protected void handleRecover(Object message) throws Exception { + protected void handleRecover(final Object message) throws Exception { if (message instanceof RecoveryCompleted) { onRecoveryCompleted(); } else if (message instanceof SnapshotOffer) { @@ -896,8 +899,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { createLocalShards(); } - private void sendResponse(ShardInformation shardInformation, boolean doWait, - boolean wantShardReady, final Supplier messageSupplier) { + private void sendResponse(final ShardInformation shardInformation, final boolean doWait, + final boolean wantShardReady, final Supplier messageSupplier) { if (!shardInformation.isShardInitialized() || wantShardReady && !shardInformation.isShardReadyWithLeaderId()) { if (doWait) { final ActorRef sender = getSender(); @@ -945,11 +948,11 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { getSender().tell(messageSupplier.get(), getSelf()); } - private static NoShardLeaderException createNoShardLeaderException(ShardIdentifier shardId) { + private static NoShardLeaderException createNoShardLeaderException(final ShardIdentifier shardId) { return new NoShardLeaderException(null, shardId.toString()); } - private static NotInitializedException createNotInitializedException(ShardIdentifier shardId) { + private static NotInitializedException createNotInitializedException(final ShardIdentifier shardId) { return new NotInitializedException(String.format( "Found primary shard %s but it's not initialized yet. Please try again later", shardId)); } @@ -959,7 +962,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { return MemberName.forName(member.roles().iterator().next()); } - private void memberRemoved(ClusterEvent.MemberRemoved message) { + private void memberRemoved(final ClusterEvent.MemberRemoved message) { MemberName memberName = memberToName(message.member()); LOG.info("{}: Received MemberRemoved: memberName: {}, address: {}", persistenceId(), memberName, @@ -972,7 +975,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private void memberExited(ClusterEvent.MemberExited message) { + private void memberExited(final ClusterEvent.MemberExited message) { MemberName memberName = memberToName(message.member()); LOG.info("{}: Received MemberExited: memberName: {}, address: {}", persistenceId(), memberName, @@ -985,7 +988,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private void memberUp(ClusterEvent.MemberUp message) { + private void memberUp(final ClusterEvent.MemberUp message) { MemberName memberName = memberToName(message.member()); LOG.info("{}: Received MemberUp: memberName: {}, address: {}", persistenceId(), memberName, @@ -994,12 +997,12 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { memberUp(memberName, message.member().address()); } - private void memberUp(MemberName memberName, Address address) { + private void memberUp(final MemberName memberName, final Address address) { addPeerAddress(memberName, address); checkReady(); } - private void memberWeaklyUp(MemberWeaklyUp message) { + private void memberWeaklyUp(final MemberWeaklyUp message) { MemberName memberName = memberToName(message.member()); LOG.info("{}: Received MemberWeaklyUp: memberName: {}, address: {}", persistenceId(), memberName, @@ -1008,7 +1011,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { memberUp(memberName, message.member().address()); } - private void addPeerAddress(MemberName memberName, Address address) { + private void addPeerAddress(final MemberName memberName, final Address address) { peerAddressResolver.addPeerAddress(memberName, address); for (ShardInformation info : localShards.values()) { @@ -1020,7 +1023,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private void memberReachable(ClusterEvent.ReachableMember message) { + private void memberReachable(final ClusterEvent.ReachableMember message) { MemberName memberName = memberToName(message.member()); LOG.info("Received ReachableMember: memberName {}, address: {}", memberName, message.member().address()); @@ -1029,7 +1032,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { markMemberAvailable(memberName); } - private void memberUnreachable(ClusterEvent.UnreachableMember message) { + private void memberUnreachable(final ClusterEvent.UnreachableMember message) { MemberName memberName = memberToName(message.member()); LOG.info("Received UnreachableMember: memberName {}, address: {}", memberName, message.member().address()); @@ -1066,7 +1069,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private void onDatastoreContextFactory(DatastoreContextFactory factory) { + private void onDatastoreContextFactory(final DatastoreContextFactory factory) { datastoreContextFactory = factory; for (ShardInformation info : localShards.values()) { info.setDatastoreContext(newShardDatastoreContext(info.getShardName()), getSelf()); @@ -1125,9 +1128,11 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { LOG.debug("Got updated SchemaContext: # of modules {}", schemaContext.getAllModuleIdentifiers().size()); for (ShardInformation info : localShards.values()) { + info.setSchemaContext(schemaContext); + if (info.getActor() == null) { LOG.debug("Creating Shard {}", info.getShardId()); - info.setActor(newShardActor(schemaContext, info)); + info.setActor(newShardActor(info)); } else { info.getActor().tell(message, getSelf()); } @@ -1140,12 +1145,12 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @VisibleForTesting - protected ActorRef newShardActor(final SchemaContext shardSchemaContext, final ShardInformation info) { - return getContext().actorOf(info.newProps(shardSchemaContext).withDispatcher(shardDispatcherPath), + protected ActorRef newShardActor(final ShardInformation info) { + return getContext().actorOf(info.newProps().withDispatcher(shardDispatcherPath), info.getShardId().toString()); } - private void findPrimary(FindPrimary message) { + private void findPrimary(final FindPrimary message) { LOG.debug("{}: In findPrimary: {}", persistenceId(), message); final String shardName = message.getShardName(); @@ -1202,7 +1207,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { Future futureObj = ask(getSelf(), new FindPrimary(shardName, true), findPrimaryTimeout); futureObj.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Object response) { + public void onComplete(final Throwable failure, final Object response) { if (failure != null) { handler.onFailure(failure); } else { @@ -1226,7 +1231,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { * @param shardName the shard name * @return a b */ - private ShardIdentifier getShardIdentifier(MemberName memberName, String shardName) { + private ShardIdentifier getShardIdentifier(final MemberName memberName, final String shardName) { return peerAddressResolver.getShardIdentifier(memberName, shardName); } @@ -1263,7 +1268,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { * * @param shardName the shard name */ - private Map getPeerAddresses(String shardName) { + private Map getPeerAddresses(final String shardName) { final Collection members = configuration.getMembersFromShardName(shardName); return getPeerAddresses(shardName, members); } @@ -1332,7 +1337,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { findPrimary(shardName, new AutoFindPrimaryFailureResponseHandler(getSender(), shardName, persistenceId(), getSelf()) { @Override - public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) { + public void onRemotePrimaryShardFound(final RemotePrimaryShardFound response) { final RunnableMessage runnable = (RunnableMessage) () -> addPrefixShard(getShardName(), message.getShardPrefix(), response, getSender()); if (!isPreviousShardActorStopInProgress(getShardName(), runnable)) { @@ -1341,7 +1346,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @Override - public void onLocalPrimaryFound(LocalPrimaryShardFound message) { + public void onLocalPrimaryFound(final LocalPrimaryShardFound message) { sendLocalReplicaAlreadyExistsReply(getShardName(), getTargetActor()); } }); @@ -1372,7 +1377,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { findPrimary(shardName, new AutoFindPrimaryFailureResponseHandler(getSender(), shardName, persistenceId(), getSelf()) { @Override - public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) { + public void onRemotePrimaryShardFound(final RemotePrimaryShardFound response) { final RunnableMessage runnable = (RunnableMessage) () -> addShard(getShardName(), response, getSender()); if (!isPreviousShardActorStopInProgress(getShardName(), runnable)) { @@ -1381,13 +1386,13 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @Override - public void onLocalPrimaryFound(LocalPrimaryShardFound message) { + public void onLocalPrimaryFound(final LocalPrimaryShardFound message) { sendLocalReplicaAlreadyExistsReply(getShardName(), getTargetActor()); } }); } - private void sendLocalReplicaAlreadyExistsReply(String shardName, ActorRef sender) { + private void sendLocalReplicaAlreadyExistsReply(final String shardName, final ActorRef sender) { String msg = String.format("Local shard %s already exists", shardName); LOG.debug("{}: {}", persistenceId(), msg); sender.tell(new Status.Failure(new AlreadyExistsException(msg)), getSelf()); @@ -1416,8 +1421,9 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { shardInfo = new ShardInformation(shardName, shardId, getPeerAddresses(shardName), datastoreContext, Shard.builder(), peerAddressResolver); shardInfo.setActiveMember(false); + shardInfo.setSchemaContext(schemaContext); localShards.put(shardName, shardInfo); - shardInfo.setActor(newShardActor(schemaContext, shardInfo)); + shardInfo.setActor(newShardActor(shardInfo)); } else { removeShardOnFailure = false; shardInfo = existingShardInfo; @@ -1446,8 +1452,9 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { shardInfo = new ShardInformation(shardName, shardId, getPeerAddresses(shardName), datastoreContext, Shard.builder(), peerAddressResolver); shardInfo.setActiveMember(false); + shardInfo.setSchemaContext(schemaContext); localShards.put(shardName, shardInfo); - shardInfo.setActor(newShardActor(schemaContext, shardInfo)); + shardInfo.setActor(newShardActor(shardInfo)); } else { removeShardOnFailure = false; shardInfo = existingShardInfo; @@ -1476,7 +1483,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { futureObj.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Object addServerResponse) { + public void onComplete(final Throwable failure, final Object addServerResponse) { if (failure != null) { LOG.debug("{}: AddServer request to {} for {} failed", persistenceId(), response.getPrimaryPath(), shardName, failure); @@ -1492,8 +1499,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client)); } - private void onAddServerFailure(String shardName, String message, Throwable failure, ActorRef sender, - boolean removeShardOnFailure) { + private void onAddServerFailure(final String shardName, final String message, final Throwable failure, + final ActorRef sender, final boolean removeShardOnFailure) { shardReplicaOperationsInProgress.remove(shardName); if (removeShardOnFailure) { @@ -1507,8 +1514,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { new RuntimeException(message, failure)), getSelf()); } - private void onAddServerReply(ShardInformation shardInfo, AddServerReply replyMsg, ActorRef sender, - String leaderPath, boolean removeShardOnFailure) { + private void onAddServerReply(final ShardInformation shardInfo, final AddServerReply replyMsg, + final ActorRef sender, final String leaderPath, final boolean removeShardOnFailure) { String shardName = shardInfo.getShardName(); shardReplicaOperationsInProgress.remove(shardName); @@ -1536,29 +1543,23 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private static Exception getServerChangeException(Class serverChange, ServerChangeStatus serverChangeStatus, - String leaderPath, ShardIdentifier shardId) { - Exception failure; + private static Exception getServerChangeException(final Class serverChange, + final ServerChangeStatus serverChangeStatus, final String leaderPath, final ShardIdentifier shardId) { switch (serverChangeStatus) { case TIMEOUT: - failure = new TimeoutException(String.format( + return new TimeoutException(String.format( "The shard leader %s timed out trying to replicate the initial data to the new shard %s." + "Possible causes - there was a problem replicating the data or shard leadership changed " + "while replicating the shard data", leaderPath, shardId.getShardName())); - break; case NO_LEADER: - failure = createNoShardLeaderException(shardId); - break; + return createNoShardLeaderException(shardId); case NOT_SUPPORTED: - failure = new UnsupportedOperationException(String.format("%s request is not supported for shard %s", + return new UnsupportedOperationException(String.format("%s request is not supported for shard %s", serverChange.getSimpleName(), shardId.getShardName())); - break; default : - failure = new RuntimeException(String.format( - "%s request to leader %s for shard %s failed with status %s", + return new RuntimeException(String.format("%s request to leader %s for shard %s failed with status %s", serverChange.getSimpleName(), leaderPath, shardId.getShardName(), serverChangeStatus)); } - return failure; } private void onRemoveShardReplica(final RemoveShardReplica shardReplicaMsg) { @@ -1567,12 +1568,12 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { findPrimary(shardReplicaMsg.getShardName(), new AutoFindPrimaryFailureResponseHandler(getSender(), shardReplicaMsg.getShardName(), persistenceId(), getSelf()) { @Override - public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) { + public void onRemotePrimaryShardFound(final RemotePrimaryShardFound response) { doRemoveShardReplicaAsync(response.getPrimaryPath()); } @Override - public void onLocalPrimaryFound(LocalPrimaryShardFound response) { + public void onLocalPrimaryFound(final LocalPrimaryShardFound response) { doRemoveShardReplicaAsync(response.getPrimaryPath()); } @@ -1593,12 +1594,12 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { findPrimary(shardName, new AutoFindPrimaryFailureResponseHandler(getSender(), shardName, persistenceId(), getSelf()) { @Override - public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) { + public void onRemotePrimaryShardFound(final RemotePrimaryShardFound response) { doRemoveShardReplicaAsync(response.getPrimaryPath()); } @Override - public void onLocalPrimaryFound(LocalPrimaryShardFound response) { + public void onLocalPrimaryFound(final LocalPrimaryShardFound response) { doRemoveShardReplicaAsync(response.getPrimaryPath()); } @@ -1627,7 +1628,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { return currentSnapshot; } - private void applyShardManagerSnapshot(ShardManagerSnapshot snapshot) { + private void applyShardManagerSnapshot(final ShardManagerSnapshot snapshot) { currentSnapshot = snapshot; LOG.debug("{}: onSnapshotOffer: {}", persistenceId(), currentSnapshot); @@ -1651,7 +1652,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } - private void onSaveSnapshotSuccess(SaveSnapshotSuccess successMessage) { + private void onSaveSnapshotSuccess(final SaveSnapshotSuccess successMessage) { LOG.debug("{} saved ShardManager snapshot successfully. Deleting the prev snapshot if available", persistenceId()); deleteSnapshots(new SnapshotSelectionCriteria(scala.Long.MaxValue(), successMessage.metadata().timestamp() - 1, @@ -1675,7 +1676,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { localShardFound.getPath(), getSender())); } - private void onFlipShardMembersVotingStatus(FlipShardMembersVotingStatus flipMembersVotingStatus) { + private void onFlipShardMembersVotingStatus(final FlipShardMembersVotingStatus flipMembersVotingStatus) { LOG.debug("{}: onFlipShardMembersVotingStatus: {}", persistenceId(), flipMembersVotingStatus); ActorRef sender = getSender(); @@ -1686,7 +1687,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { future.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Object response) { + public void onComplete(final Throwable failure, final Object response) { if (failure != null) { sender.tell(new Status.Failure(new RuntimeException( String.format("Failed to access local shard %s", shardName), failure)), self()); @@ -1710,7 +1711,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } - private void findLocalShard(FindLocalShard message) { + private void findLocalShard(final FindLocalShard message) { LOG.debug("{}: findLocalShard : {}", persistenceId(), message.getShardName()); final ShardInformation shardInformation = localShards.get(message.getShardName()); @@ -1735,7 +1736,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { Future futureObj = ask(getSelf(), new FindLocalShard(shardName, true), findLocalTimeout); futureObj.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Object response) { + public void onComplete(final Throwable failure, final Object response) { if (failure != null) { LOG.debug("{}: Received failure from FindLocalShard for shard {}", persistenceId, shardName, failure); @@ -1761,7 +1762,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client)); } - private void changeShardMembersVotingStatus(ChangeServersVotingStatus changeServersVotingStatus, + private void changeShardMembersVotingStatus(final ChangeServersVotingStatus changeServersVotingStatus, final String shardName, final ActorRef shardActorRef, final ActorRef sender) { if (isShardReplicaOperationInProgress(shardName, sender)) { return; @@ -1780,7 +1781,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { futureObj.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, Object response) { + public void onComplete(final Throwable failure, final Object response) { shardReplicaOperationsInProgress.remove(shardName); if (failure != null) { String msg = String.format("ChangeServersVotingStatus request to local shard %s failed", @@ -1817,8 +1818,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { String leaderPath; boolean removeShardOnFailure; - ForwardedAddServerReply(ShardInformation shardInfo, AddServerReply addServerReply, String leaderPath, - boolean removeShardOnFailure) { + ForwardedAddServerReply(final ShardInformation shardInfo, final AddServerReply addServerReply, + final String leaderPath, final boolean removeShardOnFailure) { this.shardInfo = shardInfo; this.addServerReply = addServerReply; this.leaderPath = leaderPath; @@ -1832,8 +1833,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { Throwable failure; boolean removeShardOnFailure; - ForwardedAddServerFailure(String shardName, String failureMessage, Throwable failure, - boolean removeShardOnFailure) { + ForwardedAddServerFailure(final String shardName, final String failureMessage, final Throwable failure, + final boolean removeShardOnFailure) { this.shardName = shardName; this.failureMessage = failureMessage; this.failure = failure; @@ -1845,7 +1846,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { private final Runnable replyRunnable; private Cancellable timeoutSchedule; - OnShardInitialized(Runnable replyRunnable) { + OnShardInitialized(final Runnable replyRunnable) { this.replyRunnable = replyRunnable; } @@ -1857,13 +1858,13 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { return timeoutSchedule; } - void setTimeoutSchedule(Cancellable timeoutSchedule) { + void setTimeoutSchedule(final Cancellable timeoutSchedule) { this.timeoutSchedule = timeoutSchedule; } } static class OnShardReady extends OnShardInitialized { - OnShardReady(Runnable replyRunnable) { + OnShardReady(final Runnable replyRunnable) { super(replyRunnable); } } @@ -1923,8 +1924,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { * @param persistenceId The persistenceId for the ShardManager * @param shardManagerActor The ShardManager actor which triggered the call to FindPrimary */ - protected AutoFindPrimaryFailureResponseHandler(ActorRef targetActor, String shardName, String persistenceId, - ActorRef shardManagerActor) { + protected AutoFindPrimaryFailureResponseHandler(final ActorRef targetActor, final String shardName, + final String persistenceId, final ActorRef shardManagerActor) { this.targetActor = Preconditions.checkNotNull(targetActor); this.shardName = Preconditions.checkNotNull(shardName); this.persistenceId = Preconditions.checkNotNull(persistenceId); @@ -1940,14 +1941,14 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @Override - public void onFailure(Throwable failure) { + public void onFailure(final Throwable failure) { LOG.debug("{}: Received failure from FindPrimary for shard {}", persistenceId, shardName, failure); targetActor.tell(new Status.Failure(new RuntimeException( String.format("Failed to find leader for shard %s", shardName), failure)), shardManagerActor); } @Override - public void onUnknownResponse(Object response) { + public void onUnknownResponse(final Object response) { String msg = String.format("Failed to find leader for shard %s: received response: %s", shardName, response); LOG.debug("{}: {}", persistenceId, msg); @@ -1964,7 +1965,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { private final Object response; private final String leaderPath; - WrappedShardResponse(ShardIdentifier shardId, Object response, String leaderPath) { + WrappedShardResponse(final ShardIdentifier shardId, final Object response, final String leaderPath) { this.shardId = shardId; this.response = response; this.leaderPath = leaderPath; @@ -1988,7 +1989,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { private final ShardInformation shardInfo; private final OnShardInitialized onShardInitialized; - ShardNotInitializedTimeout(ShardInformation shardInfo, OnShardInitialized onShardInitialized, ActorRef sender) { + ShardNotInitializedTimeout(final ShardInformation shardInfo, final OnShardInitialized onShardInitialized, + final ActorRef sender) { this.sender = sender; this.shardInfo = shardInfo; this.onShardInitialized = onShardInitialized;