X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Futils%2FInMemoryJournal.java;h=f340d1c305c5939993d09de8a4e84ef630791b1f;hb=1e25f44b70aab9998f33b2136a0cf628821cd2dc;hp=34867530821d70d4ce7085859a3f79e59a6d9a03;hpb=c36bb2bc93ea0fbdd3ca8b2663447fc921450aee;p=controller.git

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/InMemoryJournal.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/InMemoryJournal.java
index 3486753082..f340d1c305 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/InMemoryJournal.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/InMemoryJournal.java
@@ -32,6 +32,8 @@ public class InMemoryJournal extends AsyncWriteJournal {
 
     private static final Map<String, CountDownLatch> deleteMessagesCompleteLatches = new ConcurrentHashMap<>();
 
+    private static final Map<String, CountDownLatch> blockReadMessagesLatches = new ConcurrentHashMap<>();
+
     public static void addEntry(String persistenceId, long sequenceNr, Object data) {
         Map<Long, Object> journal = journals.get(persistenceId);
         if(journal == null) {
@@ -62,12 +64,21 @@ public class InMemoryJournal extends AsyncWriteJournal {
         deleteMessagesCompleteLatches.put(persistenceId, new CountDownLatch(1));
     }
 
+    public static void addBlockReadMessagesLatch(String persistenceId, CountDownLatch latch) {
+        blockReadMessagesLatches.put(persistenceId, latch);
+    }
+
     @Override
     public Future<Void> doAsyncReplayMessages(final String persistenceId, long fromSequenceNr,
             long toSequenceNr, long max, final Procedure<PersistentRepr> replayCallback) {
         return Futures.future(new Callable<Void>() {
             @Override
             public Void call() throws Exception {
+                CountDownLatch blockLatch = blockReadMessagesLatches.remove(persistenceId);
+                if(blockLatch != null) {
+                    Uninterruptibles.awaitUninterruptibly(blockLatch);
+                }
+
                 Map<Long, Object> journal = journals.get(persistenceId);
                 if(journal == null) {
                     return null;