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=2dce6a1079c4fdbb0a8e2fa090fa018908d3f5ce;hb=35f74293edf98402e2b622e060185f7874d10857;hp=34d35312838fd4346e7446161cd4f1f1c8f9cc74;hpb=d6f1e7790157461553b26ec82d246e68b62aad6b;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..2dce6a1079 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.ShardStats;
 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 ShardStats shardStats;
 
     public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort,
-        ActorRef shardActor, CompositeModification modification) {
+        ActorRef shardActor, CompositeModification modification, ShardStats shardStats) {
 
         this.cohort = cohort;
         this.shardActor = shardActor;
         this.modification = modification;
+        this.shardStats = shardStats;
     }
 
     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<ThreePhaseCommitCohort>() {
-            @Override
-            public ThreePhaseCommitCohort create() throws Exception {
-                return new ThreePhaseCommitCohort(cohort, shardActor,
-                    modification);
-            }
-        });
+            final ActorRef shardActor, final CompositeModification modification,
+            ShardStats shardStats) {
+        return Props.create(new ThreePhaseCommitCohortCreator(cohort, shardActor, modification,
+                shardStats));
     }
 
-
     @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<Void>() {
+            @Override
             public void onSuccess(Void v) {
+                shardStats.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<Void>() {
+            @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<Boolean>() {
+            @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<ThreePhaseCommitCohort> {
+        final DOMStoreThreePhaseCommitCohort cohort;
+        final ActorRef shardActor;
+        final CompositeModification modification;
+        final ShardStats shardStats;
+
+        ThreePhaseCommitCohortCreator(DOMStoreThreePhaseCommitCohort cohort,
+            ActorRef shardActor, CompositeModification modification, ShardStats shardStats) {
+            this.cohort = cohort;
+            this.shardActor = shardActor;
+            this.modification = modification;
+            this.shardStats = shardStats;
+        }
 
+        @Override
+        public ThreePhaseCommitCohort create() throws Exception {
+            return new ThreePhaseCommitCohort(cohort, shardActor, modification, shardStats);
+        }
     }
 }