X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardManager.java;h=e51d49bff2aff8b6380081e0e772765d172246b0;hb=67d58d1ab50f3c3bbe19a96fb6f0d9d94211829f;hp=6162a0327ca6ab229be75b4f8b8c2c994dcb253a;hpb=971b179000ef1cc56699de35061cf6f97d4cf36f;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java index 6162a0327c..e51d49bff2 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java @@ -17,7 +17,9 @@ import akka.actor.SupervisorStrategy; import akka.cluster.ClusterEvent; import akka.japi.Creator; import akka.japi.Function; + import com.google.common.base.Preconditions; + import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier; import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier; import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfo; @@ -30,6 +32,8 @@ import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolve import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound; import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; + +import org.opendaylight.controller.cluster.datastore.utils.ActorContext; import scala.concurrent.duration.Duration; import java.util.ArrayList; @@ -68,15 +72,19 @@ public class ShardManager extends AbstractUntypedActor { private ShardManagerInfoMBean mBean; + private final DatastoreContext datastoreContext; + /** * @param type defines the kind of data that goes into shards created by this shard manager. Examples of type would be * configuration or operational */ - private ShardManager(String type, ClusterWrapper cluster, Configuration configuration) { + private ShardManager(String type, ClusterWrapper cluster, Configuration configuration, + DatastoreContext datastoreContext) { this.type = Preconditions.checkNotNull(type, "type should not be null"); this.cluster = Preconditions.checkNotNull(cluster, "cluster should not be null"); this.configuration = Preconditions.checkNotNull(configuration, "configuration should not be null"); + this.datastoreContext = datastoreContext; // Subscribe this actor to cluster member events cluster.subscribeToMemberEvents(getSelf()); @@ -88,22 +96,16 @@ public class ShardManager extends AbstractUntypedActor { public static Props props(final String type, final ClusterWrapper cluster, - final Configuration configuration) { + final Configuration configuration, + final DatastoreContext datastoreContext) { Preconditions.checkNotNull(type, "type should not be null"); Preconditions.checkNotNull(cluster, "cluster should not be null"); Preconditions.checkNotNull(configuration, "configuration should not be null"); - return Props.create(new Creator() { - - @Override - public ShardManager create() throws Exception { - return new ShardManager(type, cluster, configuration); - } - }); + return Props.create(new ShardManagerCreator(type, cluster, configuration, datastoreContext)); } - @Override public void handleReceive(Object message) throws Exception { if (message.getClass().equals(FindPrimary.SERIALIZABLE_CLASS)) { @@ -243,8 +245,9 @@ public class ShardManager extends AbstractUntypedActor { ShardIdentifier shardId = getShardIdentifier(memberName, shardName); Map peerAddresses = getPeerAddresses(shardName); ActorRef actor = getContext() - .actorOf(Shard.props(shardId, peerAddresses), - shardId.toString()); + .actorOf(Shard.props(shardId, peerAddresses, datastoreContext). + withMailbox(ActorContext.MAILBOX), shardId.toString()); + localShardActorNames.add(shardId.toString()); localShards.put(shardName, new ShardInformation(shardName, actor, peerAddresses)); } @@ -283,10 +286,17 @@ public class ShardManager extends AbstractUntypedActor { @Override public SupervisorStrategy supervisorStrategy() { + return new OneForOneStrategy(10, Duration.create("1 minute"), new Function() { @Override public SupervisorStrategy.Directive apply(Throwable t) { + StringBuilder sb = new StringBuilder(); + for(StackTraceElement element : t.getStackTrace()) { + sb.append("\n\tat ") + .append(element.toString()); + } + LOG.warning("Supervisor Strategy of resume applied {}",sb.toString()); return SupervisorStrategy.resume(); } } @@ -337,6 +347,28 @@ public class ShardManager extends AbstractUntypedActor { } } } + + private static class ShardManagerCreator implements Creator { + private static final long serialVersionUID = 1L; + + final String type; + final ClusterWrapper cluster; + final Configuration configuration; + final DatastoreContext datastoreContext; + + ShardManagerCreator(String type, ClusterWrapper cluster, + Configuration configuration, DatastoreContext datastoreContext) { + this.type = type; + this.cluster = cluster; + this.configuration = configuration; + this.datastoreContext = datastoreContext; + } + + @Override + public ShardManager create() throws Exception { + return new ShardManager(type, cluster, configuration, datastoreContext); + } + } }