package org.opendaylight.controller.cluster.datastore.shardmanager;
import static akka.pattern.Patterns.ask;
+import static java.util.Objects.requireNonNull;
import akka.actor.ActorRef;
import akka.actor.Address;
import akka.persistence.SnapshotSelectionCriteria;
import akka.util.Timeout;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.SettableFuture;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
-import org.opendaylight.controller.cluster.datastore.AbstractDataStore;
+import org.opendaylight.controller.cluster.common.actor.Dispatchers;
import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
import org.opendaylight.controller.cluster.datastore.persisted.ShardManagerSnapshot;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.datastore.utils.CompositeOnComplete;
-import org.opendaylight.controller.cluster.datastore.utils.Dispatchers;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListenerReply;
import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus;
import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
-import org.opendaylight.controller.cluster.sharding.PrefixedShardConfigUpdateHandler;
-import org.opendaylight.controller.cluster.sharding.messages.InitConfigListener;
-import org.opendaylight.controller.cluster.sharding.messages.PrefixShardCreated;
-import org.opendaylight.controller.cluster.sharding.messages.PrefixShardRemoved;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.concepts.Registration;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;
-import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;
/**
// Stores a mapping between a shard name and it's corresponding information
// Shard names look like inventory, topology etc and are as specified in
// configuration
- private final Map<String, ShardInformation> localShards = new HashMap<>();
+ @VisibleForTesting
+ final Map<String, ShardInformation> localShards = new HashMap<>();
// The type of a ShardManager reflects the type of the datastore itself
// A data store could be of type config/operational
private final Configuration configuration;
- private final String shardDispatcherPath;
+ @VisibleForTesting
+ final String shardDispatcherPath;
private final ShardManagerInfo shardManagerMBean;
private DatastoreContextFactory datastoreContextFactory;
- private final CountDownLatch waitTillReadyCountdownLatch;
+ private final SettableFuture<Void> readinessFuture;
private final PrimaryShardInfoFutureCache primaryShardInfoCache;
- private final ShardPeerAddressResolver peerAddressResolver;
+ @VisibleForTesting
+ final ShardPeerAddressResolver peerAddressResolver;
- private SchemaContext schemaContext;
+ private EffectiveModelContext schemaContext;
private DatastoreSnapshot restoreFromSnapshot;
private final Map<String, CompositeOnComplete<Boolean>> shardActorsStopping = new HashMap<>();
- private final String persistenceId;
- private final AbstractDataStore dataStore;
+ private final Set<Consumer<String>> shardAvailabilityCallbacks = new HashSet<>();
- private ListenerRegistration<DOMDataTreeChangeListener> configListenerReg = null;
- private PrefixedShardConfigUpdateHandler configUpdateHandler;
+ private final String persistenceId;
ShardManager(final AbstractShardManagerCreator<?> builder) {
this.cluster = builder.getCluster();
this.type = datastoreContextFactory.getBaseDatastoreContext().getDataStoreName();
this.shardDispatcherPath =
new Dispatchers(context().system().dispatchers()).getDispatcherPath(Dispatchers.DispatcherType.Shard);
- this.waitTillReadyCountdownLatch = builder.getWaitTillReadyCountDownLatch();
+ this.readinessFuture = builder.getReadinessFuture();
this.primaryShardInfoCache = builder.getPrimaryShardInfoCache();
this.restoreFromSnapshot = builder.getRestoreFromSnapshot();
"shard-manager-" + this.type,
datastoreContextFactory.getBaseDatastoreContext().getDataStoreMXBeanType());
shardManagerMBean.registerMBean();
-
- dataStore = builder.getDistributedDataStore();
}
@Override
LOG.info("Stopping ShardManager {}", persistenceId());
shardManagerMBean.unregisterMBean();
-
- if (configListenerReg != null) {
- configListenerReg.close();
- configListenerReg = null;
- }
}
@Override
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,
} else if (message instanceof WrappedShardResponse) {
onWrappedShardResponse((WrappedShardResponse) message);
} else if (message instanceof GetSnapshot) {
- onGetSnapshot();
+ onGetSnapshot((GetSnapshot) message);
} else if (message instanceof ServerRemoved) {
onShardReplicaRemoved((ServerRemoved) message);
} else if (message instanceof ChangeShardMembersVotingStatus) {
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(), message);
+ LOG.debug("{}: Successfully deleted prior snapshots", persistenceId());
} else if (message instanceof RegisterRoleChangeListenerReply) {
LOG.trace("{}: Received RegisterRoleChangeListenerReply", persistenceId());
} else if (message instanceof ClusterEvent.MemberEvent) {
}
}
+ private void onRegisterForShardAvailabilityChanges(final RegisterForShardAvailabilityChanges message) {
+ LOG.debug("{}: onRegisterForShardAvailabilityChanges: {}", persistenceId(), message);
+
+ final Consumer<String> callback = message.getCallback();
+ shardAvailabilityCallbacks.add(callback);
+
+ getSender().tell(new Status.Success((Registration)
+ () -> executeInSelf(() -> shardAvailabilityCallbacks.remove(callback))), self());
+ }
+
private void onGetShardRole(final GetShardRole message) {
LOG.debug("{}: onGetShardRole for shard: {}", persistenceId(), message.getName());
getSender().tell(new GetShardRoleReply(shardInformation.getRole()), ActorRef.noSender());
}
- private void onInitConfigListener() {
- LOG.debug("{}: Initializing config listener on {}", persistenceId(), cluster.getCurrentMemberName());
-
- final org.opendaylight.mdsal.common.api.LogicalDatastoreType type =
- org.opendaylight.mdsal.common.api.LogicalDatastoreType
- .valueOf(datastoreContextFactory.getBaseDatastoreContext().getLogicalStoreType().name());
-
- if (configUpdateHandler != null) {
- configUpdateHandler.close();
- }
-
- configUpdateHandler = new PrefixedShardConfigUpdateHandler(self(), cluster.getCurrentMemberName());
- configUpdateHandler.initListener(dataStore, type);
- }
-
- private void onShutDown() {
+ void onShutDown() {
List<Future<Boolean>> stopFutures = new ArrayList<>(localShards.size());
for (ShardInformation info : localShards.values()) {
if (info.getActor() != null) {
}
}
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
private void removePrefixShardReplica(final RemovePrefixShardReplica contextMessage, final String shardName,
final String primaryPath, final ActorRef sender) {
if (isShardReplicaOperationInProgress(shardName, sender)) {
Future<Object> futureObj = ask(getContext().actorSelection(primaryPath),
new RemoveServer(shardId.toString()), removeServerTimeout);
- futureObj.onComplete(new OnComplete<Object>() {
+ futureObj.onComplete(new OnComplete<>() {
@Override
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",
- primaryPath, shardName);
- LOG.debug("{}: {}", persistenceId(), msg, failure);
+ LOG.debug("{}: RemoveServer request to leader {} for shard {} failed", persistenceId(), primaryPath,
+ shardName, failure);
// FAILURE
- sender.tell(new Status.Failure(new RuntimeException(msg, failure)), self());
+ sender.tell(new Status.Failure(new RuntimeException(
+ String.format("RemoveServer request to leader %s for shard %s failed", primaryPath, shardName),
+ failure)), self());
} else {
// SUCCESS
self().tell(new WrappedShardResponse(shardId, response, primaryPath), sender);
}, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client));
}
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
private void removeShardReplica(final RemoveShardReplica contextMessage, final String shardName,
final String primaryPath, final ActorRef sender) {
if (isShardReplicaOperationInProgress(shardName, sender)) {
Future<Object> futureObj = ask(getContext().actorSelection(primaryPath),
new RemoveServer(shardId.toString()), removeServerTimeout);
- futureObj.onComplete(new OnComplete<Object>() {
+ futureObj.onComplete(new OnComplete<>() {
@Override
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",
- primaryPath, shardName);
-
- LOG.debug("{}: {}", persistenceId(), msg, failure);
+ LOG.debug("{}: RemoveServer request to leader {} for shard {} failed", persistenceId(), primaryPath,
+ shardName, failure);
// FAILURE
- sender.tell(new Status.Failure(new RuntimeException(msg, failure)), self());
+ sender.tell(new Status.Failure(new RuntimeException(
+ String.format("RemoveServer request to leader %s for shard %s failed", primaryPath, shardName),
+ failure)), self());
} else {
// SUCCESS
self().tell(new WrappedShardResponse(shardId, response, primaryPath), sender);
final Future<Boolean> stopFuture = Patterns.gracefulStop(shardActor,
FiniteDuration.apply(timeoutInMS, TimeUnit.MILLISECONDS), Shutdown.INSTANCE);
- final CompositeOnComplete<Boolean> onComplete = new CompositeOnComplete<Boolean>() {
+ final CompositeOnComplete<Boolean> onComplete = new CompositeOnComplete<>() {
@Override
public void onComplete(final Throwable failure, final Boolean result) {
if (failure == null) {
}
self().tell((RunnableMessage) () -> {
+ // At any rate, invalidate primaryShardInfo cache
+ primaryShardInfoCache.remove(shardName);
+
shardActorsStopping.remove(shardName);
notifyOnCompleteTasks(failure, result);
}, ActorRef.noSender());
persistShardList();
}
- private void onGetSnapshot() {
+ private void onGetSnapshot(final GetSnapshot getSnapshot) {
LOG.debug("{}: onGetSnapshot", persistenceId());
List<String> notInitialized = null;
datastoreContextFactory.getBaseDatastoreContext().getShardInitializationTimeout().duration()));
for (ShardInformation shardInfo: localShards.values()) {
- shardInfo.getActor().tell(GetSnapshot.INSTANCE, replyActor);
+ shardInfo.getActor().tell(getSnapshot, replyActor);
}
}
}
}
- private void onPrefixShardCreated(final PrefixShardCreated message) {
- LOG.debug("{}: onPrefixShardCreated: {}", persistenceId(), message);
-
- final PrefixShardConfiguration config = message.getConfiguration();
- final ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(),
- ClusterUtils.getCleanShardName(config.getPrefix().getRootIdentifier()));
- final String shardName = shardId.getShardName();
-
- if (isPreviousShardActorStopInProgress(shardName, message)) {
- return;
- }
-
- if (localShards.containsKey(shardName)) {
- LOG.debug("{}: Received create for an already existing shard {}", persistenceId(), shardName);
- final PrefixShardConfiguration existing =
- configuration.getAllPrefixShardConfigurations().get(config.getPrefix());
-
- if (existing != null && existing.equals(config)) {
- // we don't have to do nothing here
- return;
- }
- }
-
- doCreatePrefixShard(config, shardId, shardName);
- }
-
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
private boolean isPreviousShardActorStopInProgress(final String shardName, final Object messageToDefer) {
final CompositeOnComplete<Boolean> stopOnComplete = shardActorsStopping.get(shardName);
if (stopOnComplete == null) {
return true;
}
- private void doCreatePrefixShard(final PrefixShardConfiguration config, final ShardIdentifier shardId,
- final String shardName) {
- configuration.addPrefixShardConfiguration(config);
-
- final Builder builder = newShardDatastoreContextBuilder(shardName);
- builder.logicalStoreType(LogicalDatastoreType.valueOf(config.getPrefix().getDatastoreType().name()))
- .storeRoot(config.getPrefix().getRootIdentifier());
- DatastoreContext shardDatastoreContext = builder.build();
-
- final Map<String, String> peerAddresses = getPeerAddresses(shardName);
- final boolean isActiveMember = true;
-
- LOG.debug("{} doCreatePrefixShard: shardId: {}, memberNames: {}, peerAddresses: {}, isActiveMember: {}",
- persistenceId(), shardId, config.getShardMemberNames(), peerAddresses, isActiveMember);
-
- final ShardInformation info = new ShardInformation(shardName, shardId, peerAddresses,
- shardDatastoreContext, Shard.builder(), peerAddressResolver);
- info.setActiveMember(isActiveMember);
- localShards.put(info.getShardName(), info);
-
- if (schemaContext != null) {
- info.setSchemaContext(schemaContext);
- info.setActor(newShardActor(info));
- }
- }
-
- private void onPrefixShardRemoved(final PrefixShardRemoved message) {
- LOG.debug("{}: onPrefixShardRemoved : {}", persistenceId(), message);
-
- final DOMDataTreeIdentifier prefix = message.getPrefix();
- final ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(),
- ClusterUtils.getCleanShardName(prefix.getRootIdentifier()));
-
- configuration.removePrefixShardConfiguration(prefix);
- removeShard(shardId);
- }
-
private void doCreateShard(final CreateShard createShard) {
final ModuleShardConfiguration moduleShardConfig = createShard.getModuleShardConfig();
final String shardName = moduleShardConfig.getShardName();
private void checkReady() {
if (isReadyWithLeaderId()) {
- LOG.info("{}: All Shards are ready - data store {} is ready, available count is {}",
- persistenceId(), type, waitTillReadyCountdownLatch.getCount());
-
- waitTillReadyCountdownLatch.countDown();
+ LOG.info("{}: All Shards are ready - data store {} is ready", persistenceId(), type);
+ readinessFuture.set(null);
}
}
shardInformation.setLeaderVersion(leaderStateChanged.getLeaderPayloadVersion());
if (shardInformation.setLeaderId(leaderStateChanged.getLeaderId())) {
primaryShardInfoCache.remove(shardInformation.getShardName());
+
+ notifyShardAvailabilityCallbacks(shardInformation);
}
checkReady();
}
}
+ private void notifyShardAvailabilityCallbacks(final ShardInformation shardInformation) {
+ shardAvailabilityCallbacks.forEach(callback -> callback.accept(shardInformation.getShardName()));
+ }
+
private void onShardNotInitializedTimeout(final ShardNotInitializedTimeout message) {
ShardInformation shardInfo = message.getShardInfo();
message.getSender().tell(createNotInitializedException(shardInfo.getShardId()), getSelf());
} else {
LOG.debug("{}: Returning NoShardLeaderException for shard {}", persistenceId(), shardInfo.getShardName());
- message.getSender().tell(createNoShardLeaderException(shardInfo.getShardId()), getSelf());
+ message.getSender().tell(new NoShardLeaderException(shardInfo.getShardId()), getSelf());
}
}
final ActorRef sender = getSender();
if (sender == null) {
- return; //why is a non-actor sending this message? Just ignore.
+ // why is a non-actor sending this message? Just ignore.
+ return;
}
String actorName = sender.path().name();
try {
shardId = ShardIdentifier.fromShardIdString(actorName);
} catch (IllegalArgumentException e) {
- LOG.debug("{}: ignoring actor {}", actorName, e);
+ LOG.debug("{}: ignoring actor {}", persistenceId, actorName, e);
return;
}
}
LOG.debug("{}: Scheduling {} ms timer to wait for shard {}", persistenceId(), timeout.toMillis(),
- shardInformation.getShardName());
+ shardInformation);
Cancellable timeoutSchedule = getContext().system().scheduler().scheduleOnce(
timeout, getSelf(),
} else {
LOG.debug("{}: Returning NoShardLeaderException for shard {}", persistenceId(),
shardInformation.getShardName());
- getSender().tell(createNoShardLeaderException(shardInformation.getShardId()), getSelf());
+ getSender().tell(new NoShardLeaderException(shardInformation.getShardId()), getSelf());
}
return;
getSender().tell(messageSupplier.get(), getSelf());
}
- private static NoShardLeaderException createNoShardLeaderException(final ShardIdentifier shardId) {
- return new NoShardLeaderException(null, shardId.toString());
- }
-
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));
}
private void markMemberUnavailable(final MemberName memberName) {
- final String memberStr = memberName.getName();
for (ShardInformation info : localShards.values()) {
String leaderId = info.getLeaderId();
- // XXX: why are we using String#contains() here?
- if (leaderId != null && leaderId.contains(memberStr)) {
+ if (leaderId != null && ShardIdentifier.fromShardIdString(leaderId).getMemberName().equals(memberName)) {
LOG.debug("Marking Leader {} as unavailable.", leaderId);
info.setLeaderAvailable(false);
primaryShardInfoCache.remove(info.getShardName());
+
+ notifyShardAvailabilityCallbacks(info);
}
info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
}
private void markMemberAvailable(final MemberName memberName) {
- final String memberStr = memberName.getName();
for (ShardInformation info : localShards.values()) {
String leaderId = info.getLeaderId();
- // XXX: why are we using String#contains() here?
- if (leaderId != null && leaderId.contains(memberStr)) {
+ if (leaderId != null && ShardIdentifier.fromShardIdString(leaderId).getMemberName().equals(memberName)) {
LOG.debug("Marking Leader {} as available.", leaderId);
info.setLeaderAvailable(true);
}
* @param message the message to send
*/
private void updateSchemaContext(final Object message) {
- schemaContext = ((UpdateSchemaContext) message).getSchemaContext();
+ schemaContext = ((UpdateSchemaContext) message).getEffectiveModelContext();
- LOG.debug("Got updated SchemaContext: # of modules {}", schemaContext.getAllModuleIdentifiers().size());
+ LOG.debug("Got updated SchemaContext: # of modules {}", schemaContext.getModules().size());
for (ShardInformation info : localShards.values()) {
info.setSchemaContext(schemaContext);
if (info.getActor() == null) {
LOG.debug("Creating Shard {}", info.getShardId());
info.setActor(newShardActor(info));
+ // Update peer address for every existing peer memeber to avoid missing sending
+ // PeerAddressResolved and PeerUp to this shard while UpdateSchemaContext comes after MemberUp.
+ String shardName = info.getShardName();
+ for (MemberName memberName : peerAddressResolver.getPeerMembers()) {
+ String peerId = getShardIdentifier(memberName, shardName).toString() ;
+ String peerAddress = peerAddressResolver.getShardActorAddress(shardName, memberName);
+ info.updatePeerAddress(peerId, peerAddress, getSelf());
+ info.peerUp(memberName, peerId, getSelf());
+ LOG.debug("{}: updated peer {} on member {} with address {} on shard {} whose actor address is {}",
+ persistenceId(), peerId, memberName, peerAddress, info.getShardId(), info.getActor());
+ }
} else {
info.getActor().tell(message, getSelf());
}
.getShardInitializationTimeout().duration().$times(2));
Future<Object> futureObj = ask(getSelf(), new FindPrimary(shardName, true), findPrimaryTimeout);
- futureObj.onComplete(new OnComplete<Object>() {
+ futureObj.onComplete(new OnComplete<>() {
@Override
public void onComplete(final Throwable failure, final Object response) {
if (failure != null) {
}
}
- restoreFromSnapshot = null; // null out to GC
+ // null out to GC
+ restoreFromSnapshot = null;
for (String shardName : memberShardNames) {
ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
LOG.debug("{}: Creating local shard: {}", persistenceId(), shardId);
Map<String, String> peerAddresses = getPeerAddresses(shardName);
- localShards.put(shardName, new ShardInformation(shardName, shardId, peerAddresses,
- newShardDatastoreContext(shardName), Shard.builder().restoreFromSnapshot(
- shardSnapshots.get(shardName)), peerAddressResolver));
+ localShards.put(shardName, createShardInfoFor(shardName, shardId, peerAddresses,
+ newShardDatastoreContext(shardName), shardSnapshots));
}
}
+ @VisibleForTesting
+ ShardInformation createShardInfoFor(final String shardName, final ShardIdentifier shardId,
+ final Map<String, String> peerAddresses,
+ final DatastoreContext datastoreContext,
+ final Map<String, DatastoreSnapshot.ShardSnapshot> shardSnapshots) {
+ return new ShardInformation(shardName, shardId, peerAddresses,
+ datastoreContext, Shard.builder().restoreFromSnapshot(shardSnapshots.get(shardName)),
+ peerAddressResolver);
+ }
+
/**
* Given the name of the shard find the addresses of all it's peers.
*
* @param shardName the shard name
*/
- private Map<String, String> getPeerAddresses(final String shardName) {
+ Map<String, String> getPeerAddresses(final String shardName) {
final Collection<MemberName> members = configuration.getMembersFromShardName(shardName);
return getPeerAddresses(shardName, members);
}
@Override
public SupervisorStrategy supervisorStrategy() {
- return new OneForOneStrategy(10, Duration.create("1 minute"),
+ return new OneForOneStrategy(10, FiniteDuration.create(1, TimeUnit.MINUTES),
(Function<Throwable, Directive>) t -> {
LOG.warn("Supervisor Strategy caught unexpected exception - resuming", t);
return SupervisorStrategy.resume();
private boolean isShardReplicaOperationInProgress(final String shardName, final ActorRef sender) {
if (shardReplicaOperationsInProgress.contains(shardName)) {
- String msg = String.format("A shard replica operation for %s is already in progress", shardName);
- LOG.debug("{}: {}", persistenceId(), msg);
- sender.tell(new Status.Failure(new IllegalStateException(msg)), getSelf());
+ LOG.debug("{}: A shard replica operation for {} is already in progress", persistenceId(), shardName);
+ sender.tell(new Status.Failure(new IllegalStateException(
+ String.format("A shard replica operation for %s is already in progress", shardName))), getSelf());
return true;
}
// Create the localShard
if (schemaContext == null) {
- String msg = String.format(
- "No SchemaContext is available in order to create a local shard instance for %s", shardName);
- LOG.debug("{}: {}", persistenceId(), msg);
- getSender().tell(new Status.Failure(new IllegalStateException(msg)), getSelf());
+ LOG.debug("{}: No SchemaContext is available in order to create a local shard instance for {}",
+ persistenceId(), shardName);
+ getSender().tell(new Status.Failure(new IllegalStateException(
+ "No SchemaContext is available in order to create a local shard instance for " + shardName)),
+ getSelf());
return;
}
// verify the shard with the specified name is present in the cluster configuration
if (!this.configuration.isShardConfigured(shardName)) {
- String msg = String.format("No module configuration exists for shard %s", shardName);
- LOG.debug("{}: {}", persistenceId(), msg);
- getSender().tell(new Status.Failure(new IllegalArgumentException(msg)), getSelf());
+ LOG.debug("{}: No module configuration exists for shard {}", persistenceId(), shardName);
+ getSender().tell(new Status.Failure(new IllegalArgumentException(
+ "No module configuration exists for shard " + shardName)), getSelf());
return;
}
// Create the localShard
if (schemaContext == null) {
- String msg = String.format(
- "No SchemaContext is available in order to create a local shard instance for %s", shardName);
- LOG.debug("{}: {}", persistenceId(), msg);
- getSender().tell(new Status.Failure(new IllegalStateException(msg)), getSelf());
+ LOG.debug("{}: No SchemaContext is available in order to create a local shard instance for {}",
+ persistenceId(), shardName);
+ getSender().tell(new Status.Failure(new IllegalStateException(
+ "No SchemaContext is available in order to create a local shard instance for " + shardName)),
+ getSelf());
return;
}
});
}
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
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());
+ LOG.debug("{}: Local shard {} already exists", persistenceId(), shardName);
+ sender.tell(new Status.Failure(new AlreadyExistsException(
+ String.format("Local shard %s already exists", shardName))), getSelf());
}
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
private void addPrefixShard(final String shardName, final YangInstanceIdentifier shardPrefix,
final RemotePrimaryShardFound response, final ActorRef sender) {
if (isShardReplicaOperationInProgress(shardName, sender)) {
execAddShard(shardName, shardInfo, response, removeShardOnFailure, sender);
}
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
private void addShard(final String shardName, final RemotePrimaryShardFound response, final ActorRef sender) {
if (isShardReplicaOperationInProgress(shardName, sender)) {
return;
final Future<Object> futureObj = ask(getContext().actorSelection(response.getPrimaryPath()),
new AddServer(shardInfo.getShardId().toString(), localShardAddress, true), addServerTimeout);
- futureObj.onComplete(new OnComplete<Object>() {
+ futureObj.onComplete(new OnComplete<>() {
@Override
public void onComplete(final Throwable failure, final Object addServerResponse) {
if (failure != null) {
+ "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 createNoShardLeaderException(shardId);
+ return new NoShardLeaderException(shardId);
case NOT_SUPPORTED:
return new UnsupportedOperationException(String.format("%s request is not supported for shard %s",
serverChange.getSimpleName(), shardId.getShardName()));
Future<Object> future = ask(localShardFound.getPath(), GetOnDemandRaftState.INSTANCE,
Timeout.apply(30, TimeUnit.SECONDS));
- future.onComplete(new OnComplete<Object>() {
+ future.onComplete(new OnComplete<>() {
@Override
public void onComplete(final Throwable failure, final Object response) {
if (failure != null) {
.getShardInitializationTimeout().duration().$times(2));
Future<Object> futureObj = ask(getSelf(), new FindLocalShard(shardName, true), findLocalTimeout);
- futureObj.onComplete(new OnComplete<Object>() {
+ futureObj.onComplete(new OnComplete<>() {
@Override
public void onComplete(final Throwable failure, final Object response) {
if (failure != null) {
getSelf().tell((RunnableMessage) () -> onLocalShardFound.accept((LocalShardFound) response),
sender);
} else if (response instanceof LocalShardNotFound) {
- String msg = String.format("Local shard %s does not exist", shardName);
- LOG.debug("{}: {}", persistenceId, msg);
- sender.tell(new Status.Failure(new IllegalArgumentException(msg)), self());
+ 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 {
- String msg = String.format("Failed to find local shard %s: received response: %s",
- shardName, response);
- LOG.debug("{}: {}", persistenceId, msg);
- sender.tell(new Status.Failure(response instanceof Throwable ? (Throwable) response :
- new RuntimeException(msg)), self());
+ 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());
}
}
}
Timeout timeout = new Timeout(datastoreContext.getShardLeaderElectionTimeout().duration().$times(2));
Future<Object> futureObj = ask(shardActorRef, changeServersVotingStatus, timeout);
- futureObj.onComplete(new OnComplete<Object>() {
+ futureObj.onComplete(new OnComplete<>() {
@Override
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",
- shardActorRef.path());
- LOG.debug("{}: {}", persistenceId(), msg, failure);
- sender.tell(new Status.Failure(new RuntimeException(msg, failure)), self());
+ LOG.debug("{}: ChangeServersVotingStatus request to local shard {} failed", persistenceId(),
+ shardActorRef.path(), failure);
+ sender.tell(new Status.Failure(new RuntimeException(
+ String.format("ChangeServersVotingStatus request to local shard %s failed",
+ shardActorRef.path()), failure)), self());
} else {
LOG.debug("{}: Received {} from local shard {}", persistenceId(), response, shardActorRef.path());
*/
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);
- this.shardManagerActor = Preconditions.checkNotNull(shardManagerActor);
+ this.targetActor = requireNonNull(targetActor);
+ this.shardName = requireNonNull(shardName);
+ this.persistenceId = requireNonNull(persistenceId);
+ this.shardManagerActor = requireNonNull(shardManagerActor);
}
public ActorRef getTargetActor() {
@Override
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);
- targetActor.tell(new Status.Failure(response instanceof Throwable ? (Throwable) response :
- new RuntimeException(msg)), shardManagerActor);
+ LOG.debug("{}: Failed to find leader for shard {}: received response: {}", persistenceId, shardName,
+ response);
+ targetActor.tell(new Status.Failure(response instanceof Throwable ? (Throwable) response
+ : new RuntimeException(String.format("Failed to find leader for shard %s: received response: %s",
+ shardName, response))), shardManagerActor);
}
}
Code Review / controller.git / blobdiff