geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hiteshkhame...@apache.org
Subject [27/50] [abbrv] geode git commit: GEODE-2230: Added AsyncEventQueue and GatewaySender waitUntilFlushed API
Date Tue, 14 Feb 2017 19:32:00 GMT
GEODE-2230: Added AsyncEventQueue and GatewaySender waitUntilFlushed API


Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/44cd72d8
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/44cd72d8
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/44cd72d8

Branch: refs/heads/master
Commit: 44cd72d8502d278a9a328a2b0a825c2460f8a383
Parents: a15daf0
Author: Barry Oglesby <boglesby@pivotal.io>
Authored: Thu Jan 19 17:07:09 2017 -0800
Committer: Barry Oglesby <boglesby@pivotal.io>
Committed: Tue Jan 24 16:52:37 2017 -0800

----------------------------------------------------------------------
 .../internal/AsyncEventQueueImpl.java           |   5 +
 .../org/apache/geode/internal/DSFIDFactory.java |   6 +-
 .../geode/internal/DataSerializableFixedID.java |   2 +
 .../geode/internal/cache/BucketRegionQueue.java |  43 +++
 .../cache/wan/AbstractGatewaySender.java        |  47 ++-
 ...aitUntilGatewaySenderFlushedCoordinator.java |  43 +++
 ...ParallelGatewaySenderFlushedCoordinator.java | 323 +++++++++++++++++++
 .../geode/internal/i18n/LocalizedStrings.java   |   9 +
 ...atewaySenderFlushedCoordinatorJUnitTest.java |  62 ++++
 ...atewaySenderFlushedCoordinatorJUnitTest.java | 126 ++++++++
 .../sanctionedDataSerializables.txt             |   4 +
 .../apache/geode/cache/lucene/LuceneIndex.java  |   9 +-
 .../cache/lucene/internal/LuceneIndexImpl.java  |  24 +-
 .../internal/xml/LuceneIndexCreation.java       |   3 +-
 .../LuceneIndexMaintenanceIntegrationTest.java  |  24 +-
 .../geode/cache/lucene/LuceneQueriesBase.java   |   3 +-
 .../lucene/LuceneQueriesIntegrationTest.java    |  22 +-
 ...LuceneQueriesPersistenceIntegrationTest.java |   3 +-
 .../internal/LuceneIndexImplJUnitTest.java      |   8 +-
 .../LuceneIndexRecoveryHAIntegrationTest.java   |  11 +-
 .../cli/LuceneIndexCommandsDUnitTest.java       |   3 +-
 .../DumpDirectoryFilesIntegrationTest.java      |   3 +-
 .../cache/lucene/test/LuceneTestUtilities.java  |   6 +-
 23 files changed, 722 insertions(+), 67 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueImpl.java b/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueImpl.java
index 2eb53be..3b99f1c 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueImpl.java
@@ -17,6 +17,7 @@ package org.apache.geode.cache.asyncqueue.internal;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.geode.cache.asyncqueue.AsyncEventListener;
 import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
@@ -202,4 +203,8 @@ public class AsyncEventQueueImpl implements AsyncEventQueue {
   public boolean isForwardExpirationDestroy() {
     return ((AbstractGatewaySender) this.sender).isForwardExpirationDestroy();
   }
+
+  public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException {
+    return ((AbstractGatewaySender) this.sender).waitUntilFlushed(timeout, unit);
+  }
 }

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java b/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
index e13a2ad..bb29239 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
@@ -23,7 +23,6 @@ import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
-import java.io.NotSerializableException;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 
@@ -76,7 +75,6 @@ import org.apache.geode.cache.query.internal.types.StructTypeImpl;
 import org.apache.geode.distributed.internal.DistributionAdvisor;
 import org.apache.geode.distributed.internal.HighPriorityAckedMessage;
 import org.apache.geode.distributed.internal.ReplyMessage;
-import org.apache.geode.distributed.internal.ReplyProcessor21;
 import org.apache.geode.distributed.internal.SerialAckedMessage;
 import org.apache.geode.distributed.internal.ShutdownMessage;
 import org.apache.geode.distributed.internal.StartupMessage;
@@ -393,11 +391,11 @@ import org.apache.geode.internal.cache.versions.VMVersionTag;
 import org.apache.geode.internal.cache.wan.GatewaySenderAdvisor;
 import org.apache.geode.internal.cache.wan.GatewaySenderEventCallbackArgument;
 import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
+import org.apache.geode.internal.cache.wan.parallel.WaitUntilParallelGatewaySenderFlushedCoordinator;
 import org.apache.geode.internal.cache.wan.parallel.ParallelQueueBatchRemovalMessage;
 import org.apache.geode.internal.cache.wan.parallel.ParallelQueueBatchRemovalMessage.BatchRemovalReplyMessage;
 import org.apache.geode.internal.cache.wan.parallel.ParallelQueueRemovalMessage;
 import org.apache.geode.internal.cache.wan.serial.BatchDestroyOperation;
-import org.apache.geode.internal.i18n.LocalizedStrings;
 import org.apache.geode.management.internal.JmxManagerAdvisor.JmxManagerProfile;
 import org.apache.geode.management.internal.JmxManagerAdvisor.JmxManagerProfileMessage;
 import org.apache.geode.management.internal.JmxManagerLocatorRequest;
@@ -924,6 +922,8 @@ public final class DSFIDFactory implements DataSerializableFixedID {
     registerDSFID(PR_DESTROY_ON_DATA_STORE_MESSAGE, DestroyRegionOnDataStoreMessage.class);
     registerDSFID(SHUTDOWN_ALL_GATEWAYHUBS_REQUEST, ShutdownAllGatewayHubsRequest.class);
     registerDSFID(BUCKET_COUNT_LOAD_PROBE, BucketCountLoadProbe.class);
+    registerDSFID(WAIT_UNTIL_GATEWAY_SENDER_FLUSHED_MESSAGE,
+        WaitUntilParallelGatewaySenderFlushedCoordinator.WaitUntilGatewaySenderFlushedMessage.class);
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/main/java/org/apache/geode/internal/DataSerializableFixedID.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/DataSerializableFixedID.java b/geode-core/src/main/java/org/apache/geode/internal/DataSerializableFixedID.java
index 24b452d..84eb8e9 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/DataSerializableFixedID.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/DataSerializableFixedID.java
@@ -810,6 +810,8 @@ public interface DataSerializableFixedID extends SerializationVersions {
   public static final short LUCENE_TOP_ENTRIES = 2175;
   public static final short LUCENE_TOP_ENTRIES_COLLECTOR = 2176;
 
+  public static final short WAIT_UNTIL_GATEWAY_SENDER_FLUSHED_MESSAGE = 2177;
+
   // NOTE, codes > 65535 will take 4 bytes to serialize
 
   /**

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegionQueue.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegionQueue.java b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegionQueue.java
index 7fea789..4830912 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegionQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegionQueue.java
@@ -30,6 +30,8 @@ import java.util.TreeSet;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.logging.log4j.Logger;
@@ -78,6 +80,10 @@ public class BucketRegionQueue extends AbstractBucketRegionQueue {
 
   private long lastKeyRecovered;
 
+  private AtomicLong latestQueuedKey = new AtomicLong();
+
+  private AtomicLong latestAcknowledgedKey = new AtomicLong();
+
   /**
    * @param regionName
    * @param attrs
@@ -437,6 +443,7 @@ public class BucketRegionQueue extends AbstractBucketRegionQueue {
     if (didPut) {
       if (this.initialized) {
         this.eventSeqNumQueue.add(key);
+        updateLargestQueuedKey((Long) key);
       }
       if (logger.isDebugEnabled()) {
         logger.debug("Put successfully in the queue : {} was initialized: {}",
@@ -448,6 +455,41 @@ public class BucketRegionQueue extends AbstractBucketRegionQueue {
     }
   }
 
+  private void updateLargestQueuedKey(Long key) {
+    Atomics.setIfGreater(this.latestQueuedKey, key);
+  }
+
+  private void setLatestAcknowledgedKey(Long key) {
+    this.latestAcknowledgedKey.set(key);
+  }
+
+  public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException {
+    if (logger.isDebugEnabled()) {
+      logger.debug("BucketRegionQueue: waitUntilFlushed bucket=" + getId() + "; time="
+          + System.currentTimeMillis() + "; timeout=" + timeout + "; unit=" + unit);
+    }
+    boolean result = false;
+    // Wait until latestAcknowledgedKey > latestQueuedKey or the queue is empty
+    if (this.initialized) {
+      long latestQueuedKeyToCheck = this.latestQueuedKey.get();
+      long nanosRemaining = unit.toNanos(timeout);
+      long endTime = System.nanoTime() + nanosRemaining;
+      while (nanosRemaining > 0) {
+        if (latestAcknowledgedKey.get() > latestQueuedKeyToCheck || isEmpty()) {
+          result = true;
+          break;
+        }
+        Thread.sleep(Math.min(TimeUnit.NANOSECONDS.toMillis(nanosRemaining) + 1, 100));
+        nanosRemaining = endTime - System.nanoTime();
+      }
+    }
+    if (logger.isDebugEnabled()) {
+      logger.debug("BucketRegionQueue: waitUntilFlushed completed bucket=" + getId() + "; time="
+          + System.currentTimeMillis() + "; result=" + result);
+    }
+    return result;
+  }
+
   /**
    * It removes the first key from the queue.
    * 
@@ -503,6 +545,7 @@ public class BucketRegionQueue extends AbstractBucketRegionQueue {
       event.setEventId(new EventID(cache.getSystem()));
       event.setRegion(this);
       basicDestroy(event, true, null);
+      setLatestAcknowledgedKey((Long) key);
       checkReadiness();
     } catch (EntryNotFoundException enf) {
       if (getPartitionedRegion().isDestroyed()) {

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
index 2c9a65d..1f8704c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
@@ -22,9 +22,13 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import org.apache.geode.InternalGemFireError;
+import org.apache.geode.internal.cache.execute.BucketMovedException;
 import org.apache.geode.internal.cache.ha.ThreadIdentifier;
+import org.apache.geode.internal.cache.wan.parallel.WaitUntilParallelGatewaySenderFlushedCoordinator;
 import org.apache.logging.log4j.Logger;
 
 import org.apache.geode.CancelCriterion;
@@ -33,15 +37,12 @@ import org.apache.geode.cache.AttributesFactory;
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.CacheException;
 import org.apache.geode.cache.DataPolicy;
-import org.apache.geode.cache.Operation;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionAttributes;
 import org.apache.geode.cache.RegionDestroyedException;
 import org.apache.geode.cache.RegionExistsException;
 import org.apache.geode.cache.Scope;
 import org.apache.geode.cache.asyncqueue.AsyncEventListener;
-import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
-import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueStats;
 import org.apache.geode.cache.client.internal.LocatorDiscoveryCallback;
 import org.apache.geode.cache.client.internal.PoolImpl;
 import org.apache.geode.cache.wan.GatewayEventFilter;
@@ -675,7 +676,7 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
     }
   }
 
-  final public RegionQueue getQueue() {
+  public RegionQueue getQueue() {
     if (this.eventProcessor != null) {
       if (!(this.eventProcessor instanceof ConcurrentSerialGatewaySenderEventProcessor)) {
         return this.eventProcessor.getQueue();
@@ -1089,7 +1090,7 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
     return substituteValue;
   }
 
-  private void initializeEventIdIndex() {
+  protected void initializeEventIdIndex() {
     final boolean isDebugEnabled = logger.isDebugEnabled();
 
     boolean gotLock = false;
@@ -1240,6 +1241,42 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
     return lifeCycleLock;
   }
 
+  public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException {
+    int attempts = 0;
+    boolean result = false;
+    if (isParallel()) {
+      // Wait until the sender is flushed. Retry if necessary.
+      while (true) {
+        try {
+          WaitUntilParallelGatewaySenderFlushedCoordinator coordinator =
+              new WaitUntilParallelGatewaySenderFlushedCoordinator(this, timeout, unit, true);
+          result = coordinator.waitUntilFlushed();
+          break;
+        } catch (BucketMovedException | CancelException | RegionDestroyedException e) {
+          attempts++;
+          logger.warn(
+              LocalizedStrings.AbstractGatewaySender_CAUGHT_EXCEPTION_ATTEMPTING_WAIT_UNTIL_FLUSHED_RETRYING
+                  .toLocalizedString(),
+              e);
+          Thread.sleep(100);
+        } catch (Throwable t) {
+          attempts++;
+          logger.warn(
+              LocalizedStrings.AbstractGatewaySender_CAUGHT_EXCEPTION_ATTEMPTING_WAIT_UNTIL_FLUSHED_RETURNING
+                  .toLocalizedString(),
+              t);
+          throw new InternalGemFireError(t);
+        }
+      }
+      return result;
+    } else {
+      // Serial senders are currently not supported
+      throw new UnsupportedOperationException(
+          LocalizedStrings.AbstractGatewaySender_WAIT_UNTIL_FLUSHED_NOT_SUPPORTED_FOR_SERIAL_SENDERS
+              .toLocalizedString());
+    }
+  }
+
   /**
    * Has a reference to a GatewayEventImpl and has a timeout value.
    */

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/main/java/org/apache/geode/internal/cache/wan/WaitUntilGatewaySenderFlushedCoordinator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/WaitUntilGatewaySenderFlushedCoordinator.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/WaitUntilGatewaySenderFlushedCoordinator.java
new file mode 100644
index 0000000..07c4e19
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/WaitUntilGatewaySenderFlushedCoordinator.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.internal.cache.wan;
+
+import org.apache.geode.internal.logging.LogService;
+import org.apache.logging.log4j.Logger;
+
+import java.util.concurrent.TimeUnit;
+
+public abstract class WaitUntilGatewaySenderFlushedCoordinator {
+
+  protected AbstractGatewaySender sender;
+
+  protected long timeout;
+
+  protected TimeUnit unit;
+
+  protected boolean initiator;
+
+  protected static final Logger logger = LogService.getLogger();
+
+  public WaitUntilGatewaySenderFlushedCoordinator(AbstractGatewaySender sender, long timeout,
+      TimeUnit unit, boolean initiator) {
+    this.sender = sender;
+    this.timeout = timeout;
+    this.unit = unit;
+    this.initiator = initiator;
+  }
+
+  public abstract boolean waitUntilFlushed() throws Throwable;
+}

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator.java
new file mode 100644
index 0000000..a4c03a9
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator.java
@@ -0,0 +1,323 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.internal.cache.wan.parallel;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.cache.Cache;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.*;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.*;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
+import org.apache.geode.internal.cache.wan.WaitUntilGatewaySenderFlushedCoordinator;
+import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderQueue;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.*;
+
+public class WaitUntilParallelGatewaySenderFlushedCoordinator
+    extends WaitUntilGatewaySenderFlushedCoordinator {
+
+  public WaitUntilParallelGatewaySenderFlushedCoordinator(AbstractGatewaySender sender,
+      long timeout, TimeUnit unit, boolean initiator) {
+    super(sender, timeout, unit, initiator);
+  }
+
+  public boolean waitUntilFlushed() throws Throwable {
+    boolean remoteResult = true, localResult = true;
+    Throwable exceptionToThrow = null;
+    ConcurrentParallelGatewaySenderQueue prq =
+        (ConcurrentParallelGatewaySenderQueue) this.sender.getQueue();
+    PartitionedRegion pr = (PartitionedRegion) prq.getRegion();
+
+    // Create callables for local buckets
+    List<WaitUntilBucketRegionQueueFlushedCallable> callables =
+        buildWaitUntilBucketRegionQueueFlushedCallables(pr);
+
+    // Submit local callables for execution
+    ExecutorService service = this.sender.getDistributionManager().getWaitingThreadPool();
+    List<Future<Boolean>> callableFutures = new ArrayList<>();
+    for (Callable<Boolean> callable : callables) {
+      callableFutures.add(service.submit(callable));
+    }
+    if (logger.isDebugEnabled()) {
+      logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Created and submitted "
+          + callables.size() + " callables=" + callables);
+    }
+
+    // Send message to remote buckets
+    if (this.initiator) {
+      remoteResult = false;
+      try {
+        remoteResult = waitUntilFlushedOnRemoteMembers(pr);
+      } catch (Throwable t) {
+        exceptionToThrow = t;
+      }
+      if (logger.isDebugEnabled()) {
+        logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Processed remote result="
+            + remoteResult + "; exceptionToThrow=" + exceptionToThrow);
+      }
+    }
+
+    // Process local future results
+    for (Future<Boolean> future : callableFutures) {
+      boolean singleBucketResult = false;
+      try {
+        singleBucketResult = future.get();
+      } catch (ExecutionException e) {
+        exceptionToThrow = e.getCause();
+      }
+      localResult = localResult && singleBucketResult;
+    }
+    if (logger.isDebugEnabled()) {
+      logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Processed local result="
+          + localResult + "; exceptionToThrow=" + exceptionToThrow);
+    }
+
+    // Return the full result
+    if (exceptionToThrow == null) {
+      if (logger.isDebugEnabled()) {
+        logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Returning full result="
+            + (remoteResult && localResult));
+      }
+      return remoteResult && localResult;
+    } else {
+      throw exceptionToThrow;
+    }
+  }
+
+  protected List<WaitUntilBucketRegionQueueFlushedCallable> buildWaitUntilBucketRegionQueueFlushedCallables(
+      PartitionedRegion pr) {
+    List<WaitUntilBucketRegionQueueFlushedCallable> callables = new ArrayList<>();
+    if (pr.isDataStore()) {
+      for (BucketRegion br : pr.getDataStore().getAllLocalBucketRegions()) {
+        callables.add(new WaitUntilBucketRegionQueueFlushedCallable((BucketRegionQueue) br,
+            this.timeout, this.unit));
+      }
+    }
+    return callables;
+  }
+
+  protected boolean waitUntilFlushedOnRemoteMembers(PartitionedRegion pr) throws Throwable {
+    boolean result = true;
+    DM dm = this.sender.getDistributionManager();
+    Set<InternalDistributedMember> recipients = pr.getRegionAdvisor().adviseDataStore();
+    if (!recipients.isEmpty()) {
+      if (logger.isDebugEnabled()) {
+        logger.debug(
+            "WaitUntilParallelGatewaySenderFlushedCoordinator: About to send message recipients="
+                + recipients);
+      }
+      WaitUntilGatewaySenderFlushedReplyProcessor processor =
+          new WaitUntilGatewaySenderFlushedReplyProcessor(dm, recipients);
+      WaitUntilGatewaySenderFlushedMessage message = new WaitUntilGatewaySenderFlushedMessage(
+          recipients, processor.getProcessorId(), this.sender.getId(), this.timeout, this.unit);
+      dm.putOutgoing(message);
+      if (logger.isDebugEnabled()) {
+        logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Sent message recipients="
+            + recipients);
+      }
+      try {
+        processor.waitForReplies();
+        result = processor.getCombinedResult();
+      } catch (ReplyException e) {
+        if (logger.isDebugEnabled()) {
+          logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Caught e=" + e
+              + "; cause=" + e.getCause());
+        }
+        throw e.getCause();
+      } catch (InterruptedException e) {
+        dm.getCancelCriterion().checkCancelInProgress(e);
+        Thread.currentThread().interrupt();
+        result = false;
+      }
+    }
+    return result;
+  }
+
+  public static class WaitUntilBucketRegionQueueFlushedCallable implements Callable<Boolean> {
+
+    private BucketRegionQueue brq;
+
+    private long timeout;
+
+    private TimeUnit unit;
+
+    public WaitUntilBucketRegionQueueFlushedCallable(BucketRegionQueue brq, long timeout,
+        TimeUnit unit) {
+      this.brq = brq;
+      this.timeout = timeout;
+      this.unit = unit;
+    }
+
+    @Override
+    public Boolean call() throws Exception {
+      return this.brq.waitUntilFlushed(this.timeout, this.unit);
+    }
+
+    @Override
+    public String toString() {
+      return new StringBuilder().append(getClass().getSimpleName()).append("[").append("brq=")
+          .append(this.brq.getId()).append("]").toString();
+    }
+  }
+
+  public static class WaitUntilGatewaySenderFlushedReplyProcessor extends ReplyProcessor21 {
+
+    private Map<DistributedMember, Boolean> responses;
+
+    public WaitUntilGatewaySenderFlushedReplyProcessor(DM dm, Collection initMembers) {
+      super(dm, initMembers);
+      initializeResponses();
+    }
+
+    private void initializeResponses() {
+      this.responses = new ConcurrentHashMap<>();
+      for (InternalDistributedMember member : getMembers()) {
+        this.responses.put(member, false);
+      }
+    }
+
+    @Override
+    public void process(DistributionMessage msg) {
+      try {
+        if (msg instanceof ReplyMessage) {
+          ReplyMessage reply = (ReplyMessage) msg;
+          if (logger.isDebugEnabled()) {
+            logger
+                .debug("WaitUntilGatewaySenderFlushedReplyProcessor: Processing reply from sender="
+                    + reply.getSender() + "; returnValue=" + reply.getReturnValue() + "; exception="
+                    + reply.getException());
+          }
+          if (reply.getException() == null) {
+            this.responses.put(reply.getSender(), (Boolean) reply.getReturnValue());
+          } else {
+            reply.getException().printStackTrace();
+          }
+        }
+      } finally {
+        super.process(msg);
+      }
+    }
+
+    public boolean getCombinedResult() {
+      boolean combinedResult = true;
+      for (boolean singleMemberResult : this.responses.values()) {
+        combinedResult = combinedResult && singleMemberResult;
+      }
+      if (logger.isDebugEnabled()) {
+        logger.debug("WaitUntilGatewaySenderFlushedReplyProcessor: Returning combinedResult="
+            + combinedResult);
+      }
+      return combinedResult;
+    }
+  }
+
+  public static class WaitUntilGatewaySenderFlushedMessage extends PooledDistributionMessage
+      implements MessageWithReply {
+
+    private int processorId;
+
+    private String gatewaySenderId;
+
+    private long timeout;
+
+    private TimeUnit unit;
+
+    /* For serialization */
+    public WaitUntilGatewaySenderFlushedMessage() {}
+
+    protected WaitUntilGatewaySenderFlushedMessage(Collection recipients, int processorId,
+        String gatewaySenderId, long timeout, TimeUnit unit) {
+      super();
+      setRecipients(recipients);
+      this.processorId = processorId;
+      this.gatewaySenderId = gatewaySenderId;
+      this.timeout = timeout;
+      this.unit = unit;
+    }
+
+    @Override
+    protected void process(DistributionManager dm) {
+      boolean result = false;
+      ReplyException replyException = null;
+      try {
+        if (logger.isDebugEnabled()) {
+          logger.debug("WaitUntilGatewaySenderFlushedMessage: Processing gatewaySenderId="
+              + this.gatewaySenderId + "; timeout=" + this.timeout + "; unit=" + this.unit);
+        }
+        Cache cache = GemFireCacheImpl.getInstance();
+        if (cache != null) {
+          AbstractGatewaySender sender =
+              (AbstractGatewaySender) cache.getGatewaySender(this.gatewaySenderId);
+          if (sender != null) {
+            try {
+              WaitUntilParallelGatewaySenderFlushedCoordinator coordinator =
+                  new WaitUntilParallelGatewaySenderFlushedCoordinator(sender, this.timeout,
+                      this.unit, false);
+              result = coordinator.waitUntilFlushed();
+            } catch (Throwable e) {
+              replyException = new ReplyException(e);
+            }
+          }
+        }
+      } finally {
+        ReplyMessage replyMsg = new ReplyMessage();
+        replyMsg.setRecipient(getSender());
+        replyMsg.setProcessorId(this.processorId);
+        if (replyException == null) {
+          replyMsg.setReturnValue(result);
+        } else {
+          replyMsg.setException(replyException);
+        }
+        if (logger.isDebugEnabled()) {
+          logger.debug("WaitUntilGatewaySenderFlushedMessage: Sending reply returnValue="
+              + replyMsg.getReturnValue() + "; exception=" + replyMsg.getException());
+        }
+        dm.putOutgoing(replyMsg);
+      }
+    }
+
+    @Override
+    public int getDSFID() {
+      return WAIT_UNTIL_GATEWAY_SENDER_FLUSHED_MESSAGE;
+    }
+
+    @Override
+    public void toData(DataOutput out) throws IOException {
+      super.toData(out);
+      out.writeInt(this.processorId);
+      DataSerializer.writeString(this.gatewaySenderId, out);
+      out.writeLong(this.timeout);
+      DataSerializer.writeEnum(this.unit, out);
+    }
+
+    @Override
+    public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+      super.fromData(in);
+      this.processorId = in.readInt();
+      this.gatewaySenderId = DataSerializer.readString(in);
+      this.timeout = in.readLong();
+      this.unit = DataSerializer.readEnum(TimeUnit.class, in);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java b/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java
index c616031..1c214e4 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java
@@ -7665,6 +7665,15 @@ public class LocalizedStrings {
       new StringId(6647,
           "Cannot create GatewaySender {0} because the maximum ({1}) has been reached");
 
+  public static final StringId AbstractGatewaySender_WAIT_UNTIL_FLUSHED_NOT_SUPPORTED_FOR_SERIAL_SENDERS =
+      new StringId(6648, "waitUntilFlushed is not currently supported for serial gateway senders");
+  public static final StringId AbstractGatewaySender_CAUGHT_EXCEPTION_ATTEMPTING_WAIT_UNTIL_FLUSHED_RETRYING =
+      new StringId(6649,
+          "Caught the following exception attempting waitUntilFlushed and will retry:");
+  public static final StringId AbstractGatewaySender_CAUGHT_EXCEPTION_ATTEMPTING_WAIT_UNTIL_FLUSHED_RETURNING =
+      new StringId(6650,
+          "Caught the following exception attempting waitUntilFlushed and will return:");
+
   /** Testing strings, messageId 90000-99999 **/
 
   /**

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/test/java/org/apache/geode/internal/cache/wan/WaitUntilGatewaySenderFlushedCoordinatorJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/wan/WaitUntilGatewaySenderFlushedCoordinatorJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/WaitUntilGatewaySenderFlushedCoordinatorJUnitTest.java
new file mode 100644
index 0000000..17696bb
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/WaitUntilGatewaySenderFlushedCoordinatorJUnitTest.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.internal.cache.wan;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.RegionQueue;
+import org.junit.After;
+import org.junit.Before;
+
+import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
+import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
+import static org.mockito.Mockito.*;
+
+public abstract class WaitUntilGatewaySenderFlushedCoordinatorJUnitTest {
+
+  protected GemFireCacheImpl cache;
+
+  protected AbstractGatewaySender sender;
+
+  @Before
+  public void setUp() {
+    createCache();
+    createGatewaySender();
+  }
+
+  @After
+  public void tearDown() {
+    if (this.cache != null) {
+      this.cache.close();
+    }
+  }
+
+  private void createCache() {
+    this.cache = (GemFireCacheImpl) new CacheFactory().set(MCAST_PORT, "0")
+        .set(LOG_LEVEL, "warning").create();
+  }
+
+  protected void createGatewaySender() {
+    this.sender = spy(AbstractGatewaySender.class);
+    this.sender.cache = this.cache;
+    this.sender.eventProcessor = getEventProcessor();
+  }
+
+  protected RegionQueue getQueue() {
+    return this.sender.eventProcessor.getQueue();
+  }
+
+  protected abstract AbstractGatewaySenderEventProcessor getEventProcessor();
+}

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinatorJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinatorJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinatorJUnitTest.java
new file mode 100644
index 0000000..c8b8ba1
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinatorJUnitTest.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.internal.cache.wan.parallel;
+
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor;
+import org.apache.geode.internal.cache.wan.WaitUntilGatewaySenderFlushedCoordinatorJUnitTest;
+import org.apache.geode.internal.cache.wan.parallel.WaitUntilParallelGatewaySenderFlushedCoordinator.WaitUntilBucketRegionQueueFlushedCallable;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+@Category(IntegrationTest.class)
+public class WaitUntilParallelGatewaySenderFlushedCoordinatorJUnitTest
+    extends WaitUntilGatewaySenderFlushedCoordinatorJUnitTest {
+
+  private PartitionedRegion region;
+
+  protected void createGatewaySender() {
+    super.createGatewaySender();
+    ConcurrentParallelGatewaySenderQueue queue =
+        (ConcurrentParallelGatewaySenderQueue) spy(getQueue());
+    doReturn(queue).when(this.sender).getQueue();
+    this.region = mock(PartitionedRegion.class);
+    doReturn(this.region).when(queue).getRegion();
+  }
+
+  protected AbstractGatewaySenderEventProcessor getEventProcessor() {
+    ConcurrentParallelGatewaySenderEventProcessor processor =
+        spy(new ConcurrentParallelGatewaySenderEventProcessor(this.sender));
+    return processor;
+  }
+
+  @Test
+  public void testWaitUntilParallelGatewaySenderFlushedSuccessfulNotInitiator() throws Throwable {
+    WaitUntilParallelGatewaySenderFlushedCoordinator coordinator =
+        new WaitUntilParallelGatewaySenderFlushedCoordinator(this.sender, 1000l,
+            TimeUnit.MILLISECONDS, false);
+    WaitUntilParallelGatewaySenderFlushedCoordinator coordinatorSpy = spy(coordinator);
+    doReturn(getSuccessfulCallables()).when(coordinatorSpy)
+        .buildWaitUntilBucketRegionQueueFlushedCallables(this.region);
+    boolean result = coordinatorSpy.waitUntilFlushed();
+    assertTrue(result);
+  }
+
+  @Test
+  public void testWaitUntilParallelGatewaySenderFlushedUnsuccessfulNotInitiator() throws Throwable {
+    WaitUntilParallelGatewaySenderFlushedCoordinator coordinator =
+        new WaitUntilParallelGatewaySenderFlushedCoordinator(this.sender, 1000l,
+            TimeUnit.MILLISECONDS, false);
+    WaitUntilParallelGatewaySenderFlushedCoordinator coordinatorSpy = spy(coordinator);
+    doReturn(getUnsuccessfulCallables()).when(coordinatorSpy)
+        .buildWaitUntilBucketRegionQueueFlushedCallables(this.region);
+    boolean result = coordinatorSpy.waitUntilFlushed();
+    assertFalse(result);
+  }
+
+  @Test
+  public void testWaitUntilParallelGatewaySenderFlushedSuccessfulInitiator() throws Throwable {
+    WaitUntilParallelGatewaySenderFlushedCoordinator coordinator =
+        new WaitUntilParallelGatewaySenderFlushedCoordinator(this.sender, 1000l,
+            TimeUnit.MILLISECONDS, true);
+    WaitUntilParallelGatewaySenderFlushedCoordinator coordinatorSpy = spy(coordinator);
+    doReturn(getSuccessfulCallables()).when(coordinatorSpy)
+        .buildWaitUntilBucketRegionQueueFlushedCallables(this.region);
+    doReturn(true).when(coordinatorSpy).waitUntilFlushedOnRemoteMembers(this.region);
+    boolean result = coordinatorSpy.waitUntilFlushed();
+    assertTrue(result);
+  }
+
+  @Test
+  public void testWaitUntilParallelGatewaySenderFlushedUnsuccessfulInitiator() throws Throwable {
+    WaitUntilParallelGatewaySenderFlushedCoordinator coordinator =
+        new WaitUntilParallelGatewaySenderFlushedCoordinator(this.sender, 1000l,
+            TimeUnit.MILLISECONDS, true);
+    WaitUntilParallelGatewaySenderFlushedCoordinator coordinatorSpy = spy(coordinator);
+    doReturn(getSuccessfulCallables()).when(coordinatorSpy)
+        .buildWaitUntilBucketRegionQueueFlushedCallables(this.region);
+    doReturn(false).when(coordinatorSpy).waitUntilFlushedOnRemoteMembers(this.region);
+    boolean result = coordinatorSpy.waitUntilFlushed();
+    assertFalse(result);
+  }
+
+  private List<WaitUntilBucketRegionQueueFlushedCallable> getSuccessfulCallables()
+      throws Exception {
+    List callables = new ArrayList();
+    WaitUntilBucketRegionQueueFlushedCallable callable =
+        mock(WaitUntilBucketRegionQueueFlushedCallable.class);
+    when(callable.call()).thenReturn(true);
+    callables.add(callable);
+    return callables;
+  }
+
+  private List<WaitUntilBucketRegionQueueFlushedCallable> getUnsuccessfulCallables()
+      throws Exception {
+    List callables = new ArrayList();
+    WaitUntilBucketRegionQueueFlushedCallable callable =
+        mock(WaitUntilBucketRegionQueueFlushedCallable.class);
+    when(callable.call()).thenReturn(false);
+    callables.add(callable);
+    return callables;
+  }
+}

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
----------------------------------------------------------------------
diff --git a/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt b/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
index 3f90ca4..0a791c4 100644
--- a/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
+++ b/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
@@ -2134,3 +2134,7 @@ org/apache/geode/redis/internal/DoubleWrapper,2
 fromData,9,2a2bb80004b50002b1
 toData,9,2ab400022bb80003b1
 
+org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator$WaitUntilGatewaySenderFlushedMessage,2
+fromData,47,2a2bb700322a2bb900330100b500032a2bb80034b500042a2bb900350100b500052a12362bb80037c00036b50006b1
+toData,42,2a2bb7002d2b2ab40003b9002e02002ab400042bb8002f2b2ab40005b9003003002ab400062bb80031b1
+

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneIndex.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneIndex.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneIndex.java
index f423c74..802b21a 100644
--- a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneIndex.java
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneIndex.java
@@ -16,6 +16,7 @@
 package org.apache.geode.cache.lucene;
 
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.lucene.analysis.Analyzer;
 
@@ -57,10 +58,12 @@ public interface LuceneIndex {
   /*
    * wait until the current entries in cache are indexed
    * 
-   * @param maxWaitInMilliseconds max wait time in millisecond
+   * @param timeout max wait time
    * 
-   * @return if entries are flushed within maxWait
+   * @param unit granularity of the timeout
+   *
+   * @return if entries are flushed within timeout
    */
-  public boolean waitUntilFlushed(int maxWaitInMillisecond);
+  public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException;
 
 }

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneIndexImpl.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneIndexImpl.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneIndexImpl.java
index 03d85c3..0b6685d 100644
--- a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneIndexImpl.java
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneIndexImpl.java
@@ -30,9 +30,7 @@ import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionAttributes;
 import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
 import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueFactoryImpl;
-import org.apache.geode.cache.lucene.internal.filesystem.ChunkKey;
-import org.apache.geode.cache.lucene.internal.filesystem.File;
-import org.apache.geode.cache.lucene.internal.filesystem.FileSystemStats;
+import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
 import org.apache.geode.cache.lucene.internal.repository.RepositoryManager;
 import org.apache.geode.cache.lucene.internal.xml.LuceneIndexCreation;
 import org.apache.geode.internal.cache.GemFireCacheImpl;
@@ -92,29 +90,15 @@ public abstract class LuceneIndexImpl implements InternalLuceneIndex {
   }
 
   @Override
-  public boolean waitUntilFlushed(int maxWaitInMillisecond) {
+  public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException {
     String aeqId = LuceneServiceImpl.getUniqueIndexName(indexName, regionPath);
-    AsyncEventQueue queue = (AsyncEventQueue) cache.getAsyncEventQueue(aeqId);
-    boolean flushed = false;
+    AsyncEventQueueImpl queue = (AsyncEventQueueImpl) cache.getAsyncEventQueue(aeqId);
     if (queue != null) {
-      long start = System.nanoTime();
-      while (System.nanoTime() - start < TimeUnit.MILLISECONDS.toNanos(maxWaitInMillisecond)) {
-        if (0 == queue.size()) {
-          flushed = true;
-          break;
-        } else {
-          try {
-            Thread.sleep(200);
-          } catch (InterruptedException e) {
-          }
-        }
-      }
+      return queue.waitUntilFlushed(timeout, unit);
     } else {
       throw new IllegalArgumentException(
           "The AEQ does not exist for the index " + indexName + " region " + regionPath);
     }
-
-    return flushed;
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/xml/LuceneIndexCreation.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/xml/LuceneIndexCreation.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/xml/LuceneIndexCreation.java
index 67b7392..030dddd 100644
--- a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/xml/LuceneIndexCreation.java
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/xml/LuceneIndexCreation.java
@@ -16,6 +16,7 @@
 package org.apache.geode.cache.lucene.internal.xml;
 
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper;
@@ -101,7 +102,7 @@ public class LuceneIndexCreation implements LuceneIndex, Extension<Region<?, ?>>
   }
 
   @Override
-  public boolean waitUntilFlushed(int maxWaitInMillisecond) {
+  public boolean waitUntilFlushed(long timeout, TimeUnit unit) {
     return true;
   }
 }

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneIndexMaintenanceIntegrationTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneIndexMaintenanceIntegrationTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneIndexMaintenanceIntegrationTest.java
index 1feba06..644ed78 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneIndexMaintenanceIntegrationTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneIndexMaintenanceIntegrationTest.java
@@ -52,7 +52,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
     region.put("object-4", new TestObject("hello world", "hello world"));
 
     LuceneIndex index = luceneService.getIndex(INDEX_NAME, REGION_NAME);
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
     LuceneQuery query = luceneService.createLuceneQueryFactory().create(INDEX_NAME, REGION_NAME,
         "description:\"hello world\"", DEFAULT_FIELD);
     PageableLuceneQueryResults<Integer, TestObject> results = query.findPages();
@@ -61,7 +61,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
     // begin transaction
     cache.getCacheTransactionManager().begin();
     region.put("object-1", new TestObject("title 1", "updated"));
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
     assertEquals(3, query.findPages().size());
   }
 
@@ -76,7 +76,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
     region.put("object-4", new TestObject("hello world", "hello world"));
 
     LuceneIndex index = luceneService.getIndex(INDEX_NAME, REGION_NAME);
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
     LuceneQuery query = luceneService.createLuceneQueryFactory().create(INDEX_NAME, REGION_NAME,
         "description:\"hello world\"", DEFAULT_FIELD);
     PageableLuceneQueryResults<Integer, TestObject> results = query.findPages();
@@ -85,7 +85,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
     cache.getCacheTransactionManager().begin();
     region.put("object-1", new TestObject("title 1", "updated"));
     cache.getCacheTransactionManager().commit();
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
 
     assertEquals(2, query.findPages().size());
   }
@@ -101,7 +101,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
     region.put("object-4", new TestObject("hello world", "hello world"));
 
     LuceneIndex index = luceneService.getIndex(INDEX_NAME, REGION_NAME);
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
     LuceneQuery query = luceneService.createLuceneQueryFactory().create(INDEX_NAME, REGION_NAME,
         "description:\"hello world\"", DEFAULT_FIELD);
     PageableLuceneQueryResults<Integer, TestObject> results = query.findPages();
@@ -110,7 +110,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
     cache.getCacheTransactionManager().begin();
     region.put("object-1", new TestObject("title 1", "updated"));
     cache.getCacheTransactionManager().rollback();
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
 
     assertEquals(3, query.findPages().size());
   }
@@ -127,7 +127,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
 
     LuceneIndexForPartitionedRegion index =
         (LuceneIndexForPartitionedRegion) luceneService.getIndex(INDEX_NAME, REGION_NAME);
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
 
     FileSystemStats fileSystemStats = index.getFileSystemStats();
     LuceneIndexStats indexStats = index.getIndexStats();
@@ -154,7 +154,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
 
     LuceneIndex index = luceneService.getIndex(INDEX_NAME, REGION_NAME);
     // Wait for events to be flushed from AEQ.
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
     // Execute query to fetch all the values for "description" field.
     LuceneQuery query = luceneService.createLuceneQueryFactory().create(INDEX_NAME, REGION_NAME,
         "description:\"hello world\"", DEFAULT_FIELD);
@@ -173,7 +173,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
     region.put(113, new TestObject("hello world", "hello world"));
     LuceneIndex index = luceneService.getIndex(INDEX_NAME, REGION_NAME);
     // Wait for events to be flushed from AEQ.
-    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME);
+    index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
     // Execute query to fetch all the values for "description" field.
     LuceneQuery query = luceneService.createLuceneQueryFactory().create(INDEX_NAME, REGION_NAME,
         "description:\"hello world\"", DEFAULT_FIELD);
@@ -182,7 +182,7 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
   }
 
   @Test
-  public void entriesFlushedToIndexAfterWaitForFlushCalled() {
+  public void entriesFlushedToIndexAfterWaitForFlushCalled() throws InterruptedException {
     luceneService.createIndex(INDEX_NAME, REGION_NAME, "title", "description");
 
     Region region = createRegion(REGION_NAME, RegionShortcut.PARTITION);
@@ -194,9 +194,9 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
     region.put("object-4", new TestObject("hello world", "hello world"));
 
     LuceneIndexImpl index = (LuceneIndexImpl) luceneService.getIndex(INDEX_NAME, REGION_NAME);
-    assertFalse(index.waitUntilFlushed(500));
+    assertFalse(index.waitUntilFlushed(500, TimeUnit.MILLISECONDS));
     LuceneTestUtilities.resumeSender(cache);
-    assertTrue(index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME));
+    assertTrue(index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS));
     assertEquals(4, index.getIndexStats().getCommits());
   }
 

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesBase.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesBase.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesBase.java
index 7c11d7c..a56fff7 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesBase.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesBase.java
@@ -22,6 +22,7 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import org.apache.geode.cache.Cache;
@@ -115,7 +116,7 @@ public abstract class LuceneQueriesBase extends LuceneDUnitTest {
       LuceneService service = LuceneServiceProvider.get(cache);
       LuceneIndexImpl index = (LuceneIndexImpl) service.getIndex(INDEX_NAME, REGION_NAME);
 
-      return index.waitUntilFlushed(ms);
+      return index.waitUntilFlushed(ms, TimeUnit.MILLISECONDS);
     });
   }
 

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesIntegrationTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesIntegrationTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesIntegrationTest.java
index 6cbb1fc..d11ea91 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesIntegrationTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesIntegrationTest.java
@@ -26,6 +26,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import org.apache.lucene.analysis.Analyzer;
@@ -89,7 +90,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     // <field1:one two three> <field2:one two three>
     // <field1:one@three> <field2:one@three>
 
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     // standard analyzer with double quote
     // this query string will be parsed as "one three"
@@ -139,7 +140,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     region.put("primitiveInt2", 223);
     region.put("primitiveInt3", 224);
 
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
     verifyQueryUsingCustomizedProvider(LuceneService.REGION_VALUE_FIELD, 123, 223, "primitiveInt1",
         "primitiveInt2");
   }
@@ -158,14 +159,14 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     region.put("primitiveInt2", 223);
     region.put("primitiveInt3", 224);
 
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     // Note: current QueryParser cannot query by range. It's a known issue in lucene
     verifyQuery(LuceneService.REGION_VALUE_FIELD + ":[123 TO 223]",
         LuceneService.REGION_VALUE_FIELD);
 
     region.put("primitiveDouble1", 123.0);
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     thrown.expectMessage("java.lang.IllegalArgumentException");
     verifyQueryUsingCustomizedProvider(LuceneService.REGION_VALUE_FIELD, 123, 223, "primitiveInt1",
@@ -202,7 +203,8 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     assertEquals(region.values(), new HashSet(query.findValues()));
   }
 
-  private LuceneQuery<Object, Object> addValuesAndCreateQuery(int pagesize) {
+  private LuceneQuery<Object, Object> addValuesAndCreateQuery(int pagesize)
+      throws InterruptedException {
     luceneService.createIndex(INDEX_NAME, REGION_NAME, "field1", "field2");
     region = cache.createRegionFactory(RegionShortcut.PARTITION).create(REGION_NAME);
     final LuceneIndex index = luceneService.getIndex(INDEX_NAME, REGION_NAME);
@@ -219,7 +221,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     region.put("F", new TestObject(value3, value3));
     region.put("G", new TestObject(value1, value2));
 
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
     return luceneService.createLuceneQueryFactory().setPageSize(pagesize).create(INDEX_NAME,
         REGION_NAME, "one", "field1");
   }
@@ -244,7 +246,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     region.put("C", new TestObject(value3, value3));
     region.put("D", new TestObject(value4, value4));
 
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     verifyQuery("field1:one AND field2:two_four", DEFAULT_FIELD, "A");
     verifyQuery("field1:one AND field2:two", DEFAULT_FIELD, "A");
@@ -263,7 +265,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     // Put two values with some of the same tokens
     String value1 = "one three";
     region.put("A", new TestObject(value1, null));
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     verifyQuery("field1:one", DEFAULT_FIELD, "A");
   }
@@ -282,7 +284,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     PdxInstance pdx1 = insertAJson(region, "jsondoc1");
     PdxInstance pdx2 = insertAJson(region, "jsondoc2");
     PdxInstance pdx10 = insertAJson(region, "jsondoc10");
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     HashMap expectedResults = new HashMap();
     expectedResults.put("jsondoc1", pdx1);
@@ -297,7 +299,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
     final LuceneIndex index = luceneService.getIndex(INDEX_NAME, REGION_NAME);
 
     region.put("A", "one three");
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     verifyQuery("one", LuceneService.REGION_VALUE_FIELD, "A");
   }

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPersistenceIntegrationTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPersistenceIntegrationTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPersistenceIntegrationTest.java
index c83763f..5ec09f1 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPersistenceIntegrationTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPersistenceIntegrationTest.java
@@ -18,6 +18,7 @@ import static org.apache.geode.cache.lucene.test.LuceneTestUtilities.*;
 import static org.junit.Assert.*;
 
 import java.io.File;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.geode.cache.EvictionAction;
 import org.apache.geode.cache.EvictionAlgorithm;
@@ -83,7 +84,7 @@ public class LuceneQueriesPersistenceIntegrationTest extends LuceneIntegrationTe
     value = new Type1("lucene world", 1, 2L, 3.0, 4.0f);
     userRegion.put("value3", value);
 
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     PartitionedRegion fileRegion = (PartitionedRegion) cache.getRegion(aeqId + ".files");
     assertNotNull(fileRegion);

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexImplJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexImplJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexImplJUnitTest.java
index 17fbcfc..aaa6dbd 100755
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexImplJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexImplJUnitTest.java
@@ -14,6 +14,7 @@
  */
 package org.apache.geode.cache.lucene.internal;
 
+import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -21,6 +22,7 @@ import org.junit.experimental.categories.Category;
 import org.junit.rules.ExpectedException;
 import static org.mockito.Mockito.*;
 
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.geode.cache.Cache;
@@ -49,13 +51,13 @@ public class LuceneIndexImplJUnitTest {
   @Test
   public void waitUnitFlushedWithMissingAEQThrowsIllegalArgument() throws Exception {
     thrown.expect(IllegalArgumentException.class);
-    index.waitUntilFlushed(MAX_WAIT);
+    index.waitUntilFlushed(MAX_WAIT, TimeUnit.MILLISECONDS);
   }
 
   @Test
   public void waitUnitFlushedWaitsForFlush() throws Exception {
     final String expectedIndexName = LuceneServiceImpl.getUniqueIndexName(INDEX, REGION);
-    final AsyncEventQueue queue = mock(AsyncEventQueue.class);
+    final AsyncEventQueueImpl queue = mock(AsyncEventQueueImpl.class);
     when(cache.getAsyncEventQueue(eq(expectedIndexName))).thenReturn(queue);
 
     AtomicInteger callCount = new AtomicInteger();
@@ -69,7 +71,7 @@ public class LuceneIndexImplJUnitTest {
         return 0;
       }
     });
-    index.waitUntilFlushed(MAX_WAIT);
+    index.waitUntilFlushed(MAX_WAIT, TimeUnit.MILLISECONDS);
     verify(cache).getAsyncEventQueue(eq(expectedIndexName));
   }
 

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexRecoveryHAIntegrationTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexRecoveryHAIntegrationTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexRecoveryHAIntegrationTest.java
index 10127ac..d194081 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexRecoveryHAIntegrationTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneIndexRecoveryHAIntegrationTest.java
@@ -35,6 +35,7 @@ import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
 import java.io.IOException;
+import java.util.concurrent.TimeUnit;
 
 import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
 import static org.junit.Assert.assertNotNull;
@@ -73,7 +74,8 @@ public class LuceneIndexRecoveryHAIntegrationTest {
    * construct index. This test simulates the same.
    */
   // @Test
-  public void recoverRepoInANewNode() throws BucketNotFoundException, IOException {
+  public void recoverRepoInANewNode()
+      throws BucketNotFoundException, IOException, InterruptedException {
     LuceneServiceImpl service = (LuceneServiceImpl) LuceneServiceProvider.get(cache);
     service.createIndex("index1", "/userRegion", indexedFields);
     PartitionAttributes<String, String> attrs =
@@ -87,7 +89,7 @@ public class LuceneIndexRecoveryHAIntegrationTest {
         (LuceneIndexForPartitionedRegion) service.getIndex("index1", "/userRegion");
     // put an entry to create the bucket
     userRegion.put("rebalance", "test");
-    index.waitUntilFlushed(30000);
+    index.waitUntilFlushed(30000, TimeUnit.MILLISECONDS);
 
     RepositoryManager manager = new PartitionedRepositoryManager((LuceneIndexImpl) index, mapper);
     IndexRepository repo = manager.getRepository(userRegion, 0, null);
@@ -112,9 +114,10 @@ public class LuceneIndexRecoveryHAIntegrationTest {
 
 
 
-  private void verifyIndexFinishFlushing(String indexName, String regionName) {
+  private void verifyIndexFinishFlushing(String indexName, String regionName)
+      throws InterruptedException {
     LuceneIndex index = LuceneServiceProvider.get(cache).getIndex(indexName, regionName);
-    boolean flushed = index.waitUntilFlushed(60000);
+    boolean flushed = index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
     assertTrue(flushed);
   }
 }

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/cli/LuceneIndexCommandsDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/cli/LuceneIndexCommandsDUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/cli/LuceneIndexCommandsDUnitTest.java
index f0f1579..1c8bc88 100755
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/cli/LuceneIndexCommandsDUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/cli/LuceneIndexCommandsDUnitTest.java
@@ -610,7 +610,8 @@ public class LuceneIndexCommandsDUnitTest extends CliCommandTestBase {
       LuceneService luceneService = LuceneServiceProvider.get(getCache());
       Region region = getCache().getRegion(REGION_NAME);
       region.putAll(entries);
-      luceneService.getIndex(INDEX_NAME, REGION_NAME).waitUntilFlushed(60000);
+      luceneService.getIndex(INDEX_NAME, REGION_NAME).waitUntilFlushed(60000,
+          TimeUnit.MILLISECONDS);
       LuceneIndexImpl index = (LuceneIndexImpl) luceneService.getIndex(INDEX_NAME, REGION_NAME);
       Awaitility.await().atMost(65, TimeUnit.SECONDS)
           .until(() -> assertEquals(countOfDocuments, index.getIndexStats().getDocuments()));

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/directory/DumpDirectoryFilesIntegrationTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/directory/DumpDirectoryFilesIntegrationTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/directory/DumpDirectoryFilesIntegrationTest.java
index 697166e..5c47e13 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/directory/DumpDirectoryFilesIntegrationTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/directory/DumpDirectoryFilesIntegrationTest.java
@@ -18,6 +18,7 @@ import static org.apache.geode.cache.lucene.test.LuceneTestUtilities.*;
 import static org.junit.Assert.*;
 
 import java.io.File;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionShortcut;
@@ -55,7 +56,7 @@ public class DumpDirectoryFilesIntegrationTest extends LuceneIntegrationTest {
     InternalLuceneIndex index =
         (InternalLuceneIndex) luceneService.getIndex(INDEX_NAME, REGION_NAME);
 
-    index.waitUntilFlushed(60000);
+    index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
 
     index.dumpFiles(diskDirRule.get().getAbsolutePath());
 

http://git-wip-us.apache.org/repos/asf/geode/blob/44cd72d8/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java
index d8816a3..21c0bbc 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java
@@ -20,6 +20,7 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 import org.apache.geode.cache.Cache;
@@ -86,10 +87,11 @@ public class LuceneTestUtilities {
     LuceneServiceProvider.get(cache).createIndex(INDEX_NAME, REGION_NAME, fieldNames);
   }
 
-  public static void verifyIndexFinishFlushing(Cache cache, String indexName, String regionName) {
+  public static void verifyIndexFinishFlushing(Cache cache, String indexName, String regionName)
+      throws InterruptedException {
     LuceneService luceneService = LuceneServiceProvider.get(cache);
     LuceneIndex index = luceneService.getIndex(indexName, regionName);
-    boolean flushed = index.waitUntilFlushed(60000);
+    boolean flushed = index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
     assertTrue(flushed);
   }
 


Mime
View raw message