You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by zh...@apache.org on 2017/02/09 01:03:39 UTC
geode git commit: GEODE-2400: use a function to implement
waitUntilFlushed, then accessor and client can have this feature.
Repository: geode
Updated Branches:
refs/heads/develop 0d0c7c424 -> a6c28a65e
GEODE-2400: use a function to implement waitUntilFlushed, then accessor and client
can have this feature.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/a6c28a65
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/a6c28a65
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/a6c28a65
Branch: refs/heads/develop
Commit: a6c28a65e96fcb911b2ac8fe6c15bc968f021851
Parents: 0d0c7c4
Author: zhouxh <gz...@pivotal.io>
Authored: Tue Feb 7 09:59:44 2017 -0800
Committer: zhouxh <gz...@pivotal.io>
Committed: Wed Feb 8 10:20:32 2017 -0800
----------------------------------------------------------------------
.../org/apache/geode/internal/DSFIDFactory.java | 2 -
.../geode/internal/DataSerializableFixedID.java | 2 +-
.../geode/internal/cache/BucketRegionQueue.java | 23 ++-
.../cache/wan/AbstractGatewaySender.java | 39 ++---
...ParallelGatewaySenderFlushedCoordinator.java | 154 +------------------
...atewaySenderFlushedCoordinatorJUnitTest.java | 2 -
.../apache/geode/cache/lucene/LuceneIndex.java | 12 --
.../geode/cache/lucene/LuceneService.java | 16 ++
.../cache/lucene/internal/LuceneIndexImpl.java | 18 +--
.../lucene/internal/LuceneServiceImpl.java | 30 ++++
.../distributed/WaitUntilFlushedFunction.java | 101 ++++++++++++
.../WaitUntilFlushedFunctionContext.java | 93 +++++++++++
.../internal/xml/LuceneIndexCreation.java | 4 -
.../LuceneIndexMaintenanceIntegrationTest.java | 33 ++--
.../geode/cache/lucene/LuceneQueriesBase.java | 26 +++-
.../lucene/LuceneQueriesIntegrationTest.java | 18 +--
.../geode/cache/lucene/LuceneQueriesPRBase.java | 7 +-
.../LuceneQueriesPeerPRRedundancyDUnitTest.java | 7 +-
...LuceneQueriesPersistenceIntegrationTest.java | 2 +-
.../internal/LuceneIndexImplJUnitTest.java | 34 ----
.../LuceneIndexRecoveryHAIntegrationTest.java | 7 +-
.../internal/LuceneServiceImplJUnitTest.java | 30 +++-
.../cli/LuceneIndexCommandsDUnitTest.java | 3 +-
.../DumpDirectoryFilesIntegrationTest.java | 2 +-
.../WaitUntilFlushedFunctionJUnitTest.java | 106 +++++++++++++
.../cache/lucene/test/IndexRepositorySpy.java | 3 +
.../cache/lucene/test/LuceneTestUtilities.java | 3 +-
27 files changed, 494 insertions(+), 283 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 bb29239..5c18639 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
@@ -922,8 +922,6 @@ 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/a6c28a65/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 84eb8e9..74e40e9 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,7 +810,7 @@ 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;
+ public static final short WAIT_UNTIL_FLUSHED_FUNCTION_CONTEXT = 2177;
// NOTE, codes > 65535 will take 4 bytes to serialize
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 4830912..3259752 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
@@ -42,6 +42,7 @@ import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.execute.BucketMovedException;
import org.apache.geode.internal.cache.persistence.query.mock.ByteComparator;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
@@ -464,9 +465,10 @@ public class BucketRegionQueue extends AbstractBucketRegionQueue {
}
public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException {
+ long then = System.currentTimeMillis();
if (logger.isDebugEnabled()) {
- logger.debug("BucketRegionQueue: waitUntilFlushed bucket=" + getId() + "; time="
- + System.currentTimeMillis() + "; timeout=" + timeout + "; unit=" + unit);
+ logger.debug("BucketRegionQueue: waitUntilFlushed bucket=" + getId() + "; timeout=" + timeout
+ + "; unit=" + unit);
}
boolean result = false;
// Wait until latestAcknowledgedKey > latestQueuedKey or the queue is empty
@@ -475,17 +477,24 @@ public class BucketRegionQueue extends AbstractBucketRegionQueue {
long nanosRemaining = unit.toNanos(timeout);
long endTime = System.nanoTime() + nanosRemaining;
while (nanosRemaining > 0) {
- if (latestAcknowledgedKey.get() > latestQueuedKeyToCheck || isEmpty()) {
- result = true;
- break;
+ try {
+ if (latestAcknowledgedKey.get() > latestQueuedKeyToCheck || isEmpty()) {
+ result = true;
+ break;
+ }
+ } catch (RegionDestroyedException e) {
+ if (this.isBucketDestroyed()) {
+ getCancelCriterion().checkCancelInProgress(e);
+ throw new BucketMovedException(this.getFullPath());
+ }
}
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);
+ logger.debug("BucketRegionQueue: waitUntilFlushed completed bucket=" + getId() + "; duration="
+ + (System.currentTimeMillis() - then) + "; result=" + result);
}
return result;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 1f8704c..a604dbf 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
@@ -1242,31 +1242,24 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
}
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);
- }
+ try {
+ WaitUntilParallelGatewaySenderFlushedCoordinator coordinator =
+ new WaitUntilParallelGatewaySenderFlushedCoordinator(this, timeout, unit, true);
+ result = coordinator.waitUntilFlushed();
+ } catch (BucketMovedException | CancelException | RegionDestroyedException e) {
+ logger.warn(
+ LocalizedStrings.AbstractGatewaySender_CAUGHT_EXCEPTION_ATTEMPTING_WAIT_UNTIL_FLUSHED_RETRYING
+ .toLocalizedString(),
+ e);
+ throw e;
+ } catch (Throwable t) {
+ logger.warn(
+ LocalizedStrings.AbstractGatewaySender_CAUGHT_EXCEPTION_ATTEMPTING_WAIT_UNTIL_FLUSHED_RETURNING
+ .toLocalizedString(),
+ t);
+ throw new InternalGemFireError(t);
}
return result;
} else {
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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
index a4c03a9..528d3bb 100644
--- 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
@@ -14,8 +14,6 @@
*/
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;
@@ -24,14 +22,10 @@ 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
@@ -43,11 +37,14 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
}
public boolean waitUntilFlushed() throws Throwable {
- boolean remoteResult = true, localResult = true;
+ boolean localResult = true;
Throwable exceptionToThrow = null;
ConcurrentParallelGatewaySenderQueue prq =
(ConcurrentParallelGatewaySenderQueue) this.sender.getQueue();
PartitionedRegion pr = (PartitionedRegion) prq.getRegion();
+ if (pr == null) {
+ sender.getCancelCriterion().checkCancelInProgress(null);
+ }
// Create callables for local buckets
List<WaitUntilBucketRegionQueueFlushedCallable> callables =
@@ -64,20 +61,6 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
+ 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;
@@ -97,9 +80,9 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
if (exceptionToThrow == null) {
if (logger.isDebugEnabled()) {
logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Returning full result="
- + (remoteResult && localResult));
+ + (localResult));
}
- return remoteResult && localResult;
+ return localResult;
} else {
throw exceptionToThrow;
}
@@ -117,43 +100,6 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
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;
@@ -232,92 +178,4 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
}
}
- 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/a6c28a65/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
index c8b8ba1..8279130 100644
--- 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
@@ -86,7 +86,6 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinatorJUnitTest
WaitUntilParallelGatewaySenderFlushedCoordinator coordinatorSpy = spy(coordinator);
doReturn(getSuccessfulCallables()).when(coordinatorSpy)
.buildWaitUntilBucketRegionQueueFlushedCallables(this.region);
- doReturn(true).when(coordinatorSpy).waitUntilFlushedOnRemoteMembers(this.region);
boolean result = coordinatorSpy.waitUntilFlushed();
assertTrue(result);
}
@@ -99,7 +98,6 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinatorJUnitTest
WaitUntilParallelGatewaySenderFlushedCoordinator coordinatorSpy = spy(coordinator);
doReturn(getSuccessfulCallables()).when(coordinatorSpy)
.buildWaitUntilBucketRegionQueueFlushedCallables(this.region);
- doReturn(false).when(coordinatorSpy).waitUntilFlushedOnRemoteMembers(this.region);
boolean result = coordinatorSpy.waitUntilFlushed();
assertFalse(result);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 802b21a..1069f74 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,7 +16,6 @@
package org.apache.geode.cache.lucene;
import java.util.Map;
-import java.util.concurrent.TimeUnit;
import org.apache.lucene.analysis.Analyzer;
@@ -55,15 +54,4 @@ public interface LuceneIndex {
*/
public Map<String, Analyzer> getFieldAnalyzers();
- /*
- * wait until the current entries in cache are indexed
- *
- * @param timeout max wait time
- *
- * @param unit granularity of the timeout
- *
- * @return if entries are flushed within timeout
- */
- public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException;
-
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneService.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneService.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneService.java
index 0ff5f0e..6fffe09 100644
--- a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneService.java
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/LuceneService.java
@@ -16,6 +16,7 @@ package org.apache.geode.cache.lucene;
import java.util.Collection;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import org.apache.lucene.analysis.Analyzer;
@@ -130,4 +131,19 @@ public interface LuceneService {
*/
public LuceneQueryFactory createLuceneQueryFactory();
+ /*
+ * wait until the current entries in cache are indexed
+ *
+ * @param indexName index name
+ *
+ * @param regionPath region name
+ *
+ * @param timeout max wait time
+ *
+ * @param unit granularity of the timeout
+ *
+ * @return if entries are flushed within timeout
+ */
+ public boolean waitUntilFlushed(String indexName, String regionPath, long timeout, TimeUnit unit)
+ throws InterruptedException;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 8d24a83..42ccc84 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
@@ -16,6 +16,7 @@
package org.apache.geode.cache.lucene.internal;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -31,6 +32,11 @@ 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.asyncqueue.internal.AsyncEventQueueImpl;
+import org.apache.geode.cache.execute.Execution;
+import org.apache.geode.cache.execute.FunctionService;
+import org.apache.geode.cache.execute.ResultCollector;
+import org.apache.geode.cache.lucene.internal.distributed.WaitUntilFlushedFunction;
+import org.apache.geode.cache.lucene.internal.distributed.WaitUntilFlushedFunctionContext;
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;
@@ -90,18 +96,6 @@ public abstract class LuceneIndexImpl implements InternalLuceneIndex {
}
@Override
- public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException {
- String aeqId = LuceneServiceImpl.getUniqueIndexName(indexName, regionPath);
- AsyncEventQueueImpl queue = (AsyncEventQueueImpl) cache.getAsyncEventQueue(aeqId);
- if (queue != null) {
- return queue.waitUntilFlushed(timeout, unit);
- } else {
- throw new IllegalArgumentException(
- "The AEQ does not exist for the index " + indexName + " region " + regionPath);
- }
- }
-
- @Override
public String[] getFieldNames() {
return searchableFieldNames;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java
index 70061eb..f1b1861 100644
--- a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/LuceneServiceImpl.java
@@ -16,6 +16,7 @@
package org.apache.geode.cache.lucene.internal;
import java.util.*;
+import java.util.concurrent.TimeUnit;
import org.apache.geode.cache.lucene.internal.management.LuceneServiceMBean;
import org.apache.geode.cache.lucene.internal.management.ManagementIndexListener;
@@ -31,7 +32,9 @@ import org.apache.geode.cache.EvictionAlgorithm;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
+import org.apache.geode.cache.execute.Execution;
import org.apache.geode.cache.execute.FunctionService;
+import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.lucene.LuceneIndex;
import org.apache.geode.cache.lucene.LuceneQueryFactory;
import org.apache.geode.cache.lucene.internal.directory.DumpDirectoryFiles;
@@ -41,6 +44,8 @@ import org.apache.geode.cache.lucene.internal.distributed.LuceneFunctionContext;
import org.apache.geode.cache.lucene.internal.distributed.TopEntries;
import org.apache.geode.cache.lucene.internal.distributed.TopEntriesCollector;
import org.apache.geode.cache.lucene.internal.distributed.TopEntriesCollectorManager;
+import org.apache.geode.cache.lucene.internal.distributed.WaitUntilFlushedFunction;
+import org.apache.geode.cache.lucene.internal.distributed.WaitUntilFlushedFunctionContext;
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.xml.LuceneServiceXmlGenerator;
@@ -85,6 +90,7 @@ public class LuceneServiceImpl implements InternalLuceneService {
this.cache = gfc;
FunctionService.registerFunction(new LuceneFunction());
+ FunctionService.registerFunction(new WaitUntilFlushedFunction());
FunctionService.registerFunction(new DumpDirectoryFiles());
registerDataSerializables();
}
@@ -312,6 +318,9 @@ public class LuceneServiceImpl implements InternalLuceneService {
DSFIDFactory.registerDSFID(DataSerializableFixedID.LUCENE_TOP_ENTRIES_COLLECTOR,
TopEntriesCollector.class);
+
+ DSFIDFactory.registerDSFID(DataSerializableFixedID.WAIT_UNTIL_FLUSHED_FUNCTION_CONTEXT,
+ WaitUntilFlushedFunctionContext.class);
}
public Collection<LuceneIndexCreationProfile> getAllDefinedIndexes() {
@@ -321,4 +330,25 @@ public class LuceneServiceImpl implements InternalLuceneService {
public LuceneIndexCreationProfile getDefinedIndex(String indexName, String regionPath) {
return definedIndexMap.get(getUniqueIndexName(indexName, regionPath));
}
+
+ public boolean waitUntilFlushed(String indexName, String regionPath, long timeout, TimeUnit unit)
+ throws InterruptedException {
+ Region dataRegion = this.cache.getRegion(regionPath);
+ if (dataRegion == null) {
+ logger.info("Data region " + regionPath + " not found");
+ return false;
+ }
+
+ WaitUntilFlushedFunctionContext context =
+ new WaitUntilFlushedFunctionContext(indexName, timeout, unit);
+ Execution execution = FunctionService.onRegion(dataRegion);
+ ResultCollector rs = execution.withArgs(context).execute(WaitUntilFlushedFunction.ID);
+ List<Boolean> results = (List<Boolean>) rs.getResult();
+ for (Boolean oneResult : results) {
+ if (oneResult == false) {
+ return false;
+ }
+ }
+ return true;
+ }
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunction.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunction.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunction.java
new file mode 100644
index 0000000..90f1b9f
--- /dev/null
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunction.java
@@ -0,0 +1,101 @@
+/*
+ * 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.cache.lucene.internal.distributed;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.geode.cache.lucene.internal.LuceneIndexImpl;
+import org.apache.geode.cache.lucene.internal.LuceneIndexStats;
+import org.apache.geode.cache.lucene.internal.LuceneServiceImpl;
+import org.apache.logging.log4j.Logger;
+import org.apache.lucene.search.Query;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
+import org.apache.geode.cache.execute.FunctionAdapter;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.execute.FunctionException;
+import org.apache.geode.cache.execute.RegionFunctionContext;
+import org.apache.geode.cache.execute.ResultSender;
+import org.apache.geode.cache.lucene.LuceneQueryException;
+import org.apache.geode.cache.lucene.LuceneQueryProvider;
+import org.apache.geode.cache.lucene.LuceneService;
+import org.apache.geode.cache.lucene.LuceneServiceProvider;
+import org.apache.geode.cache.lucene.internal.repository.IndexRepository;
+import org.apache.geode.cache.lucene.internal.repository.IndexResultCollector;
+import org.apache.geode.cache.lucene.internal.repository.RepositoryManager;
+import org.apache.geode.internal.InternalEntity;
+import org.apache.geode.internal.cache.BucketNotFoundException;
+import org.apache.geode.internal.logging.LogService;
+
+/**
+ * {@link WaitUntilFlushedFunction} will check all the members with index to wait until the events
+ * in current AEQs are flushed into index. This function enables an accessor and client to call to
+ * make sure the current events are processed.
+ */
+public class WaitUntilFlushedFunction extends FunctionAdapter implements InternalEntity {
+ private static final long serialVersionUID = 1L;
+ public static final String ID = WaitUntilFlushedFunction.class.getName();
+
+ private static final Logger logger = LogService.getLogger();
+
+ @Override
+ public void execute(FunctionContext context) {
+ RegionFunctionContext ctx = (RegionFunctionContext) context;
+ ResultSender<Boolean> resultSender = ctx.getResultSender();
+
+ Region region = ctx.getDataSet();
+
+ WaitUntilFlushedFunctionContext arg = (WaitUntilFlushedFunctionContext) ctx.getArguments();
+ String indexName = arg.getIndexName();
+ if (indexName == null) {
+ throw new IllegalArgumentException("Missing index name");
+ }
+ long timeout = arg.getTimeout();
+ TimeUnit unit = arg.getTimeunit();
+
+ LuceneService service = LuceneServiceProvider.get(region.getCache());
+ LuceneIndexImpl index = (LuceneIndexImpl) service.getIndex(indexName, region.getFullPath());
+
+ boolean result = false;
+ String aeqId = LuceneServiceImpl.getUniqueIndexName(indexName, region.getFullPath());
+ AsyncEventQueueImpl queue = (AsyncEventQueueImpl) region.getCache().getAsyncEventQueue(aeqId);
+ if (queue != null) {
+ try {
+ result = queue.waitUntilFlushed(timeout, unit);
+ } catch (InterruptedException e) {
+ }
+
+ } else {
+ throw new IllegalArgumentException(
+ "The AEQ does not exist for the index " + indexName + " region " + region.getFullPath());
+ }
+ resultSender.lastResult(result);
+ }
+
+ @Override
+ public String getId() {
+ return ID;
+ }
+
+ @Override
+ public boolean optimizeForWrite() {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunctionContext.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunctionContext.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunctionContext.java
new file mode 100644
index 0000000..88d0e63
--- /dev/null
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunctionContext.java
@@ -0,0 +1,93 @@
+/*
+ * 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.cache.lucene.internal.distributed;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.cache.lucene.LuceneQueryFactory;
+import org.apache.geode.cache.lucene.LuceneQueryProvider;
+import org.apache.geode.cache.lucene.internal.repository.IndexRepository;
+import org.apache.geode.cache.lucene.internal.repository.IndexResultCollector;
+import org.apache.geode.internal.DataSerializableFixedID;
+import org.apache.geode.internal.Version;
+
+/**
+ * Contains function arguments for WaitUntilFlushed
+ */
+public class WaitUntilFlushedFunctionContext implements DataSerializableFixedID {
+ private String indexName;
+ private long timeout;
+ private TimeUnit unit;
+
+ public WaitUntilFlushedFunctionContext() {
+ this(null, 0, TimeUnit.MILLISECONDS);
+ }
+
+ public WaitUntilFlushedFunctionContext(String indexName, long timeout, TimeUnit unit) {
+ this.indexName = indexName;
+ this.timeout = timeout;
+ this.unit = unit;
+ }
+
+ /**
+ * Get the name of the index
+ */
+ public String getIndexName() {
+ return indexName;
+ }
+
+ /**
+ * Get the timeout
+ */
+ public long getTimeout() {
+ return this.timeout;
+ }
+
+ /*
+ * Get the timeout's unit
+ */
+ public TimeUnit getTimeunit() {
+ return unit;
+ }
+
+ @Override
+ public int getDSFID() {
+ return WAIT_UNTIL_FLUSHED_FUNCTION_CONTEXT;
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+ DataSerializer.writeString(indexName, out);
+ out.writeLong(timeout);
+ DataSerializer.writeEnum(unit, out);
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException {
+ this.indexName = DataSerializer.readString(in);
+ timeout = in.readLong();
+ unit = DataSerializer.readEnum(TimeUnit.class, in);
+ }
+
+ @Override
+ public Version[] getSerializationVersions() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 030dddd..b878e47 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
@@ -101,8 +101,4 @@ public class LuceneIndexCreation implements LuceneIndex, Extension<Region<?, ?>>
this.fieldNames.addAll(Arrays.asList(fieldNames));
}
- @Override
- public boolean waitUntilFlushed(long timeout, TimeUnit unit) {
- return true;
- }
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 e6c4503..5ac01b8 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,8 @@ 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 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 +62,8 @@ 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, WAIT_FOR_FLUSH_TIME,
+ TimeUnit.MILLISECONDS);
assertEquals(3, query.findPages().size());
}
@@ -76,7 +78,8 @@ 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 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 +88,8 @@ 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, WAIT_FOR_FLUSH_TIME,
+ TimeUnit.MILLISECONDS);
assertEquals(2, query.findPages().size());
}
@@ -101,7 +105,8 @@ 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 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 +115,8 @@ 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, WAIT_FOR_FLUSH_TIME,
+ TimeUnit.MILLISECONDS);
assertEquals(3, query.findPages().size());
}
@@ -127,7 +133,8 @@ public class LuceneIndexMaintenanceIntegrationTest extends LuceneIntegrationTest
LuceneIndexForPartitionedRegion index =
(LuceneIndexForPartitionedRegion) luceneService.getIndex(INDEX_NAME, REGION_NAME);
- index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, WAIT_FOR_FLUSH_TIME,
+ TimeUnit.MILLISECONDS);
FileSystemStats fileSystemStats = index.getFileSystemStats();
LuceneIndexStats indexStats = index.getIndexStats();
@@ -154,7 +161,8 @@ 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 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 +181,8 @@ 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 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);
@@ -194,9 +203,11 @@ 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, TimeUnit.MILLISECONDS));
+ assertFalse(
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 500, TimeUnit.MILLISECONDS));
LuceneTestUtilities.resumeSender(cache);
- assertTrue(index.waitUntilFlushed(WAIT_FOR_FLUSH_TIME, TimeUnit.MILLISECONDS));
+ assertTrue(luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, WAIT_FOR_FLUSH_TIME,
+ TimeUnit.MILLISECONDS));
assertEquals(4, index.getIndexStats().getCommits());
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 a56fff7..e5210dc 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
@@ -28,6 +28,7 @@ import java.util.stream.Collectors;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.lucene.internal.LuceneIndexImpl;
+import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.SerializableRunnableIF;
import org.apache.geode.test.dunit.VM;
@@ -65,6 +66,7 @@ public abstract class LuceneQueriesBase extends LuceneDUnitTest {
accessor.invoke(() -> initAccessor(createIndex));
putDataInRegion(accessor);
+ assertTrue(waitForFlushBeforeExecuteTextSearch(accessor, 60000));
assertTrue(waitForFlushBeforeExecuteTextSearch(dataStore1, 60000));
executeTextSearch(accessor);
}
@@ -79,6 +81,7 @@ public abstract class LuceneQueriesBase extends LuceneDUnitTest {
dataStore2.invoke(() -> initDataStore(createIndex));
accessor.invoke(() -> initAccessor(createIndex));
putDataInRegion(accessor);
+ assertTrue(waitForFlushBeforeExecuteTextSearch(accessor, 60000));
assertTrue(waitForFlushBeforeExecuteTextSearch(dataStore1, 60000));
executeTextSearch(accessor, "world", "text", 3);
executeTextSearch(accessor, "world", "noEntriesMapped", 0);
@@ -94,6 +97,7 @@ public abstract class LuceneQueriesBase extends LuceneDUnitTest {
dataStore2.invoke(() -> initDataStore(createIndex));
accessor.invoke(() -> initAccessor(createIndex));
putDataInRegion(accessor);
+ assertTrue(waitForFlushBeforeExecuteTextSearch(accessor, 60000));
assertTrue(waitForFlushBeforeExecuteTextSearch(dataStore1, 60000));
// Execute a query with a custom lucene query object
@@ -109,6 +113,26 @@ public abstract class LuceneQueriesBase extends LuceneDUnitTest {
});
}
+ @Test
+ public void verifyWaitForFlushedFunctionOnAccessor() throws InterruptedException {
+ SerializableRunnableIF createIndex = () -> {
+ LuceneService luceneService = LuceneServiceProvider.get(getCache());
+ luceneService.createIndex(INDEX_NAME, REGION_NAME, "text");
+ };
+ dataStore1.invoke(() -> initDataStore(createIndex));
+ dataStore2.invoke(() -> initDataStore(createIndex));
+ accessor.invoke(() -> initAccessor(createIndex));
+ dataStore1.invoke(() -> LuceneTestUtilities.pauseSender(getCache()));
+ dataStore2.invoke(() -> LuceneTestUtilities.pauseSender(getCache()));
+ putDataInRegion(accessor);
+ assertFalse(waitForFlushBeforeExecuteTextSearch(accessor, 60000));
+ dataStore1.invoke(() -> LuceneTestUtilities.resumeSender(getCache()));
+ dataStore2.invoke(() -> LuceneTestUtilities.resumeSender(getCache()));
+ assertTrue(waitForFlushBeforeExecuteTextSearch(accessor, 60000));
+ executeTextSearch(accessor, "world", "text", 3);
+ executeTextSearch(accessor, "world", "noEntriesMapped", 0);
+ }
+
protected boolean waitForFlushBeforeExecuteTextSearch(VM vm, int ms) {
return vm.invoke(() -> {
Cache cache = getCache();
@@ -116,7 +140,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, TimeUnit.MILLISECONDS);
+ return service.waitUntilFlushed(INDEX_NAME, REGION_NAME, ms, TimeUnit.MILLISECONDS);
});
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 d11ea91..01bdca4 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
@@ -90,7 +90,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
// <field1:one two three> <field2:one two three>
// <fi...@three> <fi...@three>
- index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
// standard analyzer with double quote
// this query string will be parsed as "one three"
@@ -140,7 +140,7 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
region.put("primitiveInt2", 223);
region.put("primitiveInt3", 224);
- index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
verifyQueryUsingCustomizedProvider(LuceneService.REGION_VALUE_FIELD, 123, 223, "primitiveInt1",
"primitiveInt2");
}
@@ -159,14 +159,14 @@ public class LuceneQueriesIntegrationTest extends LuceneIntegrationTest {
region.put("primitiveInt2", 223);
region.put("primitiveInt3", 224);
- index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
thrown.expectMessage("java.lang.IllegalArgumentException");
verifyQueryUsingCustomizedProvider(LuceneService.REGION_VALUE_FIELD, 123, 223, "primitiveInt1",
@@ -221,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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
return luceneService.createLuceneQueryFactory().setPageSize(pagesize).create(INDEX_NAME,
REGION_NAME, "one", "field1");
}
@@ -246,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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
verifyQuery("field1:one AND field2:two_four", DEFAULT_FIELD, "A");
verifyQuery("field1:one AND field2:two", DEFAULT_FIELD, "A");
@@ -265,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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
verifyQuery("field1:one", DEFAULT_FIELD, "A");
}
@@ -284,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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
HashMap expectedResults = new HashMap();
expectedResults.put("jsondoc1", pdx1);
@@ -299,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, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
verifyQuery("one", LuceneService.REGION_VALUE_FIELD, "A");
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPRBase.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPRBase.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPRBase.java
index c877f93..931edc5 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPRBase.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPRBase.java
@@ -20,6 +20,7 @@ import static org.apache.geode.cache.lucene.test.LuceneTestUtilities.*;
import static org.junit.Assert.*;
import static org.mockito.Matchers.any;
+import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import org.junit.After;
@@ -29,8 +30,10 @@ import org.apache.geode.cache.Cache;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.control.RebalanceOperation;
import org.apache.geode.cache.control.RebalanceResults;
+import org.apache.geode.cache.lucene.internal.LuceneIndexImpl;
import org.apache.geode.cache.lucene.test.IndexRepositorySpy;
import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
import org.apache.geode.cache.partition.PartitionRegionHelper;
@@ -93,7 +96,7 @@ public abstract class LuceneQueriesPRBase extends LuceneQueriesBase {
putEntryInEachBucket();
dataStore2.invoke(() -> initDataStore(createIndex));
- assertTrue(waitForFlushBeforeExecuteTextSearch(dataStore2, 60000)); // accessor?
+ assertTrue(waitForFlushBeforeExecuteTextSearch(accessor, 60000));
assertTrue(waitForFlushBeforeExecuteTextSearch(dataStore1, 60000));
rebalanceRegion(dataStore2);
@@ -118,7 +121,7 @@ public abstract class LuceneQueriesPRBase extends LuceneQueriesBase {
rebalanceRegion(dataStore2);
dataStore1.invoke(() -> LuceneTestUtilities.resumeSender(getCache()));
- assertTrue(waitForFlushBeforeExecuteTextSearch(dataStore2, 60000)); // accessor?
+ assertTrue(waitForFlushBeforeExecuteTextSearch(accessor, 60000));
assertTrue(waitForFlushBeforeExecuteTextSearch(dataStore1, 60000));
executeTextSearch(accessor, "world", "text", NUM_BUCKETS);
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPeerPRRedundancyDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPeerPRRedundancyDUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPeerPRRedundancyDUnitTest.java
index a9772eb..496721a 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPeerPRRedundancyDUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/LuceneQueriesPeerPRRedundancyDUnitTest.java
@@ -20,6 +20,7 @@ import static org.junit.Assert.*;
import java.util.concurrent.TimeUnit;
+import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.lucene.test.IndexRegionSpy;
@@ -33,7 +34,6 @@ import org.apache.geode.internal.cache.partitioned.BecomePrimaryBucketMessage.Be
import org.apache.geode.test.dunit.SerializableRunnableIF;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.junit.categories.DistributedTest;
-import org.apache.geode.test.junit.categories.FlakyTest;
import org.awaitility.Awaitility;
@@ -67,7 +67,6 @@ public class LuceneQueriesPeerPRRedundancyDUnitTest extends LuceneQueriesPRBase
putEntriesAndValidateResultsWithRedundancy();
}
- @Category(FlakyTest.class) // GEODE-1956
@Test
public void returnCorrectResultsWhenCloseCacheHappensOnIndexUpdate() throws InterruptedException {
dataStore1.invoke(() -> {
@@ -76,6 +75,9 @@ public class LuceneQueriesPeerPRRedundancyDUnitTest extends LuceneQueriesPRBase
spy.beforeWriteIndexRepository(doAfterN(key -> getCache().close(), 2));
});
+ final String expectedExceptions = CacheClosedException.class.getName();
+ dataStore1.invoke(addExceptionTag1(expectedExceptions));
+
putEntriesAndValidateResultsWithRedundancy();
// Wait until the cache is closed in datastore1
@@ -83,7 +85,6 @@ public class LuceneQueriesPeerPRRedundancyDUnitTest extends LuceneQueriesPRBase
() -> Awaitility.await().atMost(60, TimeUnit.SECONDS).until(basicGetCache()::isClosed));
}
- @Category(FlakyTest.class) // GEODE-1824
@Test
public void returnCorrectResultsWhenCloseCacheHappensOnPartialIndexWrite()
throws InterruptedException {
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 5ec09f1..5fe2df5 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
@@ -84,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, TimeUnit.MILLISECONDS);
+ service.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
PartitionedRegion fileRegion = (PartitionedRegion) cache.getRegion(aeqId + ".files");
assertNotNull(fileRegion);
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 aaa6dbd..1042661 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,19 +14,12 @@
*/
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;
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;
-import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.lucene.LuceneIndex;
import org.apache.geode.test.fake.Fakes;
import org.apache.geode.test.junit.categories.UnitTest;
@@ -48,31 +41,4 @@ public class LuceneIndexImplJUnitTest {
index = new LuceneIndexForPartitionedRegion(INDEX, REGION, cache);
}
- @Test
- public void waitUnitFlushedWithMissingAEQThrowsIllegalArgument() throws Exception {
- thrown.expect(IllegalArgumentException.class);
- index.waitUntilFlushed(MAX_WAIT, TimeUnit.MILLISECONDS);
- }
-
- @Test
- public void waitUnitFlushedWaitsForFlush() throws Exception {
- final String expectedIndexName = LuceneServiceImpl.getUniqueIndexName(INDEX, REGION);
- final AsyncEventQueueImpl queue = mock(AsyncEventQueueImpl.class);
- when(cache.getAsyncEventQueue(eq(expectedIndexName))).thenReturn(queue);
-
- AtomicInteger callCount = new AtomicInteger();
- when(queue.size()).thenAnswer(invocation -> {
- if (callCount.get() == 0) {
- // when the waitUnitFlushed() called the 2nd time, queue.size() will return 0
- callCount.incrementAndGet();
- return 2;
- } else {
- // when the waitUnitFlushed() called the 2nd time, queue.size() will return 0
- return 0;
- }
- });
- index.waitUntilFlushed(MAX_WAIT, TimeUnit.MILLISECONDS);
- verify(cache).getAsyncEventQueue(eq(expectedIndexName));
- }
-
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 d194081..b50db98 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
@@ -89,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, TimeUnit.MILLISECONDS);
+ service.waitUntilFlushed("index1", "userRegion", 30000, TimeUnit.MILLISECONDS);
RepositoryManager manager = new PartitionedRepositoryManager((LuceneIndexImpl) index, mapper);
IndexRepository repo = manager.getRepository(userRegion, 0, null);
@@ -116,8 +116,9 @@ public class LuceneIndexRecoveryHAIntegrationTest {
private void verifyIndexFinishFlushing(String indexName, String regionName)
throws InterruptedException {
- LuceneIndex index = LuceneServiceProvider.get(cache).getIndex(indexName, regionName);
- boolean flushed = index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
+ LuceneService service = LuceneServiceProvider.get(cache);
+ LuceneIndex index = service.getIndex(indexName, regionName);
+ boolean flushed = service.waitUntilFlushed(indexName, regionName, 60000, TimeUnit.MILLISECONDS);
assertTrue(flushed);
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java
index 1815663..d6ec545 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/LuceneServiceImplJUnitTest.java
@@ -15,12 +15,16 @@
package org.apache.geode.cache.lucene.internal;
import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+import java.lang.reflect.Field;
import java.util.Collections;
+import java.util.concurrent.TimeUnit;
-import org.apache.geode.cache.lucene.LuceneService;
+import org.apache.geode.cache.Region;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.test.junit.categories.UnitTest;
-
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -31,19 +35,37 @@ public class LuceneServiceImplJUnitTest {
@Rule
public ExpectedException thrown = ExpectedException.none();
+ Region region;
+ GemFireCacheImpl cache;
+ LuceneServiceImpl service = new LuceneServiceImpl();
+
+ @Before
+ public void createMocks() throws NoSuchFieldException, SecurityException,
+ IllegalArgumentException, IllegalAccessException {
+ region = mock(Region.class);
+ cache = mock(GemFireCacheImpl.class);
+ Field f = LuceneServiceImpl.class.getDeclaredField("cache");
+ f.setAccessible(true);
+ f.set(service, cache);
+ }
+
@Test
public void shouldThrowIllegalArgumentExceptionIfFieldsAreMissing() {
- LuceneServiceImpl service = new LuceneServiceImpl();
thrown.expect(IllegalArgumentException.class);
service.createIndex("index", "region");
}
@Test
public void shouldThrowIllegalArgumentExceptionIfFieldsMapIsMissing() {
- LuceneServiceImpl service = new LuceneServiceImpl();
thrown.expect(IllegalArgumentException.class);
service.createIndex("index", "region", Collections.emptyMap());
}
+ @Test
+ public void shouldReturnFalseIfRegionNotFoundInWaitUntilFlush() throws InterruptedException {
+ boolean result =
+ service.waitUntilFlushed("dummyIndex", "dummyRegion", 60000, TimeUnit.MILLISECONDS);
+ assertFalse(result);
+ }
}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 2bb5b4b..49aa057 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,8 +610,7 @@ 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,
- TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 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/a6c28a65/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 5c47e13..eb6c882 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
@@ -56,7 +56,7 @@ public class DumpDirectoryFilesIntegrationTest extends LuceneIntegrationTest {
InternalLuceneIndex index =
(InternalLuceneIndex) luceneService.getIndex(INDEX_NAME, REGION_NAME);
- index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
+ luceneService.waitUntilFlushed(INDEX_NAME, REGION_NAME, 60000, TimeUnit.MILLISECONDS);
index.dumpFiles(diskDirRule.get().getAbsolutePath());
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunctionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunctionJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunctionJUnitTest.java
new file mode 100644
index 0000000..f92a296
--- /dev/null
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/WaitUntilFlushedFunctionJUnitTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.cache.lucene.internal.distributed;
+
+import static org.apache.geode.cache.lucene.test.LuceneTestUtilities.DEFAULT_FIELD;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
+import org.apache.geode.cache.execute.ResultSender;
+import org.apache.geode.cache.lucene.internal.InternalLuceneService;
+import org.apache.geode.cache.lucene.internal.LuceneIndexImpl;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.execute.InternalRegionFunctionContext;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+import org.apache.lucene.analysis.Analyzer;
+import org.apache.lucene.analysis.standard.StandardAnalyzer;
+import org.apache.lucene.search.Query;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+@Category(UnitTest.class)
+public class WaitUntilFlushedFunctionJUnitTest {
+
+ String regionPath = "/region";
+ String indexName = "index";
+ final EntryScore<String> r1_1 = new EntryScore<String>("key-1-1", .5f);
+ final EntryScore<String> r1_2 = new EntryScore<String>("key-1-2", .4f);
+ final EntryScore<String> r1_3 = new EntryScore<String>("key-1-3", .3f);
+ final EntryScore<String> r2_1 = new EntryScore<String>("key-2-1", .45f);
+ final EntryScore<String> r2_2 = new EntryScore<String>("key-2-2", .35f);
+
+ InternalRegionFunctionContext mockContext;
+ ResultSender mockResultSender;
+ Region<Object, Object> mockRegion;
+ AsyncEventQueueImpl mockAEQ;
+ InternalLuceneService mockService;
+ LuceneIndexImpl mockIndex;
+ WaitUntilFlushedFunctionContext waitArgs;
+ private InternalCache mockCache;
+
+ @Test
+ public void testExecution() throws Exception {
+ when(mockContext.getDataSet()).thenReturn(mockRegion);
+ when(mockContext.getArguments()).thenReturn(waitArgs);
+ when(mockContext.getResultSender()).thenReturn(mockResultSender);
+ when(mockCache.getAsyncEventQueue(any())).thenReturn(mockAEQ);
+ when(mockAEQ.waitUntilFlushed(10000, TimeUnit.MILLISECONDS)).thenReturn(true);
+
+ WaitUntilFlushedFunction function = new WaitUntilFlushedFunction();
+ function.execute(mockContext);
+
+ ArgumentCaptor<Boolean> resultCaptor = ArgumentCaptor.forClass(Boolean.class);
+ verify(mockResultSender).lastResult(resultCaptor.capture());
+ Boolean result = resultCaptor.getValue();
+
+ assertTrue(result);
+ }
+
+
+ @Test
+ public void testQueryFunctionId() {
+ String id = new WaitUntilFlushedFunction().getId();
+ assertEquals(WaitUntilFlushedFunction.class.getName(), id);
+ }
+
+ @Before
+ public void createMocksAndCommonObjects() throws Exception {
+ mockContext = mock(InternalRegionFunctionContext.class);
+ mockResultSender = mock(ResultSender.class);
+ mockRegion = mock(Region.class);
+ mockAEQ = mock(AsyncEventQueueImpl.class);
+
+ mockIndex = mock(LuceneIndexImpl.class);
+ mockService = mock(InternalLuceneService.class);
+ mockCache = mock(InternalCache.class);
+
+ waitArgs = new WaitUntilFlushedFunctionContext(indexName, 10000, TimeUnit.MILLISECONDS);
+
+ when(mockRegion.getCache()).thenReturn(mockCache);
+ when(mockRegion.getFullPath()).thenReturn(regionPath);
+ when(mockCache.getService(any())).thenReturn(mockService);
+ when(mockService.getIndex(eq("index"), eq(regionPath))).thenReturn(mockIndex);
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/IndexRepositorySpy.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/IndexRepositorySpy.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/IndexRepositorySpy.java
index d363903..aac3f97 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/IndexRepositorySpy.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/IndexRepositorySpy.java
@@ -55,6 +55,9 @@ public class IndexRepositorySpy extends IndexRepositoryFactory {
LuceneIndexForPartitionedRegion indexForPR = (LuceneIndexForPartitionedRegion) index;
final IndexRepository indexRepo =
super.computeIndexRepository(bucketId, serializer, index, userRegion, oldRepository);
+ if (indexRepo == null) {
+ return null;
+ }
final IndexRepository spy = Mockito.spy(indexRepo);
Answer invokeBeforeWrite = invocation -> {
http://git-wip-us.apache.org/repos/asf/geode/blob/a6c28a65/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 9f062bb..5a6f774 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
@@ -192,7 +192,8 @@ public class LuceneTestUtilities {
throws InterruptedException {
LuceneService luceneService = LuceneServiceProvider.get(cache);
LuceneIndex index = luceneService.getIndex(indexName, regionName);
- boolean flushed = index.waitUntilFlushed(60000, TimeUnit.MILLISECONDS);
+ boolean flushed =
+ luceneService.waitUntilFlushed(indexName, regionName, 60000, TimeUnit.MILLISECONDS);
assertTrue(flushed);
}