distributed/src/main/java/com/orientechnologies/orient/server/distributed/impl/ODistributedAbstractPlugin.java
protected void installDatabaseFromNetwork(final String dbPath, final String databaseName,
      final ODistributedDatabaseImpl distrDatabase, final String iNode, final ODistributedDatabaseChunk firstChunk,
      final boolean delta, final File uniqueClustersBackupDirectory, final OModifiableDistributedConfiguration cfg) {

    final String localNodeName = nodeName;

    final AtomicReference<ODistributedMomentum> momentum = new AtomicReference<ODistributedMomentum>();

    OSyncReceiver receiver = new OSyncReceiver(this, databaseName, firstChunk, momentum, iNode, dbPath);
    try {
      Thread t = new Thread(receiver);
      t.setUncaughtExceptionHandler(new OUncaughtExceptionHandler());
      t.start();

    } catch (Exception e) {
      ODistributedServerLog.error(this, nodeName, null, DIRECTION.NONE, "Error on transferring database '%s' ", e, databaseName);
      throw OException.wrapException(new ODistributedException("Error on transferring database"), e);
    }

    final ODatabaseDocumentInternal db = installDatabaseOnLocalNode(databaseName, dbPath, iNode, delta,
        uniqueClustersBackupDirectory, cfg, firstChunk.incremental, receiver);

    if (db == null)
      return;

    // OVERWRITE THE MOMENTUM FROM THE ORIGINAL SERVER AND ADD LAST LOCAL LSN
    try {
      distrDatabase.getSyncConfiguration().load();
      distrDatabase.getSyncConfiguration()
          .setLastLSN(localNodeName, ((OLocalPaginatedStorage) db.getStorage().getUnderlying()).getLSN(), false);

    } catch (IOException e) {
      ODistributedServerLog.error(this, nodeName, null, DIRECTION.NONE, "Error on loading %s file for database '%s'", e,
          DISTRIBUTED_SYNC_JSON_FILENAME, databaseName);
    }

    try {
      distrDatabase.setOnline();
    } finally {
      db.activateOnCurrentThread();
      db.close();
    }

    // ASK FOR INDIVIDUAL CLUSTERS IN CASE OF SHARDING AND NO LOCAL COPY
    final Set<String> localManagedClusters = cfg.getClustersOnServer(localNodeName);
    final Set<String> sourceNodeClusters = cfg.getClustersOnServer(iNode);
    localManagedClusters.removeAll(sourceNodeClusters);

    final HashSet<String> toSynchClusters = new HashSet<String>();
    for (String cl : localManagedClusters) {
      // FILTER CLUSTER CHECKING IF ANY NODE IS ACTIVE
      final List<String> servers = cfg.getServers(cl, localNodeName);
      getAvailableNodes(servers, databaseName);

      if (!servers.isEmpty())
        toSynchClusters.add(cl);
    }

    // SYNC ALL THE CLUSTERS
    for (String cl : toSynchClusters) {
      // FILTER CLUSTER CHECKING IF ANY NODE IS ACTIVE
      OCommandExecutorSQLHASyncCluster.replaceCluster(this, serverInstance, databaseName, cl);
    }

    try {
      rebalanceClusterOwnership(nodeName, db, cfg, false);
    } catch (Exception e) {
      // HANDLE IT AS WARNING
      ODistributedServerLog
          .warn(this, nodeName, null, DIRECTION.NONE, "Error on re-balancing the cluster for database '%s'", e, databaseName);
      // NOT CRITICAL, CONTINUE
    }
  }
protected void installDatabaseFromNetwork(final String dbPath, final String databaseName,
      final ODistributedDatabaseImpl distrDatabase, final String iNode, final ODistributedDatabaseChunk firstChunk,
      final boolean delta, final File uniqueClustersBackupDirectory, final OModifiableDistributedConfiguration cfg) {

    final String localNodeName = nodeName;

    final AtomicReference<ODistributedMomentum> momentum = new AtomicReference<ODistributedMomentum>();

    OSyncReceiver receiver = new OSyncReceiver(this, databaseName, firstChunk, momentum, iNode, dbPath);
    try {
      Thread t = new Thread(receiver);
      t.setUncaughtExceptionHandler(new OUncaughtExceptionHandler());
      t.start();

    } catch (Exception e) {
      ODistributedServerLog.error(this, nodeName, null, DIRECTION.NONE, "Error on transferring database '%s' ", e, databaseName);
      throw OException.wrapException(new ODistributedException("Error on transferring database"), e);
    }

    final ODatabaseDocumentInternal db = installDatabaseOnLocalNode(databaseName, dbPath, iNode, delta,
        uniqueClustersBackupDirectory, cfg, firstChunk.incremental, receiver);

    if (db == null)
      return;

    // OVERWRITE THE MOMENTUM FROM THE ORIGINAL SERVER AND ADD LAST LOCAL LSN
    try {
      distrDatabase.getSyncConfiguration().load();
      distrDatabase.getSyncConfiguration()
          .setLastLSN(localNodeName, ((OLocalPaginatedStorage) db.getStorage().getUnderlying()).getLSN(), false);

    } catch (IOException e) {
      ODistributedServerLog.error(this, nodeName, null, DIRECTION.NONE, "Error on loading %s file for database '%s'", e,
          DISTRIBUTED_SYNC_JSON_FILENAME, databaseName);
    }

    try {
      distrDatabase.setOnline();
    } finally {
      db.activateOnCurrentThread();
      db.close();
    }

    // ASK FOR INDIVIDUAL CLUSTERS IN CASE OF SHARDING AND NO LOCAL COPY
    final Set<String> localManagedClusters = cfg.getClustersOnServer(localNodeName);
    final Set<String> sourceNodeClusters = cfg.getClustersOnServer(iNode);
    localManagedClusters.removeAll(sourceNodeClusters);

    final HashSet<String> toSynchClusters = new HashSet<String>();
    for (String cl : localManagedClusters) {
      // FILTER CLUSTER CHECKING IF ANY NODE IS ACTIVE
      final List<String> servers = cfg.getServers(cl, localNodeName);
      getAvailableNodes(servers, databaseName);

      if (!servers.isEmpty())
        toSynchClusters.add(cl);
    }

    // SYNC ALL THE CLUSTERS
    for (String cl : toSynchClusters) {
      // FILTER CLUSTER CHECKING IF ANY NODE IS ACTIVE
      OCommandExecutorSQLHASyncCluster.replaceCluster(this, serverInstance, databaseName, cl);
    }

    try {
      rebalanceClusterOwnership(nodeName, db, cfg, false);
    } catch (Exception e) {
      // HANDLE IT AS WARNING
      ODistributedServerLog
          .warn(this, nodeName, null, DIRECTION.NONE, "Error on re-balancing the cluster for database '%s'", e, databaseName);
      // NOT CRITICAL, CONTINUE
    }
  }
Installs a database from the network.