Fixed issue #4465

orientechnologies · Jul 1, 2015 · e181c55 · e181c55
1 parent 020cc21
commit e181c55
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 16 deletions.
diff --git a/...buted/src/main/java/com/orientechnologies/orient/server/hazelcast/ODistributedWorker.java b/...buted/src/main/java/com/orientechnologies/orient/server/hazelcast/ODistributedWorker.java
@@ -101,7 +101,6 @@ public void run() {
             queuedMsg, databaseName, lastMessageId);
 
         restoringMessages = false;
-        break;
       }
 
       String senderNode = null;

diff --git a/...ain/java/com/orientechnologies/orient/server/hazelcast/OHazelcastDistributedDatabase.java b/...ain/java/com/orientechnologies/orient/server/hazelcast/OHazelcastDistributedDatabase.java
@@ -212,7 +212,10 @@ public OHazelcastDistributedDatabase configureDatabase(final boolean iRestoreMes
     final String queueName = OHazelcastDistributedMessageService.getRequestQueueName(getLocalNodeName(), databaseName);
     final IQueue<ODistributedRequest> requestQueue = msgService.getQueue(queueName);
 
-    unqueuePendingMessages(iRestoreMessages, iUnqueuePendingMessages, queueName, requestQueue);
+    final ODistributedWorker listenerThread = unqueuePendingMessages(iRestoreMessages, iUnqueuePendingMessages, queueName,
+        requestQueue);
+
+    workers.add(listenerThread);
 
     if (iCallback != null)
       try {
@@ -223,11 +226,6 @@ public OHazelcastDistributedDatabase configureDatabase(final boolean iRestoreMes
 
     setOnline();
 
-    // CREATE 1 WORKER THREAD FOR INSERT (ONLY 1 TO MAINTAIN THE SEQUENCE OF REQUESTS)
-    ODistributedWorker listenerThread = new ODistributedWorker(this, requestQueue, databaseName, 0, false);
-    workers.add(listenerThread);
-    listenerThread.start();
-
     return this;
   }
 
@@ -298,7 +296,7 @@ public void shutdown() {
       workers.get(i).shutdown();
   }
 
-  protected void unqueuePendingMessages(boolean iRestoreMessages, boolean iUnqueuePendingMessages, String queueName,
+  protected ODistributedWorker unqueuePendingMessages(boolean iRestoreMessages, boolean iUnqueuePendingMessages, String queueName,
       IQueue<ODistributedRequest> requestQueue) {
     if (ODistributedServerLog.isDebugEnabled())
       ODistributedServerLog.debug(this, getLocalNodeName(), null, DIRECTION.NONE, "listening for incoming requests on queue: %s",
@@ -312,14 +310,9 @@ protected void unqueuePendingMessages(boolean iRestoreMessages, boolean iUnqueue
     final ODistributedWorker listenerThread = new ODistributedWorker(this, requestQueue, databaseName, 0, restoringMessages);
     listenerThread.initDatabaseInstance();
 
-    if (restoringMessages) {
-      // EXECUTES PENDING MSG ONLY BEFORE TO GO ONLINE
-      listenerThread.start();
-      try {
-        listenerThread.join();
-      } catch (InterruptedException e) {
-      }
-    }
+    listenerThread.start();
+
+    return listenerThread;
   }
 
   protected void checkForServerOnline(ODistributedRequest iRequest) throws ODistributedException {