distributed-v2/src/main/java/com/orientechnologies/orient/distributed/impl/ODistributedWorker.java
protected void onMessage(final ODistributedRequest iRequest) {
    String senderNodeName = null;
    for (int retry = 0; retry < 10; retry++) {
      senderNodeName = manager.getNodeNameById(iRequest.getId().getNodeId());
      if (senderNodeName != null)
        break;

      try {
        Thread.sleep(200);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();

        throw OException.wrapException(new ODistributedException("Execution has been interrupted"), e);
      }
    }

    if (senderNodeName == null) {
      ODistributedServerLog.warn(this, localNodeName, senderNodeName, DIRECTION.IN,
          "Sender server id %d is not registered in the cluster configuration, discard the request: (%s) (worker=%d)",
          iRequest.getId().getNodeId(), iRequest, id);
      sendResponseBack(iRequest, new ODistributedException("Sender server id " + iRequest.getId().getNodeId()
          + " is not registered in the cluster configuration, discard the request"));
      return;
    }

    final ORemoteTask task = iRequest.getTask();

    if (ODistributedServerLog.isDebugEnabled())
      ODistributedServerLog
          .debug(this, localNodeName, senderNodeName, DIRECTION.IN, "Received request: (%s) (worker=%d)", iRequest, id);

    // EXECUTE IT LOCALLY
    Object responsePayload = null;
    OSecurityUser origin = null;
    try {
      waitNodeIsOnline();

      distributed.waitIsReady(task);

      // EXECUTE THE TASK
      for (int retry = 1; running; ++retry) {
        if (task.isUsingDatabase()) {
          initDatabaseInstance();
          if (database == null)
            throw new ODistributedOperationException(
                "Error on executing remote request because the database '" + databaseName + "' is not available");
          // keep original user in database, check the username passed in request and set new user in DB, after document saved,
          // reset to original user
          if (database != null) {
            database.activateOnCurrentThread();
            origin = database.getUser();
            try {
              if (iRequest.getUserRID() != null && iRequest.getUserRID().isValid() && (lastUser == null || !(lastUser.getIdentity())
                  .equals(iRequest.getUserRID()))) {
                lastUser = database.getMetadata().getSecurity().getUser(iRequest.getUserRID());
                database.setUser(lastUser);// set to new user
              } else
                origin = null;

            } catch (Exception ex) {
              OLogManager.instance().error(this, "Failed on user switching database. ", ex);
            }
          }

        }

        responsePayload = manager.executeOnLocalNode(iRequest.getId(), iRequest.getTask(), database);

        if (responsePayload instanceof OModificationOperationProhibitedException) {
          // RETRY
          try {
            ODistributedServerLog.info(this, localNodeName, senderNodeName, DIRECTION.IN,
                "Database is frozen, waiting and retrying. Request %s (retry=%d, worker=%d)", iRequest, retry, id);

            Thread.sleep(1000);
          } catch (InterruptedException e) {
          }
        } else {
          // OPERATION EXECUTED (OK OR ERROR), NO RETRY NEEDED
          if (retry > 1)
            ODistributedServerLog
                .info(this, localNodeName, senderNodeName, DIRECTION.IN, "Request %s succeed after retry=%d", iRequest, retry);

          break;
        }

      }

    } catch (RuntimeException e) {
      if (task.hasResponse())
        sendResponseBack(iRequest, e);
      throw e;

    } finally {
      if (database != null && !database.isClosed()) {
        database.activateOnCurrentThread();
        if (!database.isClosed()) {
          database.rollback();
          database.getLocalCache().clear();
          if (origin != null)
            database.setUser(origin);
        }
      }
    }

    if (task.hasResponse()) {
      if (!sendResponseBack(iRequest, responsePayload)) {
        handleError(iRequest, responsePayload);
      }
    }
  }
protected void onMessage(final ODistributedRequest iRequest) {
    String senderNodeName = null;
    for (int retry = 0; retry < 10; retry++) {
      senderNodeName = manager.getNodeNameById(iRequest.getId().getNodeId());
      if (senderNodeName != null)
        break;

      try {
        Thread.sleep(200);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();

        throw OException.wrapException(new ODistributedException("Execution has been interrupted"), e);
      }
    }

    if (senderNodeName == null) {
      ODistributedServerLog.warn(this, localNodeName, senderNodeName, DIRECTION.IN,
          "Sender server id %d is not registered in the cluster configuration, discard the request: (%s) (worker=%d)",
          iRequest.getId().getNodeId(), iRequest, id);
      sendResponseBack(iRequest, new ODistributedException("Sender server id " + iRequest.getId().getNodeId()
          + " is not registered in the cluster configuration, discard the request"));
      return;
    }

    final ORemoteTask task = iRequest.getTask();

    if (ODistributedServerLog.isDebugEnabled())
      ODistributedServerLog
          .debug(this, localNodeName, senderNodeName, DIRECTION.IN, "Received request: (%s) (worker=%d)", iRequest, id);

    // EXECUTE IT LOCALLY
    Object responsePayload = null;
    OSecurityUser origin = null;
    try {
      waitNodeIsOnline();

      distributed.waitIsReady(task);

      // EXECUTE THE TASK
      for (int retry = 1; running; ++retry) {
        if (task.isUsingDatabase()) {
          initDatabaseInstance();
          if (database == null)
            throw new ODistributedOperationException(
                "Error on executing remote request because the database '" + databaseName + "' is not available");
          // keep original user in database, check the username passed in request and set new user in DB, after document saved,
          // reset to original user
          if (database != null) {
            database.activateOnCurrentThread();
            origin = database.getUser();
            try {
              if (iRequest.getUserRID() != null && iRequest.getUserRID().isValid() && (lastUser == null || !(lastUser.getIdentity())
                  .equals(iRequest.getUserRID()))) {
                lastUser = database.getMetadata().getSecurity().getUser(iRequest.getUserRID());
                database.setUser(lastUser);// set to new user
              } else
                origin = null;

            } catch (Exception ex) {
              OLogManager.instance().error(this, "Failed on user switching database. ", ex);
            }
          }

        }

        responsePayload = manager.executeOnLocalNode(iRequest.getId(), iRequest.getTask(), database);

        if (responsePayload instanceof OModificationOperationProhibitedException) {
          // RETRY
          try {
            ODistributedServerLog.info(this, localNodeName, senderNodeName, DIRECTION.IN,
                "Database is frozen, waiting and retrying. Request %s (retry=%d, worker=%d)", iRequest, retry, id);

            Thread.sleep(1000);
          } catch (InterruptedException e) {
          }
        } else {
          // OPERATION EXECUTED (OK OR ERROR), NO RETRY NEEDED
          if (retry > 1)
            ODistributedServerLog
                .info(this, localNodeName, senderNodeName, DIRECTION.IN, "Request %s succeed after retry=%d", iRequest, retry);

          break;
        }

      }

    } catch (RuntimeException e) {
      if (task.hasResponse())
        sendResponseBack(iRequest, e);
      throw e;

    } finally {
      if (database != null && !database.isClosed()) {
        database.activateOnCurrentThread();
        if (!database.isClosed()) {
          database.rollback();
          database.getLocalCache().clear();
          if (origin != null)
            database.setUser(origin);
        }
      }
    }

    if (task.hasResponse()) {
      if (!sendResponseBack(iRequest, responsePayload)) {
        handleError(iRequest, responsePayload);
      }
    }
  }
Executes the remote call on the local node and send back the result