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%2FThreePhaseCommitCohort.java;h=5a6d0eca5c2a159963febc4ee9d6436f2e864a5d;hb=67d58d1ab50f3c3bbe19a96fb6f0d9d94211829f;hp=34d35312838fd4346e7446161cd4f1f1c8f9cc74;hpb=30faeb35260541c273a81b8f126b40da94daa825;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java index 34d3531283..5a6d0eca5c 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java @@ -14,9 +14,12 @@ import akka.actor.Props; import akka.event.Logging; import akka.event.LoggingAdapter; import akka.japi.Creator; + import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; + +import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory; import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction; import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; @@ -32,30 +35,27 @@ public class ThreePhaseCommitCohort extends AbstractUntypedActor { private final DOMStoreThreePhaseCommitCohort cohort; private final ActorRef shardActor; private final CompositeModification modification; + private final String shardName; public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort, - ActorRef shardActor, CompositeModification modification) { + ActorRef shardActor, CompositeModification modification,String shardName) { this.cohort = cohort; this.shardActor = shardActor; this.modification = modification; + this.shardName = shardName; } private final LoggingAdapter log = Logging.getLogger(getContext().system(), this); public static Props props(final DOMStoreThreePhaseCommitCohort cohort, - final ActorRef shardActor, final CompositeModification modification) { - return Props.create(new Creator() { - @Override - public ThreePhaseCommitCohort create() throws Exception { - return new ThreePhaseCommitCohort(cohort, shardActor, - modification); - } - }); + final ActorRef shardActor, final CompositeModification modification, + String shardName) { + return Props.create(new ThreePhaseCommitCohortCreator(cohort, shardActor, modification, + shardName)); } - @Override public void handleReceive(Object message) throws Exception { if (message.getClass() @@ -81,12 +81,15 @@ public class ThreePhaseCommitCohort extends AbstractUntypedActor { final ActorRef self = getSelf(); Futures.addCallback(future, new FutureCallback() { + @Override public void onSuccess(Void v) { + ShardMBeanFactory.getShardStatsMBean(shardName).incrementAbortTransactionsCount(); sender .tell(new AbortTransactionReply().toSerializable(), - self); + self); } + @Override public void onFailure(Throwable t) { LOG.error(t, "An exception happened during abort"); sender @@ -110,12 +113,14 @@ public class ThreePhaseCommitCohort extends AbstractUntypedActor { final ActorRef sender = getSender(); final ActorRef self = getSelf(); Futures.addCallback(future, new FutureCallback() { + @Override public void onSuccess(Void v) { sender .tell(new PreCommitTransactionReply().toSerializable(), self); } + @Override public void onFailure(Throwable t) { LOG.error(t, "An exception happened during pre-commit"); sender @@ -130,18 +135,38 @@ public class ThreePhaseCommitCohort extends AbstractUntypedActor { final ActorRef sender = getSender(); final ActorRef self = getSelf(); Futures.addCallback(future, new FutureCallback() { + @Override public void onSuccess(Boolean canCommit) { sender.tell(new CanCommitTransactionReply(canCommit) .toSerializable(), self); } + @Override public void onFailure(Throwable t) { LOG.error(t, "An exception happened during canCommit"); sender .tell(new akka.actor.Status.Failure(t), self); } }); + } + private static class ThreePhaseCommitCohortCreator implements Creator { + final DOMStoreThreePhaseCommitCohort cohort; + final ActorRef shardActor; + final CompositeModification modification; + final String shardName; + + ThreePhaseCommitCohortCreator(DOMStoreThreePhaseCommitCohort cohort, + ActorRef shardActor, CompositeModification modification, String shardName) { + this.cohort = cohort; + this.shardActor = shardActor; + this.modification = modification; + this.shardName = shardName; + } + @Override + public ThreePhaseCommitCohort create() throws Exception { + return new ThreePhaseCommitCohort(cohort, shardActor, modification, shardName); + } } }