activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r1498601 - in /activemq/trunk/activemq-leveldb-store/src: main/scala/org/apache/activemq/leveldb/ main/scala/org/apache/activemq/leveldb/replicated/ main/scala/org/apache/activemq/leveldb/util/ test/java/org/apache/activemq/leveldb/test/ te...
Date Mon, 01 Jul 2013 17:38:14 GMT
Author: chirino
Date: Mon Jul  1 17:38:13 2013
New Revision: 1498601

URL: http://svn.apache.org/r1498601
Log:
Improve the replicated leveldb behavior when the number of nodes in the cluster falls below
the required minimum.  The master node will switch to electing mode.  The master store startup
will now also block until it syncs up with slaves so that we don't accept connections the
master is fully online.

Removed:
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/util/RetrySupport.scala
Modified:
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterElector.scala
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterLevelDBStore.scala
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/SlaveLevelDBStore.scala
    activemq/trunk/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBStoreTest.java
    activemq/trunk/activemq-leveldb-store/src/test/scala/org/apache/activemq/leveldb/dfs/DFSLevelDBClient.scala

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/DBManager.scala
Mon Jul  1 17:38:13 2013
@@ -99,6 +99,7 @@ class CountDownFuture[T <: AnyRef]() ext
   private val latch:CountDownLatch=new CountDownLatch(1)
   @volatile
   var value:T = _
+  var error:Throwable = _
 
   def cancel(mayInterruptIfRunning: Boolean) = false
   def isCancelled = false
@@ -112,14 +113,24 @@ class CountDownFuture[T <: AnyRef]() ext
     value = v
     latch.countDown()
   }
+  def failed(v:Throwable) = {
+    error = v
+    latch.countDown()
+  }
 
   def get() = {
     latch.await()
+    if( error!=null ) {
+      throw error;
+    }
     value
   }
 
   def get(p1: Long, p2: TimeUnit) = {
     if(latch.await(p1, p2)) {
+      if( error!=null ) {
+        throw error;
+      }
       value
     } else {
       throw new TimeoutException
@@ -221,7 +232,7 @@ class DelayableUOW(val manager:DBManager
     manager.uowCanceledCounter += 1
     canceled = true
     manager.flush_queue.remove(uowId)
-    onCompleted
+    onCompleted()
   }
 
   def getAction(id:MessageId) = {
@@ -342,7 +353,7 @@ class DelayableUOW(val manager:DBManager
     }
   }
 
-  def onCompleted() = this.synchronized {
+  def onCompleted(error:Throwable=null) = this.synchronized {
     if ( state.stage < UowCompleted.stage ) {
       state = UowCompleted
       if( asyncCapacityUsed != 0 ) {
@@ -352,7 +363,11 @@ class DelayableUOW(val manager:DBManager
         manager.uow_complete_latency.add(System.nanoTime() - disposed_at)
         complete_listeners.foreach(_())
       }
-      countDownFuture.set(null)
+      if( error == null ) {
+        countDownFuture.set(null)
+      } else {
+        countDownFuture.failed(error)
+      }
 
       for( (id, action) <- actions ) {
         if( !action.enqueues.isEmpty ) {
@@ -560,12 +575,17 @@ class DBManager(val parent:LevelDBStore)
       uowStoringCounter += uows.size
       flushSource.suspend
       writeExecutor {
-        client.store(uows)
+        val e = try {
+          client.store(uows)
+          null
+        } catch {
+          case e:Throwable => e
+        }
         flushSource.resume
         dispatchQueue {
           uowStoredCounter += uows.size
           uows.foreach { uow=>
-            uow.onCompleted
+            uow.onCompleted(e)
           }
         }
       }

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBClient.scala
Mon Jul  1 17:38:13 2013
@@ -21,6 +21,7 @@ import java.{lang=>jl}
 import java.{util=>ju}
 
 import java.util.concurrent.locks.ReentrantReadWriteLock
+import java.util.concurrent.atomic.AtomicBoolean
 import collection.immutable.TreeMap
 import collection.mutable.{HashMap, ListBuffer}
 import org.iq80.leveldb._
@@ -30,10 +31,10 @@ import record.{CollectionKey, EntryKey, 
 import org.apache.activemq.leveldb.util._
 import java.util.concurrent._
 import org.fusesource.hawtbuf._
-import java.io.{ObjectInputStream, ObjectOutputStream, File}
+import java.io.{IOException, ObjectInputStream, ObjectOutputStream, File}
 import scala.Option._
 import org.apache.activemq.command.{MessageAck, Message}
-import org.apache.activemq.util.ByteSequence
+import org.apache.activemq.util.{IOExceptionSupport, ByteSequence}
 import java.text.SimpleDateFormat
 import java.util.{Date, Collections}
 import org.apache.activemq.leveldb.util.TimeMetric
@@ -505,12 +506,26 @@ class LevelDBClient(store: LevelDBStore)
     }
   }
 
-  def retry[T](func : =>T):T = RetrySupport.retry(LevelDBClient, store.isStarted, func
_)
+  def might_fail[T](func : =>T):T = {
+    def handleFailure(e:IOException) = {
+      store.stop()
+      if( store.broker_service !=null ) {
+        store.broker_service.handleIOException(e);
+      }
+      throw e;
+    }
+    try {
+      func
+    } catch {
+      case e:IOException => handleFailure(e)
+      case e:Throwable => handleFailure(IOExceptionSupport.create(e))
+    }
+  }
 
   def start() = {
     init()
     replay_init()
-    retry {
+    might_fail {
       log.open()
     }
     replay_from(lastIndexSnapshotPos, log.appender_limit)
@@ -605,7 +620,7 @@ class LevelDBClient(store: LevelDBStore)
     snapshots.filterNot(_._1 == lastIndexSnapshotPos).foreach( _._2.recursiveDelete )
     tempIndexFile.recursiveDelete
 
-    retry {
+    might_fail {
       // Setup the plist index.
       plistIndexFile.recursiveDelete
       plistIndexFile.mkdirs()
@@ -638,7 +653,7 @@ class LevelDBClient(store: LevelDBStore)
   }
 
   def replay_from(from:Long, limit:Long) = {
-    retry {
+    might_fail {
       try {
         // Update the index /w what was stored on the logs..
         var pos = from;
@@ -906,7 +921,7 @@ class LevelDBClient(store: LevelDBStore)
     }
   }
 
-  def retryUsingIndex[T](func: =>T):T = retry(usingIndex( func ))
+  def might_fail_using_index[T](func: =>T):T = might_fail(usingIndex( func ))
 
   /**
    * TODO: expose this via management APIs, handy if you want to
@@ -988,7 +1003,7 @@ class LevelDBClient(store: LevelDBStore)
       log.close
       locked_purge
     } finally {
-      retry {
+      might_fail {
         log.open()
       }
       resume()
@@ -1011,7 +1026,7 @@ class LevelDBClient(store: LevelDBStore)
   def addCollection(record: CollectionRecord.Buffer) = {
     val key = encodeLongKey(COLLECTION_PREFIX, record.getKey)
     val value = record.toUnframedBuffer
-    retryUsingIndex {
+    might_fail_using_index {
       log.appender { appender =>
         appender.append(LOG_ADD_COLLECTION, value)
         index.put(key, value.toByteArray)
@@ -1024,7 +1039,7 @@ class LevelDBClient(store: LevelDBStore)
 
   def listCollections: Seq[(Long, CollectionRecord.Buffer)] = {
     val rc = ListBuffer[(Long, CollectionRecord.Buffer)]()
-    retryUsingIndex {
+    might_fail_using_index {
       val ro = new ReadOptions
       ro.verifyChecksums(verifyChecksums)
       ro.fillCache(false)
@@ -1041,7 +1056,7 @@ class LevelDBClient(store: LevelDBStore)
     val value = encodeVLong(collectionKey)
     val entryKeyPrefix = encodeLongKey(ENTRY_PREFIX, collectionKey)
     collectionMeta.remove(collectionKey)
-    retryUsingIndex {
+    might_fail_using_index {
       log.appender { appender =>
         appender.append(LOG_REMOVE_COLLECTION, new Buffer(value))
       }
@@ -1073,7 +1088,7 @@ class LevelDBClient(store: LevelDBStore)
     meta.size = 0
     meta.last_key = null
     
-    retryUsingIndex {
+    might_fail_using_index {
       index.get(key).foreach { collectionData =>
         log.appender { appender =>
           appender.append(LOG_REMOVE_COLLECTION, new Buffer(value))
@@ -1136,7 +1151,7 @@ class LevelDBClient(store: LevelDBStore)
   }
 
   def getAckPosition(subKey: Long): Long = {
-    retryUsingIndex {
+    might_fail_using_index {
       index.get(encodeEntryKey(ENTRY_PREFIX, subKey, ACK_POSITION)).map{ value=>
         val record = decodeEntryRecord(value)
         record.getValueLocation()
@@ -1173,7 +1188,7 @@ class LevelDBClient(store: LevelDBStore)
     ro.verifyChecksums(verifyChecksums)
     val start = encodeEntryKey(ENTRY_PREFIX, collectionKey, cursorPosition)
     val end = encodeLongKey(ENTRY_PREFIX, collectionKey+1)
-    retryUsingIndex {
+    might_fail_using_index {
       index.cursorRange(start, end, ro) { case (key, value) =>
         func(key.buffer.moveHead(9), EntryRecord.FACTORY.parseUnframed(value))
       }
@@ -1187,7 +1202,7 @@ class LevelDBClient(store: LevelDBStore)
   def collectionIsEmpty(collectionKey: Long) = {
     val entryKeyPrefix = encodeLongKey(ENTRY_PREFIX, collectionKey)
     var empty = true
-    retryUsingIndex {
+    might_fail_using_index {
       val ro = new ReadOptions
       ro.fillCache(false)
       ro.verifyChecksums(verifyChecksums)
@@ -1205,7 +1220,7 @@ class LevelDBClient(store: LevelDBStore)
   val max_index_write_latency = TimeMetric()
 
   def store(uows: Array[DelayableUOW]) {
-    retryUsingIndex {
+    might_fail_using_index {
       log.appender { appender =>
         val syncNeeded = index.write(new WriteOptions, max_index_write_latency) { batch =>
           write_uows(uows, appender, batch)
@@ -1378,7 +1393,7 @@ class LevelDBClient(store: LevelDBStore)
     val ro = new ReadOptions
     ro.verifyChecksums(verifyChecksums)
     ro.fillCache(true)
-    retryUsingIndex {
+    might_fail_using_index {
       index.snapshot { snapshot =>
         ro.snapshot(snapshot)
         val start = encodeEntryKey(ENTRY_PREFIX, collectionKey, firstSeq)
@@ -1457,7 +1472,7 @@ class LevelDBClient(store: LevelDBStore)
 
     // Delete message refs for topics who's consumers have advanced..
     if( !topicPositions.isEmpty ) {
-      retryUsingIndex {
+      might_fail_using_index {
         index.write(new WriteOptions, max_index_write_latency) { batch =>
           for( (topic, first) <- topicPositions ) {
             val ro = new ReadOptions
@@ -1498,7 +1513,7 @@ class LevelDBClient(store: LevelDBStore)
   def removePlist(collectionKey: Long) = {
     val entryKeyPrefix = encodeLong(collectionKey)
     collectionMeta.remove(collectionKey)
-    retry {
+    might_fail {
       val ro = new ReadOptions
       ro.fillCache(false)
       ro.verifyChecksums(false)

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
Mon Jul  1 17:38:13 2013
@@ -31,7 +31,7 @@ import java.util._
 import collection.mutable.ListBuffer
 import org.apache.activemq.broker.jmx.{BrokerMBeanSupport, AnnotatedMBean}
 import org.apache.activemq.util._
-import org.apache.activemq.leveldb.util.{RetrySupport, Log}
+import org.apache.activemq.leveldb.util.Log
 import org.apache.activemq.store.PList.PListIterator
 import org.fusesource.hawtbuf.{UTF8Buffer, DataByteArrayOutputStream}
 import org.fusesource.hawtdispatch;
@@ -201,8 +201,6 @@ class LevelDBStore extends LockableServi
     BrokerMBeanSupport.createPersistenceAdapterName(brokerON.toString, this.toString)
   }
 
-  def retry[T](func : =>T):T = RetrySupport.retry(LevelDBStore, isStarted, func _)
-
   var snappyCompressLogs = false
 
   def doStart: Unit = {

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/RecordLog.scala
Mon Jul  1 17:38:13 2013
@@ -111,7 +111,7 @@ case class RecordLog(directory: File, lo
     override def open = new RandomAccessFile(file, "rw")
 
     override def dispose() = {
-      force
+      flush
       super.dispose()
     }
 

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/ElectingLevelDBStore.scala
Mon Jul  1 17:38:13 2013
@@ -207,10 +207,12 @@ class ElectingLevelDBStore extends Proxy
   def start_master(func: (Int) => Unit) = {
     assert(master==null)
     master = create_master()
+    master_started.set(true)
     master.blocking_executor.execute(^{
-      master_started.set(true)
       master.start();
       master_started_latch.countDown()
+    })
+    master.blocking_executor.execute(^{
       func(master.getPort)
     })
   }

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterElector.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterElector.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterElector.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterElector.scala
Mon Jul  1 17:38:13 2013
@@ -82,7 +82,7 @@ class MasterElector(store: ElectingLevel
     def disconnected = changed
 
     def changed:Unit = elector.synchronized {
-      // info(eid+" cluster state changed: "+members)
+//      info(eid+" cluster state changed: "+members)
       if (isMaster) {
         // We are the master elector, we will choose which node will startup the MasterLevelDBStore
         members.get(store.brokerName) match {
@@ -91,7 +91,8 @@ class MasterElector(store: ElectingLevel
           case Some(members) =>
 
             if (members.size < store.clusterSizeQuorum) {
-              info("Not enough cluster members connected to elect a new master.")
+              info("Not enough cluster members connected to elect a master.")
+              elected = null
             } else {
 
               // If we already elected a master, lets make sure he is still online..

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterLevelDBStore.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterLevelDBStore.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterLevelDBStore.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/MasterLevelDBStore.scala
Mon Jul  1 17:38:13 2013
@@ -86,6 +86,8 @@ class MasterLevelDBStore extends LevelDB
     unstash(directory)
     super.doStart
     start_protocol_server
+    // Lets not complete the startup until at least one slave is synced up.
+    wal_sync_to(wal_append_position)
   }
 
   override def doStop(stopper: ServiceStopper): Unit = {
@@ -103,6 +105,7 @@ class MasterLevelDBStore extends LevelDB
   // Replication Protocol Stuff
   //////////////////////////////////////
   var transport_server:TransportServer = _
+  val start_latch = new CountDownLatch(1)
 
   def start_protocol_server = {
     transport_server = new TcpTransportServer(new URI(bind))
@@ -118,14 +121,16 @@ class MasterLevelDBStore extends LevelDB
         warn(error)
       }
     })
-    val start_latch = new CountDownLatch(1)
     transport_server.start(^{
       start_latch.countDown()
     })
     start_latch.await()
   }
 
-  def getPort = transport_server.getSocketAddress.asInstanceOf[InetSocketAddress].getPort
+  def getPort = {
+    start_latch.await()
+    transport_server.getSocketAddress.asInstanceOf[InetSocketAddress].getPort
+  }
 
   def stop_protocol_server = {
     transport_server.stop(NOOP)
@@ -330,6 +335,11 @@ class MasterLevelDBStore extends LevelDB
     if( minSlaveAcks<1 || (syncToMask & SYNC_TO_REMOTE)==0) {
       return
     }
+
+    if( isStopped ) {
+      throw new IllegalStateException("Store replication stopped")
+    }
+
     val position_sync = new PositionSync(position, minSlaveAcks)
     this.position_sync = position_sync
     for( slave <- slaves.values() ) {
@@ -337,6 +347,9 @@ class MasterLevelDBStore extends LevelDB
     }
 
     while( !position_sync.await(1, TimeUnit.SECONDS) ) {
+      if( isStopped ) {
+        throw new IllegalStateException("Store replication stopped")
+      }
       val status = slaves.values().map(_.status).mkString(", ")
       warn("Store update waiting on %d replica(s) to catch up to log position %d. Connected
slaves: [%s]", minSlaveAcks, position, status)
     }

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/SlaveLevelDBStore.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/SlaveLevelDBStore.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/SlaveLevelDBStore.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/replicated/SlaveLevelDBStore.scala
Mon Jul  1 17:38:13 2013
@@ -378,7 +378,7 @@ class SlaveLevelDBStore extends LevelDBS
 
       session.request_then(DISCONNECT_ACTION, null) { body =>
         // Ok we are now caught up.
-        status = "Synchronize"
+        status = "Synchronized"
         info(status)
         stash_clear(directory) // we don't need the stash anymore.
         transport.stop(NOOP)

Modified: activemq/trunk/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBStoreTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBStoreTest.java?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBStoreTest.java
(original)
+++ activemq/trunk/activemq-leveldb-store/src/test/java/org/apache/activemq/leveldb/test/ReplicatedLevelDBStoreTest.java
Mon Jul  1 17:38:13 2013
@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.leveldb.test;
 
+import org.apache.activemq.Service;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.leveldb.CountDownFuture;
 import org.apache.activemq.leveldb.LevelDBStore;
@@ -30,6 +31,7 @@ import org.slf4j.LoggerFactory;
 import java.io.File;
 import java.util.ArrayList;
 import java.util.LinkedList;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import static org.apache.activemq.leveldb.test.ReplicationTestSupport.addMessage;
@@ -49,32 +51,30 @@ public class ReplicatedLevelDBStoreTest 
         FileSupport.toRichFile(masterDir).recursiveDelete();
         FileSupport.toRichFile(slaveDir).recursiveDelete();
 
-        MasterLevelDBStore master = createMaster(masterDir);
+        final MasterLevelDBStore master = createMaster(masterDir);
         master.setReplicas(2);
-        master.start();
+        CountDownFuture masterStartLatch = asyncStart(master);
 
-        MessageStore ms = master.createQueueMessageStore(new ActiveMQQueue("TEST"));
-
-        // Updating the store should not complete since we don't have enough
+        // Start the store should not complete since we don't have enough
         // replicas.
-        CountDownFuture f = asyncAddMessage(ms, "m1");
-        assertFalse(f.await(2, TimeUnit.SECONDS));
+        assertFalse(masterStartLatch.await(2, TimeUnit.SECONDS));
 
-        // Adding a slave should allow that update to complete.
+        // Adding a slave should allow the master startup to complete.
         SlaveLevelDBStore slave = createSlave(master, slaveDir);
         slave.start();
 
-        assertTrue(f.await(2, TimeUnit.SECONDS));
+        assertTrue(masterStartLatch.await(2, TimeUnit.SECONDS));
 
         // New updates should complete quickly now..
-        f = asyncAddMessage(ms, "m2");
+        MessageStore ms = master.createQueueMessageStore(new ActiveMQQueue("TEST"));
+        CountDownFuture f = asyncAddMessage(ms, "m1");
         assertTrue(f.await(1, TimeUnit.SECONDS));
 
         // If the slave goes offline, then updates should once again
         // not complete.
         slave.stop();
 
-        f = asyncAddMessage(ms, "m3");
+        f = asyncAddMessage(ms, "m2");
         assertFalse(f.await(2, TimeUnit.SECONDS));
 
         // Restart and the op should complete.
@@ -102,6 +102,20 @@ public class ReplicatedLevelDBStoreTest 
         return f;
     }
 
+    private CountDownFuture asyncStart(final Service service) {
+        final CountDownFuture<Throwable> f = new CountDownFuture<Throwable>();
+        LevelDBStore.BLOCKING_EXECUTOR().execute(new Runnable() {
+            public void run() {
+                try {
+                    service.start();
+                    f.set(null);
+                } catch (Throwable e) {
+                    f.set(e);
+                }
+            }
+        });
+        return f;
+    }
 
     @Test(timeout = 1000*60*60)
     public void testReplication() throws Exception {
@@ -120,10 +134,11 @@ public class ReplicatedLevelDBStoreTest 
         for (int j = 0; j < 10; j++) {
 
             MasterLevelDBStore master = createMaster(directories.get(0));
-            master.start();
+            CountDownFuture masterStart = asyncStart(master);
             SlaveLevelDBStore slave1 = createSlave(master, directories.get(1));
             SlaveLevelDBStore slave2 = createSlave(master, directories.get(2));
-            slave2.start();
+            asyncStart(slave2);
+            masterStart.await();
 
             LOG.info("Adding messages...");
             MessageStore ms = master.createQueueMessageStore(new ActiveMQQueue("TEST"));

Modified: activemq/trunk/activemq-leveldb-store/src/test/scala/org/apache/activemq/leveldb/dfs/DFSLevelDBClient.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/test/scala/org/apache/activemq/leveldb/dfs/DFSLevelDBClient.scala?rev=1498601&r1=1498600&r2=1498601&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/test/scala/org/apache/activemq/leveldb/dfs/DFSLevelDBClient.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/test/scala/org/apache/activemq/leveldb/dfs/DFSLevelDBClient.scala
Mon Jul  1 17:38:13 2013
@@ -84,7 +84,7 @@ class DFSLevelDBClient(val store:DFSLeve
   def remoteIndexPath = new Path(dfsDirectory, "index")
 
   override def start() = {
-    retry {
+    might_fail {
       directory.mkdirs()
       dfs.mkdirs(dfsDirectory)
       downloadLogFiles



Mime
View raw message