minor changes of the structure of two phase operation

distributed/src/main/java/com/orientechnologies/orient/server/distributed/impl/coordinator/ODistributedCoordinator.java

@@ -33,7 +33,7 @@ public OLogId log(ONodeRequest request) {
 
   public ORequestContext sendOperation(OSubmitRequest submitRequest, ONodeRequest nodeRequest, OResponseHandler handler) {
     OLogId id = log(nodeRequest);
-    ORequestContext context = new ORequestContext(this, submitRequest, nodeRequest, members.size() / 2 + 1, handler);
+    ORequestContext context = new ORequestContext(this, submitRequest, nodeRequest, members.values(), handler);
     contexts.put(id, context);
     for (ODistributedMember member : members.values()) {
       member.sendRequest(id, nodeRequest);

distributed/src/main/java/com/orientechnologies/orient/server/distributed/impl/coordinator/ORequestContext.java

@@ -1,6 +1,7 @@
 package com.orientechnologies.orient.server.distributed.impl.coordinator;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
@@ -10,27 +11,27 @@ public enum Status {
     STARTED, QUORUM_OK, QUORUM_KO
   }
 
-  private OSubmitRequest          submitRequest;
-  private ONodeRequest            nodeRequest;
-  private List<ONodeResponse>     responses = Collections.synchronizedList(new ArrayList<>());
-  private ODistributedCoordinator coordinator;
-  private int                     quorum;
-  private OResponseHandler        handler;
-  private Status                  status;
+  private OSubmitRequest                 submitRequest;
+  private ONodeRequest                   nodeRequest;
+  private Collection<ODistributedMember> involvedMembers;
+  private List<ONodeResponse>            responses = Collections.synchronizedList(new ArrayList<>());
+  private ODistributedCoordinator        coordinator;
+  private int                            quorum;
+  private OResponseHandler               handler;
 
-  public ORequestContext(ODistributedCoordinator coordinator, OSubmitRequest submitRequest, ONodeRequest nodeRequest, int quorum,
-      OResponseHandler handler) {
+  public ORequestContext(ODistributedCoordinator coordinator, OSubmitRequest submitRequest, ONodeRequest nodeRequest,
+      Collection<ODistributedMember> involvedMembers, OResponseHandler handler) {
     this.coordinator = coordinator;
     this.submitRequest = submitRequest;
     this.nodeRequest = nodeRequest;
-    this.quorum = quorum;
+    this.involvedMembers = involvedMembers;
     this.handler = handler;
-    this.status = Status.STARTED;
+    this.quorum = (involvedMembers.size() / 2) + 1;
   }
 
   public void receive(ONodeResponse response) {
     responses.add(response);
-    status = handler.receive(coordinator, this, response, status);
+    handler.receive(coordinator, this, response);
   }
 
   public List<ONodeResponse> getResponses() {

distributed/src/main/java/com/orientechnologies/orient/server/distributed/impl/coordinator/OResponseHandler.java

@@ -1,6 +1,5 @@
 package com.orientechnologies.orient.server.distributed.impl.coordinator;
 
 public interface OResponseHandler {
-  ORequestContext.Status receive(ODistributedCoordinator coordinator, ORequestContext context, ONodeResponse response,
-      ORequestContext.Status status);
+  void receive(ODistributedCoordinator coordinator, ORequestContext context, ONodeResponse response);
 }

distributed/src/test/java/com/orientechnologies/orient/server/distributed/impl/coordinator/ODistributedCoordinatorTest.java

@@ -25,11 +25,10 @@ public void simpleOperationTest() throws InterruptedException {
       @Override
       public void begin(ODistributedMember member, ODistributedCoordinator coordinator) {
         MockNodeRequest nodeRequest = new MockNodeRequest();
-        coordinator.sendOperation(this, nodeRequest, (coordinator1, context, response, status) -> {
+        coordinator.sendOperation(this, nodeRequest, (coordinator1, context, response) -> {
           if (context.getResponses().size() == 1) {
             responseReceived.countDown();
           }
-          return status;
         });
       }
 
@@ -55,22 +54,20 @@ public void testTwoPhase() throws InterruptedException {
       @Override
       public void begin(ODistributedMember member, ODistributedCoordinator coordinator) {
         MockNodeRequest nodeRequest = new MockNodeRequest();
-        coordinator.sendOperation(this, nodeRequest, (coordinator1, context, response, status) -> {
+        coordinator.sendOperation(this, nodeRequest, (coordinator1, context, response) -> {
           if (context.getResponses().size() == 1) {
             coordinator1.sendOperation(this, new ONodeRequest() {
               @Override
               public ONodeResponse execute(ODistributedMember nodeFrom, OLogId opId, ODistributedExecutor executor) {
                 return null;
               }
-            }, (coordinator2, context1, response1, status1) -> {
+            }, (coordinator2, context1, response1) -> {
               if (context.getResponses().size() == 1) {
                 member.reply(new OSubmitResponse() {
                 });
               }
-              return status1;
             });
           }
-          return status;
         });
       }
     });

distributed/src/test/java/com/orientechnologies/orient/server/distributed/impl/coordinator/mocktx/FirstPhaseHandler.java

@@ -0,0 +1,24 @@
+package com.orientechnologies.orient.server.distributed.impl.coordinator.mocktx;
+
+import com.orientechnologies.orient.server.distributed.impl.coordinator.*;
+
+class FirstPhaseHandler implements OResponseHandler {
+  private       OSubmitTx          submitTx;
+  private final ODistributedMember member;
+  private       boolean            done;
+
+  public FirstPhaseHandler(OSubmitTx submitTx, ODistributedMember member) {
+    this.submitTx = submitTx;
+    this.member = member;
+  }
+
+  @Override
+  public void receive(ODistributedCoordinator coordinator1, ORequestContext context, ONodeResponse response) {
+    if (context.getResponses().size() >= context.getQuorum() && !done) {
+      done = true;
+      submitTx.firstPhase = true;
+      coordinator1.sendOperation(submitTx, new OPhase2Tx(), new SecondPhaseResponseHandler(submitTx, member));
+    }
+  }
+
+}

distributed/src/test/java/com/orientechnologies/orient/server/distributed/impl/coordinator/mocktx/OSubmitTx.java

@@ -3,27 +3,13 @@
 import com.orientechnologies.orient.server.distributed.impl.coordinator.*;
 
 public class OSubmitTx implements OSubmitRequest {
-  boolean firstPhase  = false;
-  boolean secondPhase = false;
+
+  public boolean firstPhase;
+  public boolean secondPhase;
 
   @Override
   public void begin(ODistributedMember member, ODistributedCoordinator coordinator) {
-    coordinator.sendOperation(this, new OPhase1Tx(), (coordinator1, context, response, status) -> {
-      if (context.getResponses().size() >= context.getQuorum() && status == ORequestContext.Status.STARTED) {
-        status = ORequestContext.Status.QUORUM_OK;
-        firstPhase = true;
-        coordinator1.sendOperation(this, new OPhase2Tx(), (coordinator2, context1, response1, status1) -> {
-          if (context1.getResponses().size() >= context1.getQuorum() && status1 == ORequestContext.Status.STARTED) {
-            status1 = ORequestContext.Status.QUORUM_OK;
-            secondPhase = true;
-            member.reply(new OSubmitResponse() {
-            });
-          }
-          return status1;
-        });
-      }
-      return status;
-    });
+    coordinator.sendOperation(this, new OPhase1Tx(), new FirstPhaseHandler(this, member));
   }
 
 }

distributed/src/test/java/com/orientechnologies/orient/server/distributed/impl/coordinator/mocktx/SecondPhaseResponseHandler.java

@@ -0,0 +1,24 @@
+package com.orientechnologies.orient.server.distributed.impl.coordinator.mocktx;
+
+import com.orientechnologies.orient.server.distributed.impl.coordinator.*;
+
+public class SecondPhaseResponseHandler implements OResponseHandler {
+  private final OSubmitTx          submitTx;
+  private final ODistributedMember member;
+  boolean done = false;
+
+  public SecondPhaseResponseHandler(OSubmitTx submitTx, ODistributedMember member) {
+    this.member = member;
+    this.submitTx = submitTx;
+  }
+
+  @Override
+  public void receive(ODistributedCoordinator coordinator, ORequestContext context1, ONodeResponse response) {
+    if (context1.getResponses().size() >= context1.getQuorum() && !done) {
+      done = true;
+      submitTx.secondPhase = true;
+      member.reply(new OSubmitResponse() {
+      });
+    }
+  }
+}