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
    }
  }
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.
Installs a database from the network .