*/
package org.opendaylight.controller.cluster.datastore.shardmanager;
-import static akka.pattern.Patterns.ask;
import static java.util.Objects.requireNonNull;
import akka.actor.ActorRef;
@VisibleForTesting
final ShardPeerAddressResolver peerAddressResolver;
- private EffectiveModelContext schemaContext;
+ private EffectiveModelContext modelContext;
private DatastoreSnapshot restoreFromSnapshot;
@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);
+ if (message instanceof FindPrimary msg) {
+ findPrimary(msg);
+ } else if (message instanceof FindLocalShard msg) {
+ findLocalShard(msg);
+ } else if (message instanceof UpdateSchemaContext msg) {
+ updateSchemaContext(msg);
+ } else if (message instanceof ActorInitialized msg) {
+ onActorInitialized(msg);
+ } else if (message instanceof ClusterEvent.MemberUp msg) {
+ memberUp(msg);
+ } else if (message instanceof ClusterEvent.MemberWeaklyUp msg) {
+ memberWeaklyUp(msg);
+ } else if (message instanceof ClusterEvent.MemberExited msg) {
+ memberExited(msg);
+ } else if (message instanceof ClusterEvent.MemberRemoved msg) {
+ memberRemoved(msg);
+ } else if (message instanceof ClusterEvent.UnreachableMember msg) {
+ memberUnreachable(msg);
+ } else if (message instanceof ClusterEvent.ReachableMember msg) {
+ memberReachable(msg);
+ } else if (message instanceof DatastoreContextFactory msg) {
+ onDatastoreContextFactory(msg);
+ } else if (message instanceof RoleChangeNotification msg) {
+ onRoleChangeNotification(msg);
+ } else if (message instanceof FollowerInitialSyncUpStatus msg) {
+ onFollowerInitialSyncStatus(msg);
+ } else if (message instanceof ShardNotInitializedTimeout msg) {
+ onShardNotInitializedTimeout(msg);
+ } else if (message instanceof ShardLeaderStateChanged msg) {
+ onLeaderStateChanged(msg);
+ } else if (message instanceof SwitchShardBehavior msg) {
+ onSwitchShardBehavior(msg);
+ } else if (message instanceof CreateShard msg) {
+ onCreateShard(msg);
+ } else if (message instanceof AddShardReplica msg) {
+ onAddShardReplica(msg);
} else if (message instanceof ForwardedAddServerReply msg) {
- onAddServerReply(msg.shardInfo, msg.addServerReply, getSender(), msg.leaderPath,
- msg.removeShardOnFailure);
+ onAddServerReply(msg.shardInfo, msg.addServerReply, getSender(), msg.leaderPath, msg.removeShardOnFailure);
} else if (message instanceof ForwardedAddServerFailure msg) {
onAddServerFailure(msg.shardName, msg.failureMessage, msg.failure, getSender(), msg.removeShardOnFailure);
- } else if (message instanceof RemoveShardReplica) {
- onRemoveShardReplica((RemoveShardReplica) 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 RemoveShardReplica msg) {
+ onRemoveShardReplica(msg);
+ } else if (message instanceof WrappedShardResponse msg) {
+ onWrappedShardResponse(msg);
+ } else if (message instanceof GetSnapshot msg) {
+ onGetSnapshot(msg);
+ } else if (message instanceof ServerRemoved msg) {
+ onShardReplicaRemoved(msg);
+ } else if (message instanceof ChangeShardMembersVotingStatus msg) {
+ onChangeShardServersVotingStatus(msg);
+ } else if (message instanceof FlipShardMembersVotingStatus msg) {
+ onFlipShardMembersVotingStatus(msg);
+ } else if (message instanceof SaveSnapshotSuccess msg) {
+ onSaveSnapshotSuccess(msg);
+ } else if (message instanceof SaveSnapshotFailure msg) {
+ LOG.error("{}: SaveSnapshotFailure received for saving snapshot of shards", persistenceId(), msg.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 GetShardRole msg) {
+ onGetShardRole(msg);
+ } else if (message instanceof RunnableMessage msg) {
+ msg.run();
+ } else if (message instanceof RegisterForShardAvailabilityChanges msg) {
+ onRegisterForShardAvailabilityChanges(msg);
+ } else if (message instanceof DeleteSnapshotsFailure msg) {
+ LOG.warn("{}: Failed to delete prior snapshots", persistenceId(), msg.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 if (message instanceof ClusterEvent.MemberEvent msg) {
+ LOG.trace("{}: Received other ClusterEvent.MemberEvent: {}", persistenceId(), msg);
} else {
unknownMessage(message);
}
primaryPath, shardId);
Timeout removeServerTimeout = new Timeout(datastoreContext.getShardLeaderElectionTimeout().duration());
- Future<Object> futureObj = ask(getContext().actorSelection(primaryPath),
+ Future<Object> futureObj = Patterns.ask(getContext().actorSelection(primaryPath),
new RemoveServer(shardId.toString()), removeServerTimeout);
futureObj.onComplete(new OnComplete<>() {
info.setActiveMember(isActiveMember);
localShards.put(info.getShardName(), info);
- if (schemaContext != null) {
- info.setSchemaContext(schemaContext);
+ if (modelContext != null) {
+ info.setSchemaContext(modelContext);
info.setActor(newShardActor(info));
}
}
ShardInformation shardInformation = findShardInformation(leaderStateChanged.getMemberId());
if (shardInformation != null) {
- shardInformation.setLocalDataTree(leaderStateChanged.getLocalShardDataTree());
+ shardInformation.setLocalDataTree(leaderStateChanged.localShardDataTree());
shardInformation.setLeaderVersion(leaderStateChanged.getLeaderPayloadVersion());
if (shardInformation.setLeaderId(leaderStateChanged.getLeaderId())) {
primaryShardInfoCache.remove(shardInformation.getShardName());
return true;
}
- private void onActorInitialized(final Object message) {
- final ActorRef sender = getSender();
-
- if (sender == null) {
- // why is a non-actor sending this message? Just ignore.
- return;
- }
+ private void onActorInitialized(final ActorInitialized message) {
+ final var sender = message.actorRef();
String actorName = sender.path().name();
//find shard name from actor name; actor name is stringified shardId
protected void handleRecover(final Object message) throws Exception {
if (message instanceof RecoveryCompleted) {
onRecoveryCompleted();
- } else if (message instanceof SnapshotOffer) {
- applyShardManagerSnapshot((ShardManagerSnapshot)((SnapshotOffer) message).snapshot());
+ } else if (message instanceof SnapshotOffer msg) {
+ applyShardManagerSnapshot((ShardManagerSnapshot) msg.snapshot());
}
}
*
* @param message the message to send
*/
- private void updateSchemaContext(final Object message) {
- schemaContext = ((UpdateSchemaContext) message).getEffectiveModelContext();
+ private void updateSchemaContext(final UpdateSchemaContext message) {
+ modelContext = message.modelContext();
- LOG.debug("Got updated SchemaContext: # of modules {}", schemaContext.getModules().size());
+ LOG.debug("Got updated SchemaContext: # of modules {}", modelContext.getModules().size());
for (ShardInformation info : localShards.values()) {
- info.setSchemaContext(schemaContext);
+ info.setSchemaContext(modelContext);
if (info.getActor() == null) {
LOG.debug("Creating Shard {}", info.getShardId());
Timeout findPrimaryTimeout = new Timeout(datastoreContextFactory.getBaseDatastoreContext()
.getShardInitializationTimeout().duration().$times(2));
- Future<Object> futureObj = ask(getSelf(), new FindPrimary(shardName, true), findPrimaryTimeout);
+ Future<Object> futureObj = Patterns.ask(getSelf(), new FindPrimary(shardName, true), findPrimaryTimeout);
futureObj.onComplete(new OnComplete<>() {
@Override
public void onComplete(final Throwable failure, final Object response) {
if (failure != null) {
handler.onFailure(failure);
+ } else if (response instanceof RemotePrimaryShardFound msg) {
+ handler.onRemotePrimaryShardFound(msg);
+ } else if (response instanceof LocalPrimaryShardFound msg) {
+ handler.onLocalPrimaryFound(msg);
} else {
- if (response instanceof RemotePrimaryShardFound) {
- handler.onRemotePrimaryShardFound((RemotePrimaryShardFound) response);
- } else if (response instanceof LocalPrimaryShardFound) {
- handler.onLocalPrimaryFound((LocalPrimaryShardFound) response);
- } else {
- handler.onUnknownResponse(response);
- }
+ handler.onUnknownResponse(response);
}
}
}, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client));
}
// Create the localShard
- if (schemaContext == null) {
+ if (modelContext == null) {
LOG.debug("{}: No SchemaContext is available in order to create a local shard instance for {}",
persistenceId(), shardName);
getSender().tell(new Status.Failure(new IllegalStateException(
shardInfo = new ShardInformation(shardName, shardId, getPeerAddresses(shardName), datastoreContext,
Shard.builder(), peerAddressResolver);
shardInfo.setActiveMember(false);
- shardInfo.setSchemaContext(schemaContext);
+ shardInfo.setSchemaContext(modelContext);
localShards.put(shardName, shardInfo);
shardInfo.setActor(newShardActor(shardInfo));
} else {
final Timeout addServerTimeout = new Timeout(shardInfo.getDatastoreContext()
.getShardLeaderElectionTimeout().duration());
- final Future<Object> futureObj = ask(getContext().actorSelection(response.getPrimaryPath()),
+ final Future<Object> futureObj = Patterns.ask(getContext().actorSelection(response.getPrimaryPath()),
new AddServer(shardInfo.getShardId().toString(), localShardAddress, true), addServerTimeout);
futureObj.onComplete(new OnComplete<>() {
private static Exception getServerChangeException(final Class<?> serverChange,
final ServerChangeStatus serverChangeStatus, final String leaderPath, final ShardIdentifier shardId) {
- switch (serverChangeStatus) {
- case TIMEOUT:
- 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()));
- case NO_LEADER:
- return new NoShardLeaderException(shardId);
- case NOT_SUPPORTED:
- return new UnsupportedOperationException(String.format("%s request is not supported for shard %s",
- serverChange.getSimpleName(), shardId.getShardName()));
- default :
- return new RuntimeException(String.format("%s request to leader %s for shard %s failed with status %s",
- serverChange.getSimpleName(), leaderPath, shardId.getShardName(), serverChangeStatus));
- }
+ return switch (serverChangeStatus) {
+ case TIMEOUT -> new TimeoutException("""
+ 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""".formatted(leaderPath, shardId.getShardName()));
+ case NO_LEADER -> new NoShardLeaderException(shardId);
+ case NOT_SUPPORTED -> new UnsupportedOperationException(
+ "%s request is not supported for shard %s".formatted(
+ serverChange.getSimpleName(), shardId.getShardName()));
+ default -> new RuntimeException("%s request to leader %s for shard %s failed with status %s".formatted(
+ serverChange.getSimpleName(), leaderPath, shardId.getShardName(), serverChangeStatus));
+ };
}
private void onRemoveShardReplica(final RemoveShardReplica shardReplicaMsg) {
ActorRef sender = getSender();
final String shardName = flipMembersVotingStatus.getShardName();
findLocalShard(shardName, sender, localShardFound -> {
- Future<Object> future = ask(localShardFound.getPath(), GetOnDemandRaftState.INSTANCE,
+ Future<Object> future = Patterns.ask(localShardFound.getPath(), GetOnDemandRaftState.INSTANCE,
Timeout.apply(30, TimeUnit.SECONDS));
future.onComplete(new OnComplete<>() {
Timeout findLocalTimeout = new Timeout(datastoreContextFactory.getBaseDatastoreContext()
.getShardInitializationTimeout().duration().$times(2));
- Future<Object> futureObj = ask(getSelf(), new FindLocalShard(shardName, true), findLocalTimeout);
+ Future<Object> futureObj = Patterns.ask(getSelf(), new FindLocalShard(shardName, true), findLocalTimeout);
futureObj.onComplete(new OnComplete<>() {
@Override
public void onComplete(final Throwable failure, final Object response) {
if (failure != null) {
LOG.debug("{}: Received failure from FindLocalShard for shard {}", persistenceId, shardName,
- failure);
+ failure);
sender.tell(new Status.Failure(new RuntimeException(
- String.format("Failed to find local shard %s", shardName), failure)), self());
+ String.format("Failed to find local shard %s", shardName), failure)), self());
+ } if (response instanceof LocalShardFound msg) {
+ getSelf().tell((RunnableMessage) () -> onLocalShardFound.accept(msg), sender);
+ } else if (response instanceof LocalShardNotFound) {
+ LOG.debug("{}: Local shard {} does not exist", persistenceId, shardName);
+ sender.tell(new Status.Failure(new IllegalArgumentException(
+ String.format("Local shard %s does not exist", shardName))), self());
} else {
- if (response instanceof LocalShardFound) {
- getSelf().tell((RunnableMessage) () -> onLocalShardFound.accept((LocalShardFound) response),
- sender);
- } else if (response instanceof LocalShardNotFound) {
- LOG.debug("{}: Local shard {} does not exist", persistenceId, shardName);
- sender.tell(new Status.Failure(new IllegalArgumentException(
- String.format("Local shard %s does not exist", shardName))), self());
- } else {
- LOG.debug("{}: Failed to find local shard {}: received response: {}", persistenceId, shardName,
- response);
- sender.tell(new Status.Failure(response instanceof Throwable ? (Throwable) response
- : new RuntimeException(
- String.format("Failed to find local shard %s: received response: %s", shardName,
- response))), self());
- }
+ LOG.debug("{}: Failed to find local shard {}: received response: {}", persistenceId, shardName,
+ response);
+ sender.tell(new Status.Failure(response instanceof Throwable throwable ? throwable
+ : new RuntimeException(String.format("Failed to find local shard %s: received response: %s",
+ shardName, response))), self());
}
}
}, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client));
changeServersVotingStatus, shardActorRef.path());
Timeout timeout = new Timeout(datastoreContext.getShardLeaderElectionTimeout().duration().$times(2));
- Future<Object> futureObj = ask(shardActorRef, changeServersVotingStatus, timeout);
+ Future<Object> futureObj = Patterns.ask(shardActorRef, changeServersVotingStatus, timeout);
futureObj.onComplete(new OnComplete<>() {
@Override
Code Review / controller.git / blobdiff