Fixed issue #4999

core/src/main/java/com/orientechnologies/orient/core/db/document/ODatabaseDocumentTx.java

@@ -2739,7 +2739,8 @@ public void restore(final InputStream in, final Map<String, Object> options, fin
 
     getStorage().restore(in, options, callable, iListener);
 
-    getMetadata().reload();
+    if (!isClosed())
+      getMetadata().reload();
   }
 
   /**

distributed/src/main/java/com/orientechnologies/orient/server/hazelcast/OHazelcastPlugin.java

@@ -347,6 +347,11 @@ public DB_STATUS getDatabaseStatus(final String iNode, final String iDatabaseNam
   @Override
   public void setDatabaseStatus(final String iNode, final String iDatabaseName, final DB_STATUS iStatus) {
     getConfigurationMap().put(OHazelcastPlugin.CONFIG_DBSTATUS_PREFIX + iNode + "." + iDatabaseName, iStatus);
+
+    // NOTIFY DB/NODE IS CHANGING STATUS
+    for (ODistributedLifecycleListener l : listeners) {
+      l.onDatabaseChangeStatus(iNode, iDatabaseName, iStatus);
+    }
   }
 
   @Override

distributed/src/test/java/com/orientechnologies/orient/server/distributed/AbstractServerClusterTest.java

@@ -32,11 +32,11 @@
  * Test class that creates and executes distributed operations against a cluster of servers created in the same JVM.
  */
 public abstract class AbstractServerClusterTest {
-  protected int             delayServerStartup = 0;
-  protected int             delayServerAlign   = 0;
-  protected String          rootDirectory      = "target/servers/";
+  protected int    delayServerStartup = 0;
+  protected int    delayServerAlign   = 0;
+  protected String rootDirectory      = "target/servers/";
 
-  protected List<ServerRun> serverInstance     = new ArrayList<ServerRun>();
+  protected List<ServerRun> serverInstance = new ArrayList<ServerRun>();
 
   protected AbstractServerClusterTest() {
   }
@@ -93,10 +93,13 @@ public void execute() throws Exception {
 
     try {
 
-      for (ServerRun server : serverInstance) {
+      for (final ServerRun server : serverInstance) {
         banner("STARTING SERVER -> " + server.getServerId() + "...");
+
         server.startServer(getDistributedServerConfiguration(server));
 
+        onServerStarting(server);
+
         if (delayServerStartup > 0)
           try {
             Thread.sleep(delayServerStartup * serverInstance.size());
@@ -108,8 +111,8 @@ public void execute() throws Exception {
 
       if (delayServerAlign > 0)
         try {
-          System.out.println("Server started, waiting for synchronization (" + (delayServerAlign * serverInstance.size() / 1000)
-              + "secs)...");
+          System.out.println(
+              "Server started, waiting for synchronization (" + (delayServerAlign * serverInstance.size() / 1000) + "secs)...");
           Thread.sleep(delayServerAlign * serverInstance.size());
         } catch (InterruptedException e) {
         }
@@ -160,10 +163,13 @@ protected void banner(final String iMessage) {
 
   protected void log(final String iMessage) {
     OLogManager.instance().flush();
-    System.out.println("\n"+iMessage);
+    System.out.println("\n" + iMessage);
     System.out.flush();
   }
 
+  protected void onServerStarting(ServerRun server) {
+  }
+
   protected void onServerStarted(ServerRun server) {
   }
 

distributed/src/test/java/com/orientechnologies/orient/server/distributed/DistributedLifecycleListenerTest.java

@@ -20,20 +20,26 @@
 
 package com.orientechnologies.orient.server.distributed;
 
-import junit.framework.Assert;
+import com.orientechnologies.common.util.OPair;
+import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * Tests the behavior of hooks in distributed configuration.
  */
 public class DistributedLifecycleListenerTest extends AbstractServerClusterTest implements ODistributedLifecycleListener {
-  private final static int SERVERS        = 2;
+  private final static int SERVERS = 2;
 
-  private final AtomicLong beforeNodeJoin = new AtomicLong();
-  private final AtomicLong afterNodeJoin  = new AtomicLong();
-  private final AtomicLong nodeLeft       = new AtomicLong();
+  private final AtomicLong                                               beforeNodeJoin = new AtomicLong();
+  private final AtomicLong                                               afterNodeJoin  = new AtomicLong();
+  private final AtomicLong                                               nodeLeft       = new AtomicLong();
+  private final List<OPair<String, ODistributedServerManager.DB_STATUS>> changeStatus   = Collections
+      .synchronizedList(new ArrayList<OPair<String, ODistributedServerManager.DB_STATUS>>());
 
   @Override
   public boolean onNodeJoining(String iNode) {
@@ -51,6 +57,11 @@ public void onNodeLeft(String iNode) {
     nodeLeft.incrementAndGet();
   }
 
+  @Override
+  public void onDatabaseChangeStatus(String iNode, String iDatabaseName, ODistributedServerManager.DB_STATUS iNewStatus) {
+    changeStatus.add(new OPair<String, ODistributedServerManager.DB_STATUS>(iNode + "." + iDatabaseName, iNewStatus));
+  }
+
   public String getDatabaseName() {
     return "distributed-lifecycle";
   }
@@ -63,7 +74,7 @@ public void test() throws Exception {
   }
 
   @Override
-  protected void onServerStarted(ServerRun server) {
+  protected void onServerStarting(ServerRun server) {
     server.getServerInstance().getDistributedManager().registerLifecycleListener(this);
   }
 
@@ -77,15 +88,23 @@ protected void onTestEnded() {
     Assert.assertEquals(SERVERS - 1, beforeNodeJoin.get());
     Assert.assertEquals(SERVERS - 1, afterNodeJoin.get());
 
-    for (int attempt = 0; attempt < 30; ++attempt) {
-      if (nodeLeft.get() == SERVERS)
+    for (int attempt = 0; attempt < 50; ++attempt) {
+      if (nodeLeft.get() >= SERVERS - 1)
         break;
       try {
-        Thread.sleep(300);
+        Thread.sleep(500);
       } catch (InterruptedException e) {
         e.printStackTrace();
       }
     }
     Assert.assertEquals(SERVERS - 1, nodeLeft.get());
+
+    Assert.assertEquals(3, changeStatus.size());
+    Assert.assertEquals("europe1." + getDatabaseName(), changeStatus.get(0).getKey());
+    Assert.assertEquals(ODistributedServerManager.DB_STATUS.SYNCHRONIZING, changeStatus.get(0).getValue());
+    Assert.assertEquals("europe0." + getDatabaseName(), changeStatus.get(1).getKey());
+    Assert.assertEquals(ODistributedServerManager.DB_STATUS.SYNCHRONIZING, changeStatus.get(1).getValue());
+    Assert.assertEquals("europe0." + getDatabaseName(), changeStatus.get(2).getKey());
+    Assert.assertEquals(ODistributedServerManager.DB_STATUS.ONLINE, changeStatus.get(2).getValue());
   }
 }

distributed/src/test/java/com/orientechnologies/orient/server/distributed/TestShardingManualSync.java

@@ -1,14 +1,12 @@
 package com.orientechnologies.orient.server.distributed;
 
-import junit.framework.Assert;
-
-import org.junit.Test;
-
 import com.orientechnologies.orient.core.sql.OCommandSQL;
 import com.tinkerpop.blueprints.impls.orient.OrientGraphFactory;
 import com.tinkerpop.blueprints.impls.orient.OrientGraphNoTx;
 import com.tinkerpop.blueprints.impls.orient.OrientVertex;
 import com.tinkerpop.blueprints.impls.orient.OrientVertexType;
+import junit.framework.Assert;
+import org.junit.Test;
 
 public class TestShardingManualSync extends AbstractServerClusterTest {
 

server/src/main/java/com/orientechnologies/orient/server/distributed/ODistributedLifecycleListener.java

@@ -51,4 +51,17 @@ public interface ODistributedLifecycleListener {
    *          Node name that left
    */
   void onNodeLeft(String iNode);
+
+  /**
+   * Called upon change of database status on a node. Available statuses are defined in ODistributedServerManager.DB_STATUS.
+   * 
+   * @since 2.2.0
+   * @param iNode
+   *          The node name
+   * @param iDatabaseName
+   *          Database name
+   * @param iNewStatus
+   *          The new status
+   */
+  void onDatabaseChangeStatus(String iNode, String iDatabaseName, ODistributedServerManager.DB_STATUS iNewStatus);
 }