import akka.dispatch.OnComplete;
import akka.pattern.Patterns;
import akka.util.Timeout;
-import com.google.common.base.Function;
import com.google.common.base.Strings;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.AbstractMap.SimpleEntry;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import org.apache.commons.lang3.SerializationUtils;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
this.serializer = serializer;
this.makeLeaderLocalTimeout =
- new Timeout(configDataStore.getActorContext().getDatastoreContext()
+ new Timeout(configDataStore.getActorUtils().getDatastoreContext()
.getShardLeaderElectionTimeout().duration().$times(2));
}
return newFailedRpcResultFuture("A valid DataStoreType must be specified");
}
- ActorContext actorContext = dataStoreType == DataStoreType.Config
- ? configDataStore.getActorContext()
- : operDataStore.getActorContext();
+ ActorUtils actorUtils = dataStoreType == DataStoreType.Config
+ ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
- actorContext.getCurrentMemberName().getName(), shardName, dataStoreType);
+ actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
final scala.concurrent.Future<ActorRef> localShardReply =
- actorContext.findLocalShardAsync(shardName);
+ actorUtils.findLocalShardAsync(shardName);
final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
localShardReply.onComplete(new OnComplete<ActorRef>() {
@Override
- public void onComplete(final Throwable failure, final ActorRef actorRef) throws Throwable {
+ public void onComplete(final Throwable failure, final ActorRef actorRef) {
if (failure != null) {
LOG.warn("No local shard found for {} datastoreType {} - Cannot request leadership transfer to"
- + " local shard.", shardName, failure);
+ + " local shard.", shardName, dataStoreType, failure);
makeLeaderLocalAsk.failure(failure);
} else {
makeLeaderLocalAsk
- .completeWith(actorContext
+ .completeWith(actorUtils
.executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
@Override
- public void onComplete(final Throwable failure, final Object success) throws Throwable {
+ public void onComplete(final Throwable failure, final Object success) {
if (failure != null) {
LOG.error("Leadership transfer failed for shard {}.", shardName, failure);
future.set(RpcResultBuilder.<MakeLeaderLocalOutput>failed().withError(ErrorType.APPLICATION,
LOG.debug("Leadership transfer complete");
future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
return future;
}
LOG.info("Adding replicas for all shards");
final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
- Function<String, Object> messageSupplier = AddShardReplica::new;
- sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
- sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
+ sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
+ sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
return waitForShardResults(shardResultData, shardResults ->
new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
final Function<String, Object> messageSupplier) {
- ActorContext actorContext = dataStoreType == DataStoreType.Config ? configDataStore.getActorContext()
- : operDataStore.getActorContext();
- Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
+ ActorUtils actorUtils = dataStoreType == DataStoreType.Config ? configDataStore.getActorUtils()
+ : operDataStore.getActorUtils();
+ Set<String> allShardNames = actorUtils.getConfiguration().getAllShardNames();
- LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
+ LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
for (String shardName: allShardNames) {
- ListenableFuture<T> future = this.ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
+ ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
SHARD_MGR_TIMEOUT);
shardResultData.add(new SimpleEntry<>(future,
new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
Timeout timeout = SHARD_MGR_TIMEOUT;
- ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
- ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
+ ListenableFuture<T> configFuture = ask(configDataStore.getActorUtils().getShardManager(), message, timeout);
+ ListenableFuture<T> operFuture = ask(operDataStore.getActorUtils().getShardManager(), message, timeout);
return Futures.allAsList(configFuture, operFuture);
}
private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
ActorRef shardManager = dataStoreType == DataStoreType.Config
- ? configDataStore.getActorContext().getShardManager()
- : operDataStore.getActorContext().getShardManager();
+ ? configDataStore.getActorUtils().getShardManager()
+ : operDataStore.getActorUtils().getShardManager();
return ask(shardManager, message, SHARD_MGR_TIMEOUT);
}
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
@SuppressWarnings("checkstyle:IllegalCatch")
private static void saveSnapshotsToFile(final DatastoreSnapshotList snapshots, final String fileName,
final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture) {
onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
}
+ @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
private static <T> void onMessageFailure(final String msg, final SettableFuture<RpcResult<T>> returnFuture,
final Throwable failure) {
- LOG.error(msg, failure);
+ LOG.error("{}", msg, failure);
returnFuture.set(ClusterAdminRpcService.<T>newFailedRpcResultBuilder(String.format("%s: %s", msg,
failure.getMessage())).build());
}
returnFuture.set(resp);
}
}
- }, configDataStore.getActorContext().getClientDispatcher());
+ }, configDataStore.getActorUtils().getClientDispatcher());
return returnFuture;
}
Code Review / controller.git / blobdiff