You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by kl...@apache.org on 2017/04/21 20:44:14 UTC
[1/4] geode git commit: GEODE-576 & GEODE-516:
GemFireDeadlockDetectorDUnitTest failures [Forced Update!]
Repository: geode
Updated Branches:
refs/heads/feature/GEODE-2632-5 bbb3b153f -> 222ebf722 (forced update)
GEODE-576 & GEODE-516: GemFireDeadlockDetectorDUnitTest failures
Replaced pauses with Awaitility. Modified asyncs to use the DUnit
blackboard to synchronize their actions for repeatable behavior.
Cleaned up static locks to allow their reuse in other tests or in
repeating the same test.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/50686b0b
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/50686b0b
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/50686b0b
Branch: refs/heads/feature/GEODE-2632-5
Commit: 50686b0b44024d2bcbb4bea8a36ce3a40ac158c2
Parents: 5891ed7
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Fri Apr 21 09:05:07 2017 -0700
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Fri Apr 21 09:06:14 2017 -0700
----------------------------------------------------------------------
.../GemFireDeadlockDetectorDUnitTest.java | 116 +++++++++++--------
.../geode/test/dunit/DUnitBlackboard.java | 13 +++
.../test/dunit/internal/InternalBlackboard.java | 5 +
.../dunit/internal/InternalBlackboardImpl.java | 5 +
4 files changed, 91 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/50686b0b/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java
index e0bbde0..4a03c2d 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java
@@ -14,34 +14,14 @@
*/
package org.apache.geode.distributed.internal.deadlock;
-import org.apache.geode.test.dunit.ThreadUtils;
-import org.junit.experimental.categories.Category;
-import org.junit.Test;
-
-import static org.junit.Assert.*;
-
-import org.awaitility.Awaitility;
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.junit.experimental.categories.Category;
-
-import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.ResultCollector;
-import org.apache.geode.cache30.CacheTestCase;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.LockServiceDestroyedException;
@@ -54,7 +34,20 @@ import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
-import org.apache.geode.test.junit.categories.FlakyTest;
+import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
+import org.awaitility.Awaitility;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
@Category(DistributedTest.class)
public class GemFireDeadlockDetectorDUnitTest extends JUnit4CacheTestCase {
@@ -108,7 +101,8 @@ public class GemFireDeadlockDetectorDUnitTest extends JUnit4CacheTestCase {
private static final Lock lock = new ReentrantLock();
- @Category(FlakyTest.class) // GEODE-516 & GEODE-576: async actions, thread sleeps, time sensitive
+ // @Category(FlakyTest.class) // GEODE-516 & GEODE-576: async actions, thread sleeps, time
+ // sensitive
@Test
public void testDistributedDeadlockWithFunction() throws Throwable {
Host host = Host.getHost(0);
@@ -117,41 +111,62 @@ public class GemFireDeadlockDetectorDUnitTest extends JUnit4CacheTestCase {
getSystem();
InternalDistributedMember member1 = createCache(vm0);
final InternalDistributedMember member2 = createCache(vm1);
+ getBlackboard().initBlackboard();
// Have two threads lock locks on different members in different orders.
+ String gateOnMember1 = "gateOnMember1";
+ String gateOnMember2 = "gateOnMember2";
// This thread locks the lock member1 first, then member2.
- AsyncInvocation async1 = lockTheLocks(vm0, member2);
- // This thread locks the lock member2 first, then member1.
- AsyncInvocation async2 = lockTheLocks(vm1, member1);
+ AsyncInvocation async1 = lockTheLocks(vm0, member2, gateOnMember1, gateOnMember2);
- Thread.sleep(5000);
- GemFireDeadlockDetector detect = new GemFireDeadlockDetector();
- LinkedList<Dependency> deadlock = detect.find().findCycle();
- LogWriterUtils.getLogWriter().info("Deadlock=" + DeadlockDetector.prettyFormat(deadlock));
- assertEquals(8, deadlock.size());
- stopStuckThreads();
- async1.getResult(30000);
- async2.getResult(30000);
+ // This thread locks the lock member2 first, then member1.
+ AsyncInvocation async2 = lockTheLocks(vm1, member1, gateOnMember2, gateOnMember1);
+ try {
+ final LinkedList<Dependency> deadlockHolder[] = new LinkedList[1];
+ Awaitility.await("waiting for deadlock").atMost(20, TimeUnit.SECONDS).until(() -> {
+ GemFireDeadlockDetector detect = new GemFireDeadlockDetector();
+ LinkedList<Dependency> deadlock = detect.find().findCycle();
+ if (deadlock != null) {
+ deadlockHolder[0] = deadlock;
+ }
+ return deadlock != null;
+ });
+ LinkedList<Dependency> deadlock = deadlockHolder[0];
+ LogWriterUtils.getLogWriter().info("Deadlock=" + DeadlockDetector.prettyFormat(deadlock));
+ assertEquals(8, deadlock.size());
+ stopStuckThreads();
+ } finally {
+ try {
+ waitForAsyncInvocation(async1, 45, TimeUnit.SECONDS);
+ } finally {
+ waitForAsyncInvocation(async2, 45, TimeUnit.SECONDS);
+ }
+ }
}
- private AsyncInvocation lockTheLocks(VM vm0, final InternalDistributedMember member) {
+ private AsyncInvocation lockTheLocks(VM vm0, final InternalDistributedMember member,
+ final String gateToSignal, final String gateToWaitOn) {
return vm0.invokeAsync(new SerializableRunnable() {
public void run() {
lock.lock();
try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- Assert.fail("interrupted", e);
+ try {
+ getBlackboard().signalGate(gateToSignal);
+ getBlackboard().waitForGate(gateToWaitOn, 10, TimeUnit.SECONDS);
+ } catch (TimeoutException | InterruptedException e) {
+ throw new RuntimeException("failed", e);
+ }
+ ResultCollector collector = FunctionService.onMember(member).execute(new TestFunction());
+ // wait the function to lock the lock on member.
+ collector.getResult();
+ } finally {
+ lock.unlock();
}
- ResultCollector collector = FunctionService.onMember(member).execute(new TestFunction());
- // wait the function to lock the lock on member.
- collector.getResult();
- lock.unlock();
}
});
}
@@ -244,14 +259,19 @@ public class GemFireDeadlockDetectorDUnitTest extends JUnit4CacheTestCase {
public void execute(FunctionContext context) {
+ boolean acquired = false;
try {
stuckThreads.add(Thread.currentThread());
- lock.tryLock(LOCK_WAIT_TIME, TimeUnit.SECONDS);
+ acquired = lock.tryLock(LOCK_WAIT_TIME, TimeUnit.SECONDS);
} catch (InterruptedException e) {
- // ingore
+ // ignore
+ } finally {
+ if (acquired) {
+ lock.unlock();
+ }
+ stuckThreads.remove(Thread.currentThread());
+ context.getResultSender().lastResult(null);
}
- stuckThreads.remove(Thread.currentThread());
- context.getResultSender().lastResult(null);
}
public String getId() {
http://git-wip-us.apache.org/repos/asf/geode/blob/50686b0b/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java b/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java
index a097cd4e..62c92bd 100755
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java
@@ -56,6 +56,7 @@ public class DUnitBlackboard {
* signals a boolean gate
*/
public void signalGate(String gateName) {
+ // System.out.println(Thread.currentThread().getName()+": signaling gate " + gateName);
try {
blackboard.signalGate(gateName);
} catch (RemoteException e) {
@@ -68,6 +69,7 @@ public class DUnitBlackboard {
*/
public void waitForGate(String gateName, long timeout, TimeUnit units)
throws TimeoutException, InterruptedException {
+ // System.out.println(Thread.currentThread().getName()+": waiting for gate " + gateName);
try {
blackboard.waitForGate(gateName, timeout, units);
} catch (RemoteException e) {
@@ -77,6 +79,17 @@ public class DUnitBlackboard {
}
/**
+ * clear a gate
+ */
+ public void clearGate(String gateName) {
+ try {
+ blackboard.clearGate(gateName);
+ } catch (RemoteException e) {
+ throw new RuntimeException("remote call failed", e);
+ }
+ }
+
+ /**
* test to see if a gate has been signeled
*/
public boolean isGateSignaled(String gateName) {
http://git-wip-us.apache.org/repos/asf/geode/blob/50686b0b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
index 63f833b..bc5b9b7 100755
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
@@ -50,6 +50,11 @@ public interface InternalBlackboard extends Remote, Serializable {
throws RemoteException, TimeoutException, InterruptedException;
/**
+ * clears a gate
+ */
+ void clearGate(String gateName) throws RemoteException;
+
+ /**
* test to see if a gate has been signeled
*/
boolean isGateSignaled(String gateName) throws RemoteException;
http://git-wip-us.apache.org/repos/asf/geode/blob/50686b0b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
index e7657ed..feeae15 100755
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
@@ -79,6 +79,11 @@ public class InternalBlackboardImpl extends UnicastRemoteObject implements Inter
}
@Override
+ public void clearGate(final String gateName) throws RemoteException {
+ gates.remove(gateName);
+ }
+
+ @Override
public void signalGate(final String gateName) throws RemoteException {
gates.put(gateName, Boolean.TRUE);
}
[4/4] geode git commit: 2632: refactor code to use InternalCache
instead of GemFireCacheImpl
Posted by kl...@apache.org.
2632: refactor code to use InternalCache instead of GemFireCacheImpl
* minor cleanup also
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/222ebf72
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/222ebf72
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/222ebf72
Branch: refs/heads/feature/GEODE-2632-5
Commit: 222ebf7223ef92ff2c6728944863b27dd27b4737
Parents: 0862174
Author: Kirk Lund <kl...@apache.org>
Authored: Wed Apr 19 14:41:42 2017 -0700
Committer: Kirk Lund <kl...@apache.org>
Committed: Fri Apr 21 13:43:41 2017 -0700
----------------------------------------------------------------------
.../query/internal/cq/CqServiceProvider.java | 22 +-
.../query/internal/cq/spi/CqServiceFactory.java | 8 +-
.../cache/query/internal/cq/ClientCQImpl.java | 95 +--
.../cache/query/internal/cq/CqQueryImpl.java | 91 ++-
.../query/internal/cq/CqServiceFactoryImpl.java | 17 +-
.../cache/query/internal/cq/CqServiceImpl.java | 673 ++++---------------
.../internal/cq/CqServiceStatisticsImpl.java | 21 +-
.../query/internal/cq/CqServiceVsdStats.java | 73 +-
.../cache/query/internal/cq/ServerCQImpl.java | 121 +---
.../cache/tier/sockets/command/ExecuteCQ.java | 4 +-
.../cache/tier/sockets/command/ExecuteCQ61.java | 4 +-
.../tier/sockets/command/GetDurableCQs.java | 4 +-
.../cache/query/cq/dunit/CqStatsDUnitTest.java | 44 +-
.../cq/dunit/CqStatsUsingPoolDUnitTest.java | 47 +-
.../TopEntriesFunctionCollector.java | 22 +-
.../LuceneQueryFunctionJUnitTest.java | 54 +-
.../TopEntriesCollectorJUnitTest.java | 23 +-
.../TopEntriesFunctionCollectorJUnitTest.java | 48 +-
.../distributed/TopEntriesJUnitTest.java | 44 +-
.../cache/lucene/test/LuceneTestUtilities.java | 16 +
20 files changed, 442 insertions(+), 989 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java
index cded9c3..90fbf4b 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java
@@ -16,7 +16,7 @@ package org.apache.geode.cache.query.internal.cq;
import org.apache.geode.cache.query.internal.cq.spi.CqServiceFactory;
import org.apache.geode.distributed.internal.DistributionConfig;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import java.io.DataInput;
import java.io.IOException;
@@ -26,17 +26,19 @@ import java.util.ServiceLoader;
public class CqServiceProvider {
private static final CqServiceFactory factory;
- // System property to maintain the CQ event references for optimizing the updates.
- // This will allows to run the CQ query only once during update events.
+
+ /**
+ * System property to maintain the CQ event references for optimizing the updates. This will allow
+ * running the CQ query only once during update events.
+ */
public static boolean MAINTAIN_KEYS = Boolean
- .valueOf(System.getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.MAINTAIN_KEYS", "true"))
- .booleanValue();
+ .valueOf(System.getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.MAINTAIN_KEYS", "true"));
+
/**
* A debug flag used for testing vMotion during CQ registration
*/
public static boolean VMOTION_DURING_CQ_REGISTRATION_FLAG = false;
-
static {
ServiceLoader<CqServiceFactory> loader = ServiceLoader.load(CqServiceFactory.class);
Iterator<CqServiceFactory> itr = loader.iterator();
@@ -48,8 +50,7 @@ public class CqServiceProvider {
}
}
- public static CqService create(GemFireCacheImpl cache) {
-
+ public static CqService create(InternalCache cache) {
if (factory == null) {
return new MissingCqService();
}
@@ -63,10 +64,7 @@ public class CqServiceProvider {
} else {
return factory.readCqQuery(in);
}
-
}
- private CqServiceProvider() {
-
- }
+ private CqServiceProvider() {}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java
index 68ebbd5..2b8a47e 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java
@@ -19,16 +19,16 @@ import java.io.IOException;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
public interface CqServiceFactory {
- public void initialize();
+ void initialize();
/**
* Create a new CqService for the given cache
*/
- public CqService create(GemFireCacheImpl cache);
+ CqService create(InternalCache cache);
- public ServerCQ readCqQuery(DataInput in) throws ClassNotFoundException, IOException;
+ ServerCQ readCqQuery(DataInput in) throws ClassNotFoundException, IOException;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java
index 00a0aa5..111bf84 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java
@@ -35,7 +35,7 @@ import org.apache.geode.cache.query.CqResults;
import org.apache.geode.cache.query.CqStatusListener;
import org.apache.geode.cache.query.RegionNotFoundException;
import org.apache.geode.cache.query.internal.CqStateImpl;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
@@ -57,7 +57,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
*/
private volatile ConcurrentLinkedQueue<CqEventImpl> queuedEvents = null;
- public final Object queuedEventsSynchObject = new Object();
+ final Object queuedEventsSynchObject = new Object();
private boolean connected = false;
@@ -73,22 +73,15 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
return this.cqName;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#getCQProxy()
- */
- public ServerCQProxyImpl getCQProxy() {
+ ServerCQProxyImpl getCQProxy() {
return this.cqProxy;
}
/**
* Initializes the connection using the pool from the client region. Also sets the cqBaseRegion
* value of this CQ.
- *
- * @throws CqException
*/
- public void initConnectionProxy() throws CqException, RegionNotFoundException {
+ private void initConnectionProxy() throws CqException, RegionNotFoundException {
cqBaseRegion = (LocalRegion) cqService.getCache().getRegion(regionName);
// Check if the region exists on the local cache.
// In the current implementation of 5.1 the Server Connection is (ConnectionProxyImpl)
@@ -113,17 +106,9 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
throw new CqException(
"Unable to get the connection pool. The Region does not have a pool configured.");
}
-
- // if (proxy == null) {
- // throw new
- // CqException(LocalizedStrings.CqQueryImpl_UNABLE_TO_GET_THE_CONNECTIONPROXY_THE_REGION_MAY_NOT_HAVE_A_BRIDGEWRITER_OR_BRIDGECLIENT_INSTALLED_ON_IT.toLocalizedString());
- // } else if(!proxy.getEstablishCallbackConnection()){
- // throw new
- // CqException(LocalizedStrings.CqQueryImpl_THE_ESTABLISHCALLBACKCONNECTION_ON_BRIDGEWRITER_CLIENT_INSTALLED_ON_REGION_0_IS_SET_TO_FALSE
- // .toLocalizedString(regionName));
- // }
}
+ @Override
public void close() throws CqClosedException, CqException {
this.close(true);
}
@@ -182,15 +167,15 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (cqProxy == null || !sendRequestToServer || isClosed) {
// Stat update.
if (stateBeforeClosing == CqStateImpl.RUNNING) {
- cqService.stats.decCqsActive();
+ cqService.stats().decCqsActive();
} else if (stateBeforeClosing == CqStateImpl.STOPPED) {
- cqService.stats.decCqsStopped();
+ cqService.stats().decCqsStopped();
}
// Set the state to close, and update stats
this.cqState.setState(CqStateImpl.CLOSED);
- cqService.stats.incCqsClosed();
- cqService.stats.decCqsOnClient();
+ cqService.stats().incCqsClosed();
+ cqService.stats().decCqsOnClient();
if (this.stats != null)
this.stats.close();
} else {
@@ -201,7 +186,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (exception != null) {
throw new CqException(
LocalizedStrings.CqQueryImpl_FAILED_TO_CLOSE_THE_CQ_CQNAME_0_ERROR_FROM_LAST_ENDPOINT_1
- .toLocalizedString(new Object[] {this.cqName, exception.getLocalizedMessage()}),
+ .toLocalizedString(this.cqName, exception.getLocalizedMessage()),
exception.getCause());
} else {
throw new CqException(
@@ -261,31 +246,28 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
/**
* Clears the resource used by CQ.
- *
- * @throws CqException
*/
+ @Override
protected void cleanup() throws CqException {
this.cqService.removeFromBaseRegionToCqNameMap(this.regionName, this.getServerCqName());
}
+ @Override
public CqAttributes getCqAttributes() {
return cqAttributes;
}
-
-
/**
* @return Returns the cqListeners.
*/
public CqListener[] getCqListeners() {
-
return cqAttributes.getCqListeners();
}
-
/**
* Start or resume executing the query.
*/
+ @Override
public void execute() throws CqClosedException, RegionNotFoundException, CqException {
executeCqOnRedundantsAndPrimary(false);
}
@@ -293,7 +275,8 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
/**
* Start or resume executing the query. Gets or updates the CQ results and returns them.
*/
- public CqResults executeWithInitialResults()
+ @Override
+ public <E> CqResults<E> executeWithInitialResults()
throws CqClosedException, RegionNotFoundException, CqException {
synchronized (queuedEventsSynchObject) {
@@ -320,16 +303,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
CqResults initialResults;
try {
initialResults = (CqResults) executeCqOnRedundantsAndPrimary(true);
- } catch (CqClosedException e) {
- queuedEvents = null;
- throw e;
- } catch (RegionNotFoundException e) {
- queuedEvents = null;
- throw e;
- } catch (CqException e) {
- queuedEvents = null;
- throw e;
- } catch (RuntimeException e) {
+ } catch (RegionNotFoundException | CqException | RuntimeException e) {
queuedEvents = null;
throw e;
}
@@ -343,6 +317,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (!this.queuedEvents.isEmpty()) {
try {
Runnable r = new Runnable() {
+ @Override
public void run() {
Object[] eventArray = null;
if (CqQueryImpl.testHook != null) {
@@ -395,7 +370,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
* @param executeWithInitialResults boolean
* @return Object SelectResults in case of executeWithInitialResults
*/
- public Object executeCqOnRedundantsAndPrimary(boolean executeWithInitialResults)
+ private Object executeCqOnRedundantsAndPrimary(boolean executeWithInitialResults)
throws CqClosedException, RegionNotFoundException, CqException {
Object initialResults = null;
@@ -461,8 +436,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
} else {
String errMsg =
LocalizedStrings.CqQueryImpl_FAILED_TO_EXECUTE_THE_CQ_CQNAME_0_QUERY_STRING_IS_1_ERROR_FROM_LAST_SERVER_2
- .toLocalizedString(
- new Object[] {this.cqName, this.queryString, ex.getLocalizedMessage()});
+ .toLocalizedString(this.cqName, this.queryString, ex.getLocalizedMessage());
if (logger.isDebugEnabled()) {
logger.debug(errMsg, ex);
}
@@ -498,8 +472,8 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
}
}
// Update CQ-base region for book keeping.
- this.cqService.stats.incCqsActive();
- this.cqService.stats.decCqsStopped();
+ this.cqService.stats().incCqsActive();
+ this.cqService.stats().decCqsStopped();
return initialResults;
}
@@ -509,23 +483,22 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
* @return true if shutdown in progress else false.
*/
private boolean shutdownInProgress() {
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ InternalCache cache = cqService.getInternalCache();
if (cache == null || cache.isClosed()) {
return true; // bail, things are shutting down
}
-
String reason = cqProxy.getPool().getCancelCriterion().cancelInProgress();
if (reason != null) {
return true;
}
return false;
-
}
/**
* Stop or pause executing the query.
*/
+ @Override
public void stop() throws CqClosedException, CqException {
boolean isStopped = false;
synchronized (this.cqState) {
@@ -558,8 +531,8 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (cqProxy == null || isStopped) {
// Change state and stats on the client side
this.cqState.setState(CqStateImpl.STOPPED);
- this.cqService.stats.incCqsStopped();
- this.cqService.stats.decCqsActive();
+ this.cqService.stats().incCqsStopped();
+ this.cqService.stats().decCqsActive();
if (logger.isDebugEnabled()) {
logger.debug("Successfully stopped the CQ. {}", cqName);
}
@@ -568,7 +541,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (exception != null) {
throw new CqException(
LocalizedStrings.CqQueryImpl_FAILED_TO_STOP_THE_CQ_CQNAME_0_ERROR_FROM_LAST_SERVER_1
- .toLocalizedString(new Object[] {this.cqName, exception.getLocalizedMessage()}),
+ .toLocalizedString(this.cqName, exception.getLocalizedMessage()),
exception.getCause());
} else {
throw new CqException(
@@ -579,24 +552,15 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
}
}
+ @Override
public CqAttributesMutator getCqAttributesMutator() {
return (CqAttributesMutator) this.cqAttributes;
}
-
- public ConcurrentLinkedQueue<CqEventImpl> getQueuedEvents() {
+ ConcurrentLinkedQueue<CqEventImpl> getQueuedEvents() {
return this.queuedEvents;
}
-
-
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqQuery2#setProxyCache(org.apache.geode.cache.
- * client.internal.ProxyCache)
- */
@Override
public void setProxyCache(ProxyCache proxyCache) {
this.proxyCache = proxyCache;
@@ -612,7 +576,6 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
@Override
public void createOn(Connection conn, boolean isDurable) {
-
byte regionDataPolicyOrdinal = getCqBaseRegion() == null ? (byte) 0
: getCqBaseRegion().getAttributes().getDataPolicy().ordinal;
@@ -620,6 +583,4 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
this.cqProxy.createOn(getName(), conn, getQueryString(), state, isDurable,
regionDataPolicyOrdinal);
}
-
-
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java
index 22b4137..07e3171 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java
@@ -21,11 +21,9 @@ import java.util.Set;
import org.apache.logging.log4j.Logger;
import org.apache.geode.StatisticsFactory;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.query.CqClosedException;
import org.apache.geode.cache.query.CqEvent;
import org.apache.geode.cache.query.CqException;
-import org.apache.geode.cache.query.CqExistsException;
import org.apache.geode.cache.query.CqState;
import org.apache.geode.cache.query.CqStatistics;
import org.apache.geode.cache.query.Query;
@@ -38,7 +36,7 @@ import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.ExecutionContext;
import org.apache.geode.cache.query.internal.QueryExecutionContext;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.InternalLogWriter;
@@ -58,13 +56,13 @@ public abstract class CqQueryImpl implements InternalCqQuery {
protected String queryString;
- protected static final Object TOKEN = new Object();
+ static final Object TOKEN = new Object();
- protected LocalRegion cqBaseRegion;
+ LocalRegion cqBaseRegion;
protected Query query = null;
- protected InternalLogWriter securityLogWriter;
+ InternalLogWriter securityLogWriter;
protected CqServiceImpl cqService;
@@ -72,14 +70,14 @@ public abstract class CqQueryImpl implements InternalCqQuery {
protected boolean isDurable = false;
- // Stats counters
- protected CqStatisticsImpl cqStats;
+ /** Stats counters */
+ private CqStatisticsImpl cqStats;
protected CqQueryVsdStats stats;
protected final CqStateImpl cqState = new CqStateImpl();
- protected ExecutionContext queryExecutionContext = null;
+ private ExecutionContext queryExecutionContext = null;
public static TestHook testHook = null;
@@ -100,6 +98,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
/**
* returns CQ name
*/
+ @Override
public String getName() {
return this.cqName;
}
@@ -109,6 +108,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
this.cqName = cqName;
}
+ @Override
public void setCqService(CqService cqService) {
this.cqService = (CqServiceImpl) cqService;
}
@@ -121,25 +121,24 @@ public abstract class CqQueryImpl implements InternalCqQuery {
return this.regionName;
}
- public void updateCqCreateStats() {
+ void updateCqCreateStats() {
// Initialize the VSD statistics
StatisticsFactory factory = cqService.getCache().getDistributedSystem();
this.stats = new CqQueryVsdStats(factory, getServerCqName());
this.cqStats = new CqStatisticsImpl(this);
// Update statistics with CQ creation.
- this.cqService.stats.incCqsStopped();
- this.cqService.stats.incCqsCreated();
- this.cqService.stats.incCqsOnClient();
+ this.cqService.stats().incCqsStopped();
+ this.cqService.stats().incCqsCreated();
+ this.cqService.stats().incCqsOnClient();
}
/**
* Validates the CQ. Checks for cq constraints. Also sets the base region name.
*/
- public void validateCq() {
- Cache cache = cqService.getCache();
- DefaultQuery locQuery =
- (DefaultQuery) ((GemFireCacheImpl) cache).getLocalQueryService().newQuery(this.queryString);
+ void validateCq() {
+ InternalCache cache = cqService.getInternalCache();
+ DefaultQuery locQuery = (DefaultQuery) cache.getLocalQueryService().newQuery(this.queryString);
this.query = locQuery;
// assert locQuery != null;
@@ -221,10 +220,8 @@ public abstract class CqQueryImpl implements InternalCqQuery {
/**
* Removes the CQ from CQ repository.
- *
- * @throws CqException
*/
- protected void removeFromCqMap() throws CqException {
+ void removeFromCqMap() throws CqException {
try {
cqService.removeCq(this.getServerCqName());
} catch (Exception ex) {
@@ -243,6 +240,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
/**
* Returns the QueryString of this CQ.
*/
+ @Override
public String getQueryString() {
return queryString;
}
@@ -252,23 +250,16 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return the Query for the query string
*/
+ @Override
public Query getQuery() {
return query;
}
-
- /**
- * @see org.apache.geode.cache.query.CqQuery#getStatistics()
- */
+ @Override
public CqStatistics getStatistics() {
return cqStats;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#getCqBaseRegion()
- */
@Override
public LocalRegion getCqBaseRegion() {
return this.cqBaseRegion;
@@ -279,11 +270,12 @@ public abstract class CqQueryImpl implements InternalCqQuery {
/**
* @return Returns the Region name on which this cq is created.
*/
- public String getBaseRegionName() {
+ String getBaseRegionName() {
return this.regionName;
}
+ @Override
public abstract String getServerCqName();
/**
@@ -291,15 +283,11 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return STOPPED RUNNING or CLOSED
*/
+ @Override
public CqState getState() {
return this.cqState;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#setCqState(int)
- */
@Override
public void setCqState(int state) {
if (this.isClosed()) {
@@ -309,18 +297,13 @@ public abstract class CqQueryImpl implements InternalCqQuery {
synchronized (cqState) {
if (state == CqStateImpl.RUNNING) {
- if (this.isRunning()) {
- // throw new
- // IllegalStateException(LocalizedStrings.CqQueryImpl_CQ_IS_NOT_IN_RUNNING_STATE_STOP_CQ_DOES_NOT_APPLY_CQNAME_0
- // .toLocalizedString(this.cqName));
- }
this.cqState.setState(CqStateImpl.RUNNING);
- this.cqService.stats.incCqsActive();
- this.cqService.stats.decCqsStopped();
+ this.cqService.stats().incCqsActive();
+ this.cqService.stats().decCqsStopped();
} else if (state == CqStateImpl.STOPPED) {
this.cqState.setState(CqStateImpl.STOPPED);
- this.cqService.stats.incCqsStopped();
- this.cqService.stats.decCqsActive();
+ this.cqService.stats().incCqsStopped();
+ this.cqService.stats().decCqsActive();
} else if (state == CqStateImpl.CLOSING) {
this.cqState.setState(state);
}
@@ -332,7 +315,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @param cqEvent object
*/
- public void updateStats(CqEvent cqEvent) {
+ void updateStats(CqEvent cqEvent) {
this.stats.updateStats(cqEvent); // Stats for VSD
}
@@ -341,15 +324,17 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return true if running, false otherwise
*/
+ @Override
public boolean isRunning() {
return this.cqState.isRunning();
}
/**
- * Return true if the CQ is in Sstopped state
+ * Return true if the CQ is in stopped state
*
* @return true if stopped, false otherwise
*/
+ @Override
public boolean isStopped() {
return this.cqState.isStopped();
}
@@ -359,6 +344,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return true if closed, false otherwise
*/
+ @Override
public boolean isClosed() {
return this.cqState.isClosed();
}
@@ -377,6 +363,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return true if durable, false otherwise
*/
+ @Override
public boolean isDurable() {
return this.isDurable;
}
@@ -391,22 +378,22 @@ public abstract class CqQueryImpl implements InternalCqQuery {
return stats;
}
- public ExecutionContext getQueryExecutionContext() {
+ ExecutionContext getQueryExecutionContext() {
return queryExecutionContext;
}
- public void setQueryExecutionContext(ExecutionContext queryExecutionContext) {
+ private void setQueryExecutionContext(ExecutionContext queryExecutionContext) {
this.queryExecutionContext = queryExecutionContext;
}
/** Test Hook */
public interface TestHook {
- public void pauseUntilReady();
+ void pauseUntilReady();
- public void ready();
+ void ready();
- public int numQueuedEvents();
+ int numQueuedEvents();
- public void setEventCount(int count);
+ void setEventCount(int count);
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java
index db90632..9cc2eea 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java
@@ -22,7 +22,7 @@ import java.util.Map;
import org.apache.geode.cache.query.internal.cq.spi.CqServiceFactory;
import org.apache.geode.internal.Version;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.CommandInitializer;
@@ -36,14 +36,13 @@ import org.apache.geode.internal.cache.tier.sockets.command.StopCQ;
public class CqServiceFactoryImpl implements CqServiceFactory {
+ @Override
public void initialize() {
- {
- Map<Version, Command> versions = new HashMap<Version, Command>();
- versions.put(Version.GFE_57, ExecuteCQ.getCommand());
- versions.put(Version.GFE_61, ExecuteCQ61.getCommand());
- CommandInitializer.registerCommand(MessageType.EXECUTECQ_MSG_TYPE, versions);
- CommandInitializer.registerCommand(MessageType.EXECUTECQ_WITH_IR_MSG_TYPE, versions);
- }
+ Map<Version, Command> versions = new HashMap<>();
+ versions.put(Version.GFE_57, ExecuteCQ.getCommand());
+ versions.put(Version.GFE_61, ExecuteCQ61.getCommand());
+ CommandInitializer.registerCommand(MessageType.EXECUTECQ_MSG_TYPE, versions);
+ CommandInitializer.registerCommand(MessageType.EXECUTECQ_WITH_IR_MSG_TYPE, versions);
CommandInitializer.registerCommand(MessageType.GETCQSTATS_MSG_TYPE,
Collections.singletonMap(Version.GFE_57, GetCQStats.getCommand()));
@@ -58,7 +57,7 @@ public class CqServiceFactoryImpl implements CqServiceFactory {
}
@Override
- public CqService create(GemFireCacheImpl cache) {
+ public CqService create(InternalCache cache) {
return new CqServiceImpl(cache);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java
index f1ca832..570c06c 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java
@@ -14,19 +14,63 @@
*/
package org.apache.geode.cache.query.internal.cq;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.apache.logging.log4j.Logger;
+
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
-import org.apache.geode.cache.*;
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheEvent;
+import org.apache.geode.cache.CacheLoaderException;
+import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.RegionEvent;
+import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.client.Pool;
-import org.apache.geode.cache.client.internal.*;
-import org.apache.geode.cache.query.*;
-import org.apache.geode.cache.query.internal.*;
+import org.apache.geode.cache.client.internal.GetEventValueOp;
+import org.apache.geode.cache.client.internal.InternalPool;
+import org.apache.geode.cache.client.internal.QueueManager;
+import org.apache.geode.cache.client.internal.ServerCQProxyImpl;
+import org.apache.geode.cache.client.internal.UserAttributes;
+import org.apache.geode.cache.query.CqAttributes;
+import org.apache.geode.cache.query.CqClosedException;
+import org.apache.geode.cache.query.CqException;
+import org.apache.geode.cache.query.CqExistsException;
+import org.apache.geode.cache.query.CqListener;
+import org.apache.geode.cache.query.CqQuery;
+import org.apache.geode.cache.query.CqServiceStatistics;
+import org.apache.geode.cache.query.CqStatusListener;
+import org.apache.geode.cache.query.QueryException;
+import org.apache.geode.cache.query.QueryInvalidException;
+import org.apache.geode.cache.query.RegionNotFoundException;
+import org.apache.geode.cache.query.SelectResults;
+import org.apache.geode.cache.query.internal.CompiledSelect;
+import org.apache.geode.cache.query.internal.CqQueryVsdStats;
+import org.apache.geode.cache.query.internal.CqStateImpl;
+import org.apache.geode.cache.query.internal.DefaultQuery;
+import org.apache.geode.cache.query.internal.ExecutionContext;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.i18n.StringId;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
-import org.apache.geode.internal.cache.*;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.FilterProfile;
+import org.apache.geode.internal.cache.FilterRoutingInfo;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
@@ -35,57 +79,43 @@ import org.apache.geode.internal.cache.tier.sockets.Part;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
-import org.apache.logging.log4j.Logger;
-
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
/**
- * @since GemFire 5.5
- *
- * Implements the CqService functionality.
- *
- */
-/**
+ * Implements the CqService functionality.
*
+ * @since GemFire 5.5
*/
public final class CqServiceImpl implements CqService {
private static final Logger logger = LogService.getLogger();
- private static final Integer MESSAGE_TYPE_LOCAL_CREATE =
- Integer.valueOf(MessageType.LOCAL_CREATE);
- private static final Integer MESSAGE_TYPE_LOCAL_UPDATE =
- Integer.valueOf(MessageType.LOCAL_UPDATE);
- private static final Integer MESSAGE_TYPE_LOCAL_DESTROY =
- Integer.valueOf(MessageType.LOCAL_DESTROY);
- private static final Integer MESSAGE_TYPE_EXCEPTION = Integer.valueOf(MessageType.EXCEPTION);
+ private static final Integer MESSAGE_TYPE_LOCAL_CREATE = MessageType.LOCAL_CREATE;
+ private static final Integer MESSAGE_TYPE_LOCAL_UPDATE = MessageType.LOCAL_UPDATE;
+ private static final Integer MESSAGE_TYPE_LOCAL_DESTROY = MessageType.LOCAL_DESTROY;
+ private static final Integer MESSAGE_TYPE_EXCEPTION = MessageType.EXCEPTION;
/**
* System property to evaluate the query even though the initial results are not required when cq
* is executed using the execute() method.
*/
- public static boolean EXECUTE_QUERY_DURING_INIT = Boolean
- .valueOf(System
- .getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.EXECUTE_QUERY_DURING_INIT", "true"))
- .booleanValue();
+ public static boolean EXECUTE_QUERY_DURING_INIT = Boolean.valueOf(System
+ .getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.EXECUTE_QUERY_DURING_INIT", "true"));
private static final String CQ_NAME_PREFIX = "GfCq";
- private final Cache cache;
+ private final InternalCache cache;
/**
* Manages cq pools to determine if a status of connect or disconnect needs to be sent out
*/
- private final HashMap<String, Boolean> cqPoolsConnected = new HashMap<String, Boolean>();
-
+ private final HashMap<String, Boolean> cqPoolsConnected = new HashMap<>();
/**
* Manages CQ objects. uses serverCqName as key and CqQueryImpl as value
*
- * @guarded.By cqQueryMapLock
+ * GuardedBy cqQueryMapLock
*/
- private volatile HashMap<String, CqQueryImpl> cqQueryMap = new HashMap<String, CqQueryImpl>();
+ private volatile HashMap<String, CqQueryImpl> cqQueryMap = new HashMap<>();
+
private final Object cqQueryMapLock = new Object();
private volatile boolean isRunning = false;
@@ -93,36 +123,21 @@ public final class CqServiceImpl implements CqService {
/**
* Used by client when multiuser-authentication is true.
*/
- private final HashMap<String, UserAttributes> cqNameToUserAttributesMap =
- new HashMap<String, UserAttributes>();
-
- // private boolean isServer = true;
-
- /*
- * // Map to manage CQ to satisfied CQ events (keys) for optimizing updates. private final HashMap
- * cqToCqEventKeysMap = CqService.MAINTAIN_KEYS ? new HashMap() : null;
- */
+ private final HashMap<String, UserAttributes> cqNameToUserAttributesMap = new HashMap<>();
// Map to manage the similar CQs (having same query - performance optimization).
// With query as key and Set of CQs as values.
private final ConcurrentHashMap matchingCqMap;
// CQ Service statistics
- public final CqServiceStatisticsImpl cqServiceStats;
- public final CqServiceVsdStats stats;
+ private final CqServiceStatisticsImpl cqServiceStats;
+ private final CqServiceVsdStats stats;
// CQ identifier, also used in auto generated CQ names
private volatile long cqId = 1;
- /**
- * Used to synchronize access to CQs in the repository
- */
- final Object cqSync = new Object();
-
/* This is to manage region to CQs map, client side book keeping. */
- private HashMap<String, ArrayList<String>> baseRegionToCqNameMap =
- new HashMap<String, ArrayList<String>>();
-
+ private HashMap<String, ArrayList<String>> baseRegionToCqNameMap = new HashMap<>();
/**
* Access and modification to the contents of this map do not necessarily need to be lock
@@ -135,33 +150,24 @@ public final class CqServiceImpl implements CqService {
/**
* Constructor.
- *
- * @param c The cache used for the service
+ *
+ * @param cache The cache used for the service
*/
- public CqServiceImpl(final Cache c) {
- if (c == null) {
+ public CqServiceImpl(final InternalCache cache) {
+ if (cache == null) {
throw new IllegalStateException(LocalizedStrings.CqService_CACHE_IS_NULL.toLocalizedString());
}
- GemFireCacheImpl gfc = (GemFireCacheImpl) c;
- gfc.getCancelCriterion().checkCancelInProgress(null);
-
- this.cache = gfc;
+ cache.getCancelCriterion().checkCancelInProgress(null);
+ this.cache = cache;
// Initialize the Map which maintains the matching cqs.
this.matchingCqMap = new ConcurrentHashMap<String, HashSet<String>>();
// Initialize the VSD statistics
- StatisticsFactory factory = cache.getDistributedSystem();
+ StatisticsFactory factory = this.cache.getDistributedSystem();
this.stats = new CqServiceVsdStats(factory);
this.cqServiceStats = new CqServiceStatisticsImpl(this);
-
- // final LoggingThreadGroup group =
- // LoggingThreadGroup.createThreadGroup("CqExecutor Threads", logger);
-
- // if (this.cache.getCacheServers().isEmpty()) {
- // isServer = false;
- // }
}
/**
@@ -171,13 +177,14 @@ public final class CqServiceImpl implements CqService {
return this.cache;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#newCq(java.lang.String,
- * java.lang.String, org.apache.geode.cache.query.CqAttributes,
- * org.apache.geode.cache.client.internal.ServerCQProxy, boolean)
- */
+ public InternalCache getInternalCache() {
+ return this.cache;
+ }
+
+ public CqServiceVsdStats stats() {
+ return this.stats;
+ }
+
@Override
public synchronized ClientCQ newCq(String cqName, String queryString, CqAttributes cqAttributes,
InternalPool pool, boolean isDurable)
@@ -242,22 +249,15 @@ public final class CqServiceImpl implements CqService {
return cQuery;
}
-
/**
* Executes the given CqQuery, if the CqQuery for that name is not there it registers the one and
* executes. This is called on the Server.
*
- * @param cqName
- * @param queryString
- * @param cqState
- * @param clientProxyId
- * @param ccn
* @param manageEmptyRegions whether to update the 6.1 emptyRegions map held in the CCN
* @param regionDataPolicy the data policy of the region associated with the query. This is only
* needed if manageEmptyRegions is true.
* @param emptyRegionsMap map of empty regions.
* @throws IllegalStateException if this is called at client side.
- * @throws CqException
*/
@Override
public synchronized ServerCQ executeCq(String cqName, String queryString, int cqState,
@@ -271,7 +271,7 @@ public final class CqServiceImpl implements CqService {
}
String serverCqName = constructServerCqName(cqName, clientProxyId);
- ServerCQImpl cQuery = null;
+ ServerCQImpl cQuery;
// If this CQ is not yet registered in Server, register CQ.
if (!isCqExists(serverCqName)) {
@@ -292,7 +292,6 @@ public final class CqServiceImpl implements CqService {
logger.info(LocalizedMessage.create(
LocalizedStrings.CqService_EXCEPTION_WHILE_REGISTERING_CQ_ON_SERVER_CQNAME___0,
cQuery.getName()));
- cQuery = null;
throw cqe;
}
@@ -308,6 +307,7 @@ public final class CqServiceImpl implements CqService {
return cQuery;
}
+ @Override
public void resumeCQ(int cqState, ServerCQ cQuery) {
// Initialize the state of CQ.
if (((CqStateImpl) cQuery.getState()).getState() != cqState) {
@@ -324,25 +324,10 @@ public final class CqServiceImpl implements CqService {
}
}
- /*
- * public void addToCqEventKeysMap(CqQuery cq){ if (cqToCqEventKeysMap != null) { synchronized
- * (cqToCqEventKeysMap){ String serverCqName = ((CqQueryImpl)cq).getServerCqName(); if
- * (!cqToCqEventKeysMap.containsKey(serverCqName)){ cqToCqEventKeysMap.put(serverCqName, new
- * HashSet()); if (_logger.isDebugEnabled()) {
- * _logger.debug("CQ Event key maintenance for CQ, CqName: " + serverCqName + " is Enabled." +
- * " key maintenance map size is: " + cqToCqEventKeysMap.size()); } } } // synchronized } }
- */
-
- public boolean hasCq() {
- HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
- return (cqMap.size() > 0);
- }
-
-
/**
* Adds the given CQ and cqQuery object into the CQ map.
*/
- public void addToCqMap(CqQueryImpl cq) throws CqExistsException, CqException {
+ void addToCqMap(CqQueryImpl cq) throws CqExistsException, CqException {
// On server side cqName will be server side cqName.
String sCqName = cq.getServerCqName();
if (logger.isDebugEnabled()) {
@@ -355,7 +340,7 @@ public final class CqServiceImpl implements CqService {
.toLocalizedString(sCqName));
}
synchronized (cqQueryMapLock) {
- HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<String, CqQueryImpl>(cqQueryMap);
+ HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<>(cqQueryMap);
try {
tmpCqQueryMap.put(sCqName, cq);
} catch (Exception ex) {
@@ -377,66 +362,34 @@ public final class CqServiceImpl implements CqService {
/**
* Removes given CQ from the cqMap..
*/
- public void removeCq(String cqName) {
+ void removeCq(String cqName) {
// On server side cqName will be server side cqName.
synchronized (cqQueryMapLock) {
- HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<String, CqQueryImpl>(cqQueryMap);
+ HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<>(cqQueryMap);
tmpCqQueryMap.remove(cqName);
this.cqNameToUserAttributesMap.remove(cqName);
cqQueryMap = tmpCqQueryMap;
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#getClientCqFromServer(org.apache.geode.
- * internal.cache.tier.sockets.ClientProxyMembershipID, java.lang.String)
- */
@Override
public CqQuery getClientCqFromServer(ClientProxyMembershipID clientProxyId, String clientCqName) {
// On server side cqName will be server side cqName.
HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
- return (CqQuery) cqMap.get(this.constructServerCqName(clientCqName, clientProxyId));
+ return cqMap.get(this.constructServerCqName(clientCqName, clientProxyId));
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getCq(java.lang.String)
- */
@Override
public InternalCqQuery getCq(String cqName) {
// On server side cqName will be server side cqName.
- return (InternalCqQuery) cqQueryMap.get(cqName);
+ return cqQueryMap.get(cqName);
}
- /**
- * Clears the CQ Query Map.
- */
- public void clearCqQueryMap() {
- // On server side cqName will be server side cqName.
- synchronized (cqQueryMapLock) {
- cqQueryMap = new HashMap<String, CqQueryImpl>();
- }
- }
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getAllCqs()
- */
@Override
public Collection<? extends InternalCqQuery> getAllCqs() {
return cqQueryMap.values();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getAllCqs(java.lang.String)
- */
@Override
public Collection<? extends InternalCqQuery> getAllCqs(final String regionName)
throws CqException {
@@ -445,7 +398,7 @@ public final class CqServiceImpl implements CqService {
LocalizedStrings.CqService_NULL_ARGUMENT_0.toLocalizedString("regionName"));
}
- String[] cqNames = null;
+ String[] cqNames;
synchronized (this.baseRegionToCqNameMap) {
ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName);
@@ -456,7 +409,7 @@ public final class CqServiceImpl implements CqService {
cqs.toArray(cqNames);
}
- ArrayList<InternalCqQuery> cQueryList = new ArrayList<InternalCqQuery>();
+ ArrayList<InternalCqQuery> cQueryList = new ArrayList<>();
for (int cqCnt = 0; cqCnt < cqNames.length; cqCnt++) {
InternalCqQuery cq = getCq(cqNames[cqCnt]);
if (cq != null) {
@@ -467,34 +420,16 @@ public final class CqServiceImpl implements CqService {
return cQueryList;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#executeAllClientCqs()
- */
@Override
public synchronized void executeAllClientCqs() throws CqException {
executeCqs(this.getAllCqs());
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#executeAllRegionCqs(java.lang.String)
- */
@Override
public synchronized void executeAllRegionCqs(final String regionName) throws CqException {
executeCqs(getAllCqs(regionName));
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#executeCqs(org.apache.geode.cache.query
- * .CqQuery[])
- */
@Override
public synchronized void executeCqs(Collection<? extends InternalCqQuery> cqs)
throws CqException {
@@ -503,53 +438,31 @@ public final class CqServiceImpl implements CqService {
}
String cqName = null;
for (InternalCqQuery internalCq : cqs) {
- CqQuery cq = (CqQuery) internalCq;
+ CqQuery cq = internalCq;
if (!cq.isClosed() && cq.isStopped()) {
try {
cqName = cq.getName();
cq.execute();
- } catch (QueryException qe) {
- if (logger.isDebugEnabled()) {
- logger.debug("Failed to execute the CQ, CqName : {} Error : {}", cqName,
- qe.getMessage());
- }
- } catch (CqClosedException cce) {
+ } catch (QueryException | CqClosedException e) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to execute the CQ, CqName : {} Error : {}", cqName,
- cce.getMessage());
+ e.getMessage());
}
}
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#stopAllClientCqs()
- */
@Override
public synchronized void stopAllClientCqs() throws CqException {
stopCqs(this.getAllCqs());
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#stopAllRegionCqs(java.lang.String)
- */
@Override
public synchronized void stopAllRegionCqs(final String regionName) throws CqException {
stopCqs(this.getAllCqs(regionName));
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#stopCqs(org.apache.geode.cache.query.
- * CqQuery[])
- */
@Override
public synchronized void stopCqs(Collection<? extends InternalCqQuery> cqs) throws CqException {
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -567,29 +480,20 @@ public final class CqServiceImpl implements CqService {
String cqName = null;
for (InternalCqQuery internalCqQuery : cqs) {
- CqQuery cq = (CqQuery) internalCqQuery;
+ CqQuery cq = internalCqQuery;
if (!cq.isClosed() && cq.isRunning()) {
try {
cqName = cq.getName();
cq.stop();
- } catch (QueryException qe) {
- if (isDebugEnabled) {
- logger.debug("Failed to stop the CQ, CqName : {} Error : {}", cqName, qe.getMessage());
- }
- } catch (CqClosedException cce) {
+ } catch (QueryException | CqClosedException e) {
if (isDebugEnabled) {
- logger.debug("Failed to stop the CQ, CqName : {} Error : {}", cqName, cce.getMessage());
+ logger.debug("Failed to stop the CQ, CqName : {} Error : {}", cqName, e.getMessage());
}
}
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeCqs(java.lang.String)
- */
@Override
public void closeCqs(final String regionName) throws CqException {
Collection<? extends InternalCqQuery> cqs = this.getAllCqs(regionName);
@@ -603,8 +507,8 @@ public final class CqServiceImpl implements CqService {
// invoked on the server
cq.close(false);
} else {
- // @todo grid: if regionName has a pool check its keepAlive
- boolean keepAlive = ((GemFireCacheImpl) this.cache).keepDurableSubscriptionsAlive();
+ // TODO: grid: if regionName has a pool check its keepAlive
+ boolean keepAlive = this.cache.keepDurableSubscriptionsAlive();
if (cq.isDurable() && keepAlive) {
logger.warn(LocalizedMessage.create(
LocalizedStrings.CqService_NOT_SENDING_CQ_CLOSE_TO_THE_SERVER_AS_IT_IS_A_DURABLE_CQ));
@@ -614,14 +518,9 @@ public final class CqServiceImpl implements CqService {
}
}
- } catch (QueryException qe) {
+ } catch (QueryException | CqClosedException e) {
if (logger.isDebugEnabled()) {
- logger.debug("Failed to close the CQ, CqName : {} Error : {}", cqName, qe.getMessage());
- }
- } catch (CqClosedException cce) {
- if (logger.isDebugEnabled()) {
- logger.debug("Failed to close the CQ, CqName : {} Error : {}", cqName,
- cce.getMessage());
+ logger.debug("Failed to close the CQ, CqName : {} Error : {}", cqName, e.getMessage());
}
}
}
@@ -630,10 +529,6 @@ public final class CqServiceImpl implements CqService {
/**
* Called directly on server side.
- *
- * @param cqName
- * @param clientId
- * @throws CqException
*/
@Override
public void stopCq(String cqName, ClientProxyMembershipID clientId) throws CqException {
@@ -650,8 +545,6 @@ public final class CqServiceImpl implements CqService {
try {
HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
if (!cqMap.containsKey(serverCqName)) {
- // throw new
- // CqException(LocalizedStrings.CqService_CQ_NOT_FOUND_FAILED_TO_STOP_THE_SPECIFIED_CQ_0.toLocalizedString(serverCqName));
/*
* gregp 052808: We should silently fail here instead of throwing error. This is to deal
* with races in recovery
@@ -689,15 +582,8 @@ public final class CqServiceImpl implements CqService {
}
// Send stop message to peers.
cQuery.getCqBaseRegion().getFilterProfile().stopCq(cQuery);
-
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeCq(java.lang.String,
- * org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID)
- */
@Override
public void closeCq(String cqName, ClientProxyMembershipID clientProxyId) throws CqException {
String serverCqName = cqName;
@@ -713,9 +599,6 @@ public final class CqServiceImpl implements CqService {
try {
HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
if (!cqMap.containsKey(serverCqName)) {
- // throw new
- // CqException(LocalizedStrings.CqService_CQ_NOT_FOUND_FAILED_TO_CLOSE_THE_SPECIFIED_CQ_0
- // .toLocalizedString(serverCqName));
/*
* gregp 052808: We should silently fail here instead of throwing error. This is to deal
* with races in recovery
@@ -791,12 +674,6 @@ public final class CqServiceImpl implements CqService {
}
}
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeAllCqs(boolean)
- */
@Override
public void closeAllCqs(boolean clientInitiated) {
closeAllCqs(clientInitiated, getAllCqs());
@@ -807,21 +684,13 @@ public final class CqServiceImpl implements CqService {
* CqQuerys created by other VMs are unaffected.
*/
private void closeAllCqs(boolean clientInitiated, Collection<? extends InternalCqQuery> cqs) {
- closeAllCqs(clientInitiated, cqs,
- ((GemFireCacheImpl) this.cache).keepDurableSubscriptionsAlive());
+ closeAllCqs(clientInitiated, cqs, this.cache.keepDurableSubscriptionsAlive());
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeAllCqs(boolean,
- * org.apache.geode.cache.query.CqQuery[], boolean)
- */
@Override
public void closeAllCqs(boolean clientInitiated, Collection<? extends InternalCqQuery> cqs,
boolean keepAlive) {
- // CqQuery[] cqs = getAllCqs();
if (cqs != null) {
String cqName = null;
if (logger.isDebugEnabled()) {
@@ -830,7 +699,6 @@ public final class CqServiceImpl implements CqService {
for (InternalCqQuery cQuery : cqs) {
try {
cqName = cQuery.getName();
- // boolean keepAlive = ((GemFireCache)this.cache).keepDurableSubscriptionsAlive();
if (isServer()) {
cQuery.close(false);
@@ -847,47 +715,26 @@ public final class CqServiceImpl implements CqService {
}
}
}
- } catch (QueryException cqe) {
+ } catch (QueryException | CqClosedException e) {
if (!isRunning()) {
// Not cache shutdown
logger
.warn(LocalizedMessage.create(LocalizedStrings.CqService_FAILED_TO_CLOSE_CQ__0___1,
- new Object[] {cqName, cqe.getMessage()}));
+ new Object[] {cqName, e.getMessage()}));
}
if (logger.isDebugEnabled()) {
- logger.debug(cqe.getMessage(), cqe);
- }
- } catch (CqClosedException cqe) {
- if (!isRunning()) {
- // Not cache shutdown
- logger
- .warn(LocalizedMessage.create(LocalizedStrings.CqService_FAILED_TO_CLOSE_CQ__0___1,
- new Object[] {cqName, cqe.getMessage()}));
- }
- if (logger.isDebugEnabled()) {
- logger.debug(cqe.getMessage(), cqe);
+ logger.debug(e.getMessage(), e);
}
}
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getCqStatistics()
- */
@Override
public CqServiceStatistics getCqStatistics() {
return cqServiceStats;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeClientCqs(org.apache.geode.
- * internal.cache.tier.sockets.ClientProxyMembershipID)
- */
@Override
public void closeClientCqs(ClientProxyMembershipID clientProxyId) throws CqException {
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -899,30 +746,19 @@ public final class CqServiceImpl implements CqService {
CqQueryImpl cQuery = (CqQueryImpl) cq;
try {
cQuery.close(false);
- } catch (QueryException qe) {
+ } catch (QueryException | CqClosedException e) {
if (isDebugEnabled) {
logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
- qe.getMessage());
- }
- } catch (CqClosedException cce) {
- if (isDebugEnabled) {
- logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
- cce.getMessage());
+ e.getMessage());
}
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getAllClientCqs(org.apache.geode.
- * internal.cache.tier.sockets.ClientProxyMembershipID)
- */
@Override
public List<ServerCQ> getAllClientCqs(ClientProxyMembershipID clientProxyId) {
Collection<? extends InternalCqQuery> cqs = getAllCqs();
- ArrayList<ServerCQ> clientCqs = new ArrayList<ServerCQ>();
+ ArrayList<ServerCQ> clientCqs = new ArrayList<>();
for (InternalCqQuery cq : cqs) {
ServerCQImpl cQuery = (ServerCQImpl) cq;
@@ -934,23 +770,16 @@ public final class CqServiceImpl implements CqService {
return clientCqs;
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#getAllDurableClientCqs(org.apache.geode
- * .internal.cache.tier.sockets.ClientProxyMembershipID)
- */
@Override
public List<String> getAllDurableClientCqs(ClientProxyMembershipID clientProxyId)
throws CqException {
if (clientProxyId == null) {
throw new CqException(
LocalizedStrings.CqService_UNABLE_TO_RETRIEVE_DURABLE_CQS_FOR_CLIENT_PROXY_ID
- .toLocalizedString(clientProxyId));
+ .toLocalizedString(null));
}
List<ServerCQ> cqs = getAllClientCqs(clientProxyId);
- ArrayList<String> durableClientCqs = new ArrayList<String>();
+ ArrayList<String> durableClientCqs = new ArrayList<>();
for (ServerCQ cq : cqs) {
ServerCQImpl cQuery = (ServerCQImpl) cq;
@@ -966,9 +795,6 @@ public final class CqServiceImpl implements CqService {
/**
* Server side method. Closes non-durable CQs for the given client proxy id.
- *
- * @param clientProxyId
- * @throws CqException
*/
@Override
public void closeNonDurableClientCqs(ClientProxyMembershipID clientProxyId) throws CqException {
@@ -983,15 +809,10 @@ public final class CqServiceImpl implements CqService {
if (!cQuery.isDurable()) {
cQuery.close(false);
}
- } catch (QueryException qe) {
- if (isDebugEnabled) {
- logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
- qe.getMessage());
- }
- } catch (CqClosedException cce) {
+ } catch (QueryException | CqClosedException e) {
if (isDebugEnabled) {
logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
- cce.getMessage());
+ e.getMessage());
}
}
}
@@ -1028,6 +849,7 @@ public final class CqServiceImpl implements CqService {
return this.isRunning;
}
+ @Override
public void start() {
this.isRunning = true;
}
@@ -1035,9 +857,10 @@ public final class CqServiceImpl implements CqService {
/**
* @return Returns the serverCqName.
*/
+ @Override
public String constructServerCqName(String cqName, ClientProxyMembershipID clientProxyId) {
- ConcurrentHashMap<ClientProxyMembershipID, String> cache = serverCqNameCache
- .computeIfAbsent(cqName, key -> new ConcurrentHashMap<ClientProxyMembershipID, String>());
+ ConcurrentHashMap<ClientProxyMembershipID, String> cache =
+ serverCqNameCache.computeIfAbsent(cqName, key -> new ConcurrentHashMap<>());
String cName = cache.get(clientProxyId);
if (null == cName) {
@@ -1065,7 +888,7 @@ public final class CqServiceImpl implements CqService {
}
}
- /*
+ /**
* Checks if CQ with the given name already exists.
*
* @param cqName name of the CQ.
@@ -1073,17 +896,15 @@ public final class CqServiceImpl implements CqService {
* @return true if exists else false.
*/
private synchronized boolean isCqExists(String cqName) {
- boolean status = false;
HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
- status = cqMap.containsKey(cqName);
- return status;
+ return cqMap.containsKey(cqName);
}
- /*
+ /**
* Generates a name for CQ. Checks if CQ with that name already exists if so generates a new
* cqName.
*/
- public synchronized String generateCqName() {
+ private synchronized String generateCqName() {
while (true) {
String cqName = CQ_NAME_PREFIX + (cqId++);
if (!isCqExists(cqName)) {
@@ -1092,18 +913,9 @@ public final class CqServiceImpl implements CqService {
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#dispatchCqListeners(java.util.HashMap,
- * int, java.lang.Object, java.lang.Object, byte[],
- * org.apache.geode.cache.client.internal.QueueManager, org.apache.geode.internal.cache.EventID)
- */
@Override
public void dispatchCqListeners(HashMap<String, Integer> cqs, int messageType, Object key,
Object value, byte[] delta, QueueManager qManager, EventID eventId) {
- ClientCQImpl cQuery = null;
Object[] fullValue = new Object[1];
Iterator<Map.Entry<String, Integer>> iter = cqs.entrySet().iterator();
String cqName = null;
@@ -1112,7 +924,7 @@ public final class CqServiceImpl implements CqService {
try {
Map.Entry<String, Integer> entry = iter.next();
cqName = entry.getKey();
- cQuery = (ClientCQImpl) this.getCq(cqName);
+ ClientCQImpl cQuery = (ClientCQImpl) this.getCq(cqName);
if (cQuery == null || (!cQuery.isRunning() && cQuery.getQueuedEvents() == null)) {
if (isDebugEnabled) {
@@ -1122,7 +934,7 @@ public final class CqServiceImpl implements CqService {
continue;
}
- Integer cqOp = (Integer) entry.getValue();
+ Integer cqOp = entry.getValue();
// If Region destroy event, close the cq.
if (cqOp.intValue() == MessageType.DESTROY_REGION) {
@@ -1136,8 +948,7 @@ public final class CqServiceImpl implements CqService {
}
// Construct CqEvent.
- CqEventImpl cqEvent = null;
- cqEvent = new CqEventImpl(cQuery, getOperation(messageType), getOperation(cqOp.intValue()),
+ CqEventImpl cqEvent = new CqEventImpl(cQuery, getOperation(messageType), getOperation(cqOp),
key, value, delta, qManager, eventId);
// Update statistics
@@ -1181,11 +992,11 @@ public final class CqServiceImpl implements CqService {
} // iteration.
}
- public void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent) {
+ void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent) {
invokeListeners(cqName, cQuery, cqEvent, null);
}
- public void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent,
+ private void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent,
Object[] fullValue) {
if (!cQuery.isRunning() || cQuery.getCqAttributes() == null) {
return;
@@ -1217,8 +1028,8 @@ public final class CqServiceImpl implements CqService {
}
Part result = (Part) GetEventValueOp
.executeOnPrimary(cqEvent.getQueueManager().getPool(), cqEvent.getEventID(), null);
- Object newVal = null;
- if (result == null || (newVal = result.getObject()) == null) {
+ Object newVal = result.getObject();
+ if (result == null || newVal == null) {
if (!cache.getCancelCriterion().isCancelInProgress()) {
Exception ex =
new Exception("Failed to retrieve full value from server for eventID "
@@ -1231,7 +1042,7 @@ public final class CqServiceImpl implements CqService {
}
}
} else {
- ((GemFireCacheImpl) this.cache).getCachePerfStats().incDeltaFullValuesRequested();
+ this.cache.getCachePerfStats().incDeltaFullValuesRequested();
cqEvent = new CqEventImpl(cQuery, cqEvent.getBaseOperation(),
cqEvent.getQueryOperation(), cqEvent.getKey(), newVal, cqEvent.getDeltaValue(),
cqEvent.getQueueManager(), cqEvent.getEventID());
@@ -1278,7 +1089,7 @@ public final class CqServiceImpl implements CqService {
}
}
- public void invokeCqConnectedListeners(String cqName, ClientCQImpl cQuery, boolean connected) {
+ private void invokeCqConnectedListeners(String cqName, ClientCQImpl cQuery, boolean connected) {
if (!cQuery.isRunning() || cQuery.getCqAttributes() == null) {
return;
}
@@ -1335,12 +1146,8 @@ public final class CqServiceImpl implements CqService {
}
}
-
/**
* Returns the Operation for the given EnumListenerEvent type.
- *
- * @param eventType
- * @return Operation
*/
private Operation getOperation(int eventType) {
Operation op = null;
@@ -1372,15 +1179,6 @@ public final class CqServiceImpl implements CqService {
return op;
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#processEvents(org.apache.geode.cache.
- * CacheEvent, org.apache.geode.distributed.internal.DistributionAdvisor.Profile,
- * org.apache.geode.distributed.internal.DistributionAdvisor.Profile[],
- * org.apache.geode.internal.cache.FilterRoutingInfo)
- */
@Override
public void processEvents(CacheEvent event, Profile localProfile, Profile[] profiles,
FilterRoutingInfo frInfo) throws CqException {
@@ -1421,7 +1219,7 @@ public final class CqServiceImpl implements CqService {
continue;
}
Map cqs = pf.getCqMap();
- HashMap<Long, Integer> cqInfo = new HashMap<Long, Integer>();
+ HashMap<Long, Integer> cqInfo = new HashMap<>();
Iterator cqIter = cqs.entrySet().iterator();
while (cqIter.hasNext()) {
Map.Entry cqEntry = (Map.Entry) cqIter.next();
@@ -1454,10 +1252,10 @@ public final class CqServiceImpl implements CqService {
private void processEntryEvent(CacheEvent event, Profile localProfile, Profile[] profiles,
FilterRoutingInfo frInfo) throws CqException {
final boolean isDebugEnabled = logger.isDebugEnabled();
- HashSet<Object> cqUnfilteredEventsSet_newValue = new HashSet<Object>();
- HashSet<Object> cqUnfilteredEventsSet_oldValue = new HashSet<Object>();
- boolean b_cqResults_newValue = false;
- boolean b_cqResults_oldValue = false;
+ HashSet<Object> cqUnfilteredEventsSet_newValue = new HashSet<>();
+ HashSet<Object> cqUnfilteredEventsSet_oldValue = new HashSet<>();
+ boolean b_cqResults_newValue;
+ boolean b_cqResults_oldValue;
boolean queryOldValue;
EntryEvent entryEvent = (EntryEvent) event;
Object eventKey = entryEvent.getKey();
@@ -1472,8 +1270,8 @@ public final class CqServiceImpl implements CqService {
|| event.getOperation().isDestroy() || event.getOperation().isInvalidate()
|| (event.getOperation().isCreate() && isDupEvent));
- HashMap<String, Integer> matchedCqs = new HashMap<String, Integer>();
- long executionStartTime = 0;
+ HashMap<String, Integer> matchedCqs = new HashMap<>();
+ long executionStartTime;
for (int i = -1; i < profiles.length; i++) {
CacheProfile cf;
if (i < 0) {
@@ -1498,7 +1296,6 @@ public final class CqServiceImpl implements CqService {
continue;
}
-
// Get new value. If its not retrieved.
if (cqUnfilteredEventsSet_newValue.isEmpty()
&& (event.getOperation().isCreate() || event.getOperation().isUpdate())) {
@@ -1509,7 +1306,7 @@ public final class CqServiceImpl implements CqService {
}
}
- HashMap<Long, Integer> cqInfo = new HashMap<Long, Integer>();
+ HashMap<Long, Integer> cqInfo = new HashMap<>();
Iterator cqIter = cqs.entrySet().iterator();
while (cqIter.hasNext()) {
@@ -1546,7 +1343,6 @@ public final class CqServiceImpl implements CqService {
}
} else {
boolean error = false;
- // synchronized (cQuery)
{
try {
synchronized (cQuery) {
@@ -1644,7 +1440,7 @@ public final class CqServiceImpl implements CqService {
cQuery.markAsDestroyedInCqResultKeys(eventKey);
}
}
- } // end synchronized(cQuery)
+ }
// Get the matching CQs if any.
// synchronized (this.matchingCqMap){
@@ -1663,7 +1459,6 @@ public final class CqServiceImpl implements CqService {
}
}
}
- // }
}
if (cqEvent != null && cQuery.isRunning()) {
@@ -1694,153 +1489,35 @@ public final class CqServiceImpl implements CqService {
} // iteration over Profiles.
}
-
- /*
- * public void processEvents (EnumListenerEvent operation, CacheEvent event, ClientUpdateMessage
- * clientMessage, CM<ClientProxyMembershipID, CM<CqQuery, Boolean>> clientIds) throws CqException
- * {
- *
- * //Is this a region event or an entry event if (event instanceof RegionEvent){
- * processRegionEvent(operation, event, clientMessage, clientIds); } else { processEntryEvent
- * (operation, event, clientMessage, clientIds); }
- *
- * }
- *
- * private void processRegionEvent(EnumListenerEvent operation, CacheEvent event,
- * ClientUpdateMessage clientMessage, CM<ClientProxyMembershipID, CM<CqQuery, Boolean>> clientIds)
- * throws CqException {
- *
- * if (logger.isDebugEnabled()) { logger.debug("Processing region event for region " +
- * ((LocalRegion)(event.getRegion())).getName()); } HashMap filteredCqs = new HashMap(); Integer
- * cqRegionEvent = generateCqRegionEvent(operation); Iterator it =
- * clientIds.entrySet().iterator(); while (it.hasNext()) { Map.Entry me = (Map.Entry)it.next();
- * ClientProxyMembershipID clientId = (ClientProxyMembershipID)me.getKey(); CM cqsToBooleans =
- * (CM)me.getValue(); if (cqsToBooleans == null) { continue; } Set<CqQuery> cqs =
- * cqsToBooleans.keySet(); if (cqs.isEmpty()) { continue; } filteredCqs.clear(); Iterator cqIt =
- * cqs.iterator(); while (cqIt.hasNext()) { CqQueryImpl cQuery = (CqQueryImpl)cqIt.next(); if
- * (operation == EnumListenerEvent.AFTER_REGION_DESTROY) { try { if (logger.isDebugEnabled()){
- * logger.debug("Closing CQ on region destroy event. CqName :" + cQuery.getName()); }
- * cQuery.close(false); } catch (Exception ex) {
- * logger.debug("Failed to Close CQ on region destroy. CqName :" + cQuery.getName(), ex); }
- *
- * } filteredCqs.put(cQuery.cqName, cqRegionEvent);
- * cQuery.getVsdStats().updateStats(cqRegionEvent);
- *
- * } if (!filteredCqs.isEmpty()){ ((ClientUpdateMessageImpl)clientMessage).addClientCqs( clientId,
- * filteredCqs); }
- *
- * }
- *
- * }
- *
- * private void processEntryEvent(EnumListenerEvent operation, CacheEvent event,
- * ClientUpdateMessage clientMessage, CM<ClientProxyMembershipID, CM<CqQuery, Boolean>> clientIds)
- * throws CqException { HashSet cqUnfilteredEventsSet_newValue = new HashSet(); HashSet
- * cqUnfilteredEventsSet_oldValue = new HashSet(); boolean b_cqResults_newValue = false; boolean
- * b_cqResults_oldValue = false; EntryEvent entryEvent = (EntryEvent)event; Object eventKey =
- * entryEvent.getKey(); if (operation == EnumListenerEvent.AFTER_CREATE || operation ==
- * EnumListenerEvent.AFTER_UPDATE) { if (entryEvent.getNewValue() != null) { //We have a new value
- * to run the query on cqUnfilteredEventsSet_newValue.clear();
- * cqUnfilteredEventsSet_newValue.add(entryEvent.getNewValue()); } }
- *
- * HashMap matchedCqs = new HashMap(); long executionStartTime = 0; Iterator it =
- * clientIds.entrySet().iterator(); while (it.hasNext()) { Map.Entry me = (Map.Entry)it.next();
- * ClientProxyMembershipID clientId = (ClientProxyMembershipID)me.getKey(); if
- * (logger.isDebugEnabled()) { logger.debug("Processing event for CQ filter, ClientId : " +
- * clientId); } CM cqsToBooleans = (CM)me.getValue(); if (cqsToBooleans == null) { continue; }
- * Set<CqQuery> cqs = cqsToBooleans.keySet(); if (cqs.isEmpty()) { continue; } HashMap filteredCqs
- * = new HashMap(); Iterator cqIt = cqs.iterator(); while (cqIt.hasNext()) { CqQueryImpl cQuery =
- * (CqQueryImpl)cqIt.next(); b_cqResults_newValue = false; b_cqResults_oldValue = false; if
- * (cQuery == null || !(cQuery.isRunning())){ continue; } String cqName =
- * cQuery.getServerCqName(); Integer cqEvent = null; if (matchedCqs.containsKey(cqName)) { if
- * (logger.isDebugEnabled()){ logger.
- * debug("Similar cq/query is already processed, getting the cq event-type from the matched cq.");
- * } cqEvent = (Integer)matchedCqs.get(cqName); } else { boolean error = false; boolean
- * hasSeenEvent = false; HashSet cqEventKeys = null; synchronized (cQuery) { try { // Apply query
- * on new value. if (!cqUnfilteredEventsSet_newValue.isEmpty()) { executionStartTime =
- * this.stats.startCqQueryExecution(); b_cqResults_newValue = evaluateQuery(cQuery, new Object[]
- * {cqUnfilteredEventsSet_newValue}); this.stats.endCqQueryExecution(executionStartTime); } //
- * Check if old value is cached, if not apply query on old value. if (cqToCqEventKeysMap != null)
- * { synchronized (cqToCqEventKeysMap) { if ((cqEventKeys =
- * (HashSet)cqToCqEventKeysMap.get(cqName)) != null) { hasSeenEvent =
- * cqEventKeys.contains(eventKey); } } } if (!hasSeenEvent) { // get the oldValue. // In case of
- * Update, destroy and invalidate. if (operation == EnumListenerEvent.AFTER_UPDATE || operation ==
- * EnumListenerEvent.AFTER_DESTROY || operation == EnumListenerEvent.AFTER_INVALIDATE) { if
- * (entryEvent.getOldValue() != null) { cqUnfilteredEventsSet_oldValue.clear();
- * cqUnfilteredEventsSet_oldValue.add(entryEvent.getOldValue()); // Apply query on old value.
- * executionStartTime = this.stats.startCqQueryExecution(); b_cqResults_oldValue =
- * evaluateQuery(cQuery, new Object[] {cqUnfilteredEventsSet_oldValue});
- * this.stats.endCqQueryExecution(executionStartTime); } } } } catch (Exception ex) { //Any
- * exception in running the query // should be caught here and buried //because this code is
- * running inline with the //message processing code and we don't want to //kill that thread error
- * = true; logger.info( LocalizedStrings.
- * CqService_ERROR_WHILE_PROCESSING_CQ_ON_THE_EVENT_KEY_0_CQNAME_1_CLIENTID_2_ERROR_3, new
- * Object[] { ((EntryEvent)event).getKey(), cQuery.getName(), clientId,
- * ex.getLocalizedMessage()}); }
- *
- * if (error) { cqEvent = Integer.valueOf(MessageType.EXCEPTION); } else { if
- * (b_cqResults_newValue) { if (hasSeenEvent || b_cqResults_oldValue) { cqEvent =
- * Integer.valueOf(MessageType.LOCAL_UPDATE); } else { cqEvent =
- * Integer.valueOf(MessageType.LOCAL_CREATE); } // If its create and caching is enabled, cache the
- * key for this CQ. if (!hasSeenEvent && cqEventKeys != null) { cqEventKeys.add(eventKey); } }
- * else if (hasSeenEvent || (b_cqResults_oldValue)) { // Base invalidate operation is treated as
- * destroy. // When the invalidate comes through, the entry will no longer satisfy // the query
- * and will need to be deleted. cqEvent = Integer.valueOf(MessageType.LOCAL_DESTROY); // If
- * caching is enabled, remove this event's key from the cache. if (hasSeenEvent && cqEventKeys !=
- * null) { cqEventKeys.remove(eventKey); } } }
- *
- * } //end synchronized(cQuery)
- *
- * // Get the matching CQs if any. synchronized (this.matchingCqMap){ String query =
- * cQuery.getQueryString(); ArrayList matchingCqs = (ArrayList)matchingCqMap.get(query); if
- * (matchingCqs != null) { Iterator iter = matchingCqs.iterator(); while (iter.hasNext()) { String
- * matchingCqName = (String)iter.next(); if (!matchingCqName.equals(cqName)){
- * matchedCqs.put(matchingCqName, cqEvent); } } } }
- *
- * }
- *
- * if (cqEvent != null){ if (logger.isDebugEnabled()) {
- * logger.debug("Event is added for the CQ, CqName (clientside): " + cQuery.cqName +
- * " With CQ Op : " + cqEvent + " for Client : " + clientId); } filteredCqs.put(cQuery.cqName,
- * cqEvent); cQuery.getVsdStats().updateStats(cqEvent); }
- *
- * } // iteration over cqsToBooleans.keySet() if (!filteredCqs.isEmpty()){
- * logger.debug("Adding event map for client : "+clientId +
- * " with event map size : "+filteredCqs.size());
- * ((ClientUpdateMessageImpl)clientMessage).addClientCqs(clientId, filteredCqs); } } // iteration
- * over clientIds.entrySet() }
- */
-
private Integer generateCqRegionEvent(CacheEvent event) {
Integer cqEvent = null;
if (event.getOperation().isRegionDestroy()) {
- cqEvent = Integer.valueOf(MessageType.DESTROY_REGION);
+ cqEvent = MessageType.DESTROY_REGION;
} else if (event.getOperation().isRegionInvalidate()) {
- cqEvent = Integer.valueOf(MessageType.INVALIDATE_REGION);
+ cqEvent = MessageType.INVALIDATE_REGION;
} else if (event.getOperation().isClear()) {
- cqEvent = Integer.valueOf(MessageType.CLEAR_REGION);
+ cqEvent = MessageType.CLEAR_REGION;
}
return cqEvent;
}
-
/**
* Manages the CQs created for the base region. This is managed here, instead of on the base
* region; since the cq could be created on the base region, before base region is created (using
* newCq()).
*/
- public void addToBaseRegionToCqNameMap(String regionName, String cqName) {
+ private void addToBaseRegionToCqNameMap(String regionName, String cqName) {
synchronized (this.baseRegionToCqNameMap) {
ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName);
if (cqs == null) {
- cqs = new ArrayList<String>();
+ cqs = new ArrayList<>();
}
cqs.add(cqName);
this.baseRegionToCqNameMap.put(regionName, cqs);
}
}
- public void removeFromBaseRegionToCqNameMap(String regionName, String cqName) {
+ void removeFromBaseRegionToCqNameMap(String regionName, String cqName) {
synchronized (this.baseRegionToCqNameMap) {
ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName);
if (cqs != null) {
@@ -1864,37 +1541,12 @@ public final class CqServiceImpl implements CqService {
}
/**
- * Removes this CQ from CQ event Cache map. This disables the caching events for this CQ.
- *
- * @param cqName
- */
- /*
- * synchronized public void removeCQFromCaching(String cqName){ if (cqToCqEventKeysMap != null) {
- * // Take a lock on CqQuery object. In processEvents the maps are // handled under CqQuery
- * object. if (cqToCqEventKeysMap != null){ synchronized (cqToCqEventKeysMap) {
- * cqToCqEventKeysMap.remove(cqName); } } } }
- */
-
- /**
- * Returns the CQ event cache map.
- *
- * @return HashMap cqToCqEventKeysMap
- *
- * Caller must synchronize on the returned value in order to inspect.
- */
- /*
- * public HashMap getCqToCqEventKeysMap(){ return cqToCqEventKeysMap; }
- */
-
- /**
* Adds the query from the given CQ to the matched CQ map.
- *
- * @param cq
*/
- public void addToMatchingCqMap(CqQueryImpl cq) {
+ void addToMatchingCqMap(CqQueryImpl cq) {
synchronized (this.matchingCqMap) {
String cqQuery = cq.getQueryString();
- Set<String> matchingCQs = null;
+ Set<String> matchingCQs;
if (!matchingCqMap.containsKey(cqQuery)) {
matchingCQs = Collections.newSetFromMap(new ConcurrentHashMap());
matchingCqMap.put(cqQuery, matchingCQs);
@@ -1912,10 +1564,8 @@ public final class CqServiceImpl implements CqService {
/**
* Removes the query from the given CQ from the matched CQ map.
- *
- * @param cq
*/
- public void removeFromMatchingCqMap(CqQueryImpl cq) {
+ private void removeFromMatchingCqMap(CqQueryImpl cq) {
synchronized (this.matchingCqMap) {
String cqQuery = cq.getQueryString();
if (matchingCqMap.containsKey(cqQuery)) {
@@ -1947,10 +1597,6 @@ public final class CqServiceImpl implements CqService {
* Applies the query on the event. This method takes care of the performance related changed done
* to improve the CQ-query performance. When CQ-query is executed first time, it saves the query
* related information in the execution context and uses that info in later executions.
- *
- * @param cQuery
- * @param event
- * @return boolean
*/
private boolean evaluateQuery(CqQueryImpl cQuery, Object[] event) throws Exception {
ExecutionContext execContext = cQuery.getQueryExecutionContext();
@@ -1983,19 +1629,6 @@ public final class CqServiceImpl implements CqService {
return this.cqNameToUserAttributesMap.get(cqName);
}
- // public static void memberLeft(String poolName) {
- // if (cqServiceSingleton != null && !cqServiceSingleton.isServer()) {
- // cqServiceSingleton.sendMemberDisconnectedMessageToCqs(poolName);
- // }
- // }
- //
- // public static void memberCrashed(String poolName) {
- // if (cqServiceSingleton != null && !cqServiceSingleton.isServer()) {
- // cqServiceSingleton.sendMemberDisconnectedMessageToCqs(poolName);
- // }
- // }
- //
-
@Override
public void cqsDisconnected(Pool pool) {
invokeCqsConnected(pool, false);
@@ -2014,7 +1647,7 @@ public final class CqServiceImpl implements CqService {
// Check to see if we are already connected/disconnected.
// If state has not changed, do not invoke another connected/disconnected
synchronized (cqPoolsConnected) {
- // don't repeatily send same connect/disconnect message to cq's on repeated fails of
+ // don't repeatedly send same connect/disconnect message to cq's on repeated fails of
// RedundancySatisfier
if (cqPoolsConnected.containsKey(poolName) && connected == cqPoolsConnected.get(poolName)) {
return;
@@ -2059,13 +1692,6 @@ public final class CqServiceImpl implements CqService {
SystemFailure.checkFailure();
logger.warn(LocalizedMessage
.create(LocalizedStrings.CqService_ERROR_SENDING_CQ_CONNECTION_STATUS, cqName), t);
-
- if (t instanceof VirtualMachineError) {
- logger.warn(LocalizedMessage.create(
- LocalizedStrings.CqService_VIRTUALMACHINEERROR_PROCESSING_CQLISTENER_FOR_CQ_0,
- cqName), t);
- return;
- }
}
}
}
@@ -2075,7 +1701,4 @@ public final class CqServiceImpl implements CqService {
public List<String> getAllDurableCqsFromServer(InternalPool pool) {
return new ServerCQProxyImpl(pool).getAllDurableCqsFromServer();
}
-
-
}
-
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java
index ba71143..a675162 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java
@@ -14,11 +14,9 @@
*/
package org.apache.geode.cache.query.internal.cq;
-import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.query.CqServiceStatistics;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.internal.DefaultQueryService;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
/**
* Provides statistical information about CqService.
@@ -26,24 +24,22 @@ import org.apache.geode.internal.cache.GemFireCacheImpl;
* @since GemFire 5.5
*/
public class CqServiceStatisticsImpl implements CqServiceStatistics {
+
private CqServiceImpl cqService;
- // private long activeCqs;
- // private long stoppedCqs;
- // private long closedCqs;
- // private long createdCqs;
/**
* Constructor for CqStatisticsImpl
*
* @param cqs - CqService
*/
- public CqServiceStatisticsImpl(CqServiceImpl cqs) {
+ CqServiceStatisticsImpl(CqServiceImpl cqs) {
cqService = cqs;
}
/**
* Returns the number of CQs currently executing
*/
+ @Override
public long numCqsActive() {
return this.cqService.getCqServiceVsdStats().getNumCqsActive();
}
@@ -53,6 +49,7 @@ public class CqServiceStatisticsImpl implements CqServiceStatistics {
*
* @return long number of cqs created.
*/
+ @Override
public long numCqsCreated() {
return this.cqService.getCqServiceVsdStats().getNumCqsCreated();
}
@@ -60,6 +57,7 @@ public class CqServiceStatisticsImpl implements CqServiceStatistics {
/**
* Returns number of Cqs that are closed.
*/
+ @Override
public long numCqsClosed() {
return this.cqService.getCqServiceVsdStats().getNumCqsClosed();
}
@@ -67,6 +65,7 @@ public class CqServiceStatisticsImpl implements CqServiceStatistics {
/**
* Returns number of Cqs that are stopped.
*/
+ @Override
public long numCqsStopped() {
return this.cqService.getCqServiceVsdStats().getNumCqsStopped();
}
@@ -74,20 +73,18 @@ public class CqServiceStatisticsImpl implements CqServiceStatistics {
/**
* Returns number of CQs created from the client.
*/
+ @Override
public long numCqsOnClient() {
return this.cqService.getCqServiceVsdStats().getNumCqsOnClient();
}
/**
* Returns the number of CQs (active + suspended) on the given region.
- *
- * @param regionName
*/
+ @Override
public long numCqsOnRegion(String regionName) {
-
DefaultQueryService queryService =
- (DefaultQueryService) ((GemFireCacheImpl) CacheFactory.getAnyInstance())
- .getLocalQueryService();
+ (DefaultQueryService) cqService.getInternalCache().getLocalQueryService();
try {
CqQuery[] cqs = queryService.getCqs(regionName);
[2/4] geode git commit: GEODE-2806: if the batch is dispatched,
even the bucket is no longer primary,
the batch should still be deleted as planned.
Posted by kl...@apache.org.
GEODE-2806: if the batch is dispatched, even the bucket is no longer primary, the batch should still be deleted as planned.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/0862174c
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/0862174c
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/0862174c
Branch: refs/heads/feature/GEODE-2632-5
Commit: 0862174c30cad1536f2c105b783653bd0d4344e8
Parents: 50686b0
Author: zhouxh <gz...@pivotal.io>
Authored: Fri Apr 21 09:57:05 2017 -0700
Committer: zhouxh <gz...@pivotal.io>
Committed: Fri Apr 21 09:58:19 2017 -0700
----------------------------------------------------------------------
.../parallel/ParallelGatewaySenderQueue.java | 58 ++++++++++----------
1 file changed, 28 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/0862174c/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
index cf4c5a9..9696b90 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
@@ -1104,38 +1104,36 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
private void destroyEventFromQueue(PartitionedRegion prQ, int bucketId, Object key) {
boolean isPrimary = prQ.getRegionAdvisor().getBucketAdvisor(bucketId).isPrimary();
- if (isPrimary) {
- BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId);
- // TODO : Kishor : Make sure we dont need to initalize a bucket
- // before destroying a key from it
- try {
- if (brq != null) {
- brq.destroyKey(key);
- }
- stats.decQueueSize();
- } catch (EntryNotFoundException e) {
- if (!this.sender.isBatchConflationEnabled() && logger.isDebugEnabled()) {
- logger.debug(
- "ParallelGatewaySenderQueue#remove: Got EntryNotFoundException while removing key {} for {} for bucket = {} for GatewaySender {}",
- key, this, bucketId, this.sender);
- }
- } catch (ForceReattemptException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("Bucket :{} moved to other member", bucketId);
- }
- } catch (PrimaryBucketException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("Primary bucket :{} moved to other member", bucketId);
- }
- } catch (RegionDestroyedException e) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Caught RegionDestroyedException attempting to remove key {} from bucket {} in {}",
- key, bucketId, prQ.getFullPath());
- }
+ BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId);
+ // TODO : Kishor : Make sure we dont need to initalize a bucket
+ // before destroying a key from it
+ try {
+ if (brq != null) {
+ brq.destroyKey(key);
+ }
+ stats.decQueueSize();
+ } catch (EntryNotFoundException e) {
+ if (!this.sender.isBatchConflationEnabled() && logger.isDebugEnabled()) {
+ logger.debug(
+ "ParallelGatewaySenderQueue#remove: Got EntryNotFoundException while removing key {} for {} for bucket = {} for GatewaySender {}",
+ key, this, bucketId, this.sender);
+ }
+ } catch (ForceReattemptException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Bucket :{} moved to other member", bucketId);
+ }
+ } catch (PrimaryBucketException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Primary bucket :{} moved to other member", bucketId);
+ }
+ } catch (RegionDestroyedException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Caught RegionDestroyedException attempting to remove key {} from bucket {} in {}", key,
+ bucketId, prQ.getFullPath());
}
- addRemovedEvent(prQ, bucketId, key);
}
+ addRemovedEvent(prQ, bucketId, key);
}
public void resetLastPeeked() {
[3/4] geode git commit: 2632: refactor code to use InternalCache
instead of GemFireCacheImpl
Posted by kl...@apache.org.
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java
index 8435c4c..5dc7bb0 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java
@@ -25,9 +25,9 @@ import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.internal.NanoTimer;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.statistics.StatisticsTypeFactoryImpl;
import org.apache.geode.internal.cache.FilterProfile;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.logging.LogService;
/**
@@ -44,35 +44,34 @@ public class CqServiceVsdStats {
private static final StatisticsType _type;
/** Name of the created CQs statistic */
- protected static final String CQS_CREATED = "numCqsCreated";
+ private static final String CQS_CREATED = "numCqsCreated";
/** Name of the active CQs statistic */
- protected static final String CQS_ACTIVE = "numCqsActive";
+ private static final String CQS_ACTIVE = "numCqsActive";
/** Name of the stopped CQs statistic */
- protected static final String CQS_STOPPED = "numCqsStopped";
+ private static final String CQS_STOPPED = "numCqsStopped";
/** Name of the closed CQs statistic */
- protected static final String CQS_CLOSED = "numCqsClosed";
+ private static final String CQS_CLOSED = "numCqsClosed";
/** Name of the client's CQs statistic */
- protected static final String CQS_ON_CLIENT = "numCqsOnClient";
+ private static final String CQS_ON_CLIENT = "numCqsOnClient";
/** Number of clients with CQs statistic */
- protected static final String CLIENTS_WITH_CQS = "numClientsWithCqs";
-
+ private static final String CLIENTS_WITH_CQS = "numClientsWithCqs";
/** CQ query execution time. */
- protected static final String CQ_QUERY_EXECUTION_TIME = "cqQueryExecutionTime";
+ private static final String CQ_QUERY_EXECUTION_TIME = "cqQueryExecutionTime";
/** CQ query execution in progress */
- protected static final String CQ_QUERY_EXECUTION_IN_PROGRESS = "cqQueryExecutionInProgress";
+ private static final String CQ_QUERY_EXECUTION_IN_PROGRESS = "cqQueryExecutionInProgress";
/** Completed CQ query executions */
- protected static final String CQ_QUERY_EXECUTIONS_COMPLETED = "cqQueryExecutionsCompleted";
+ private static final String CQ_QUERY_EXECUTIONS_COMPLETED = "cqQueryExecutionsCompleted";
/** Unique CQs, number of different CQ queries */
- protected static final String UNIQUE_CQ_QUERY = "numUniqueCqQuery";
+ private static final String UNIQUE_CQ_QUERY = "numUniqueCqQuery";
/** Id of the CQs created statistic */
private static final int _numCqsCreatedId;
@@ -104,7 +103,7 @@ public class CqServiceVsdStats {
/** Id for unique CQs, difference in CQ queries */
private static final int _numUniqueCqQuery;
- /**
+ /*
* Static initializer to create and initialize the <code>StatisticsType</code>
*/
static {
@@ -140,7 +139,6 @@ public class CqServiceVsdStats {
_cqQueryExecutionsCompletedId = _type.nameToId(CQ_QUERY_EXECUTIONS_COMPLETED);
_cqQueryExecutionInProgressId = _type.nameToId(CQ_QUERY_EXECUTION_IN_PROGRESS);
_numUniqueCqQuery = _type.nameToId(UNIQUE_CQ_QUERY);
-
}
/** The <code>Statistics</code> instance to which most behavior is delegated */
@@ -152,12 +150,10 @@ public class CqServiceVsdStats {
* @param factory The <code>StatisticsFactory</code> which creates the <code>Statistics</code>
* instance
*/
- public CqServiceVsdStats(StatisticsFactory factory) {
+ CqServiceVsdStats(StatisticsFactory factory) {
this._stats = factory.createAtomicStatistics(_type, "CqServiceStats");
}
- // /////////////////// Instance Methods /////////////////////
-
/**
* Closes the <code>HARegionQueueStats</code>.
*/
@@ -170,14 +166,14 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsCreated" stat
*/
- public long getNumCqsCreated() {
+ long getNumCqsCreated() {
return this._stats.getLong(_numCqsCreatedId);
}
/**
* Increments the "numCqsCreated" stat by 1.
*/
- public void incCqsCreated() {
+ void incCqsCreated() {
this._stats.incLong(_numCqsCreatedId, 1);
}
@@ -186,21 +182,21 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsActive" stat
*/
- public long getNumCqsActive() {
+ long getNumCqsActive() {
return this._stats.getLong(_numCqsActiveId);
}
/**
* Increments the "numCqsActive" stat by 1.
*/
- public void incCqsActive() {
+ void incCqsActive() {
this._stats.incLong(_numCqsActiveId, 1);
}
/**
* Decrements the "numCqsActive" stat by 1.
*/
- public void decCqsActive() {
+ void decCqsActive() {
this._stats.incLong(_numCqsActiveId, -1);
}
@@ -209,21 +205,21 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsStopped" stat
*/
- public long getNumCqsStopped() {
+ long getNumCqsStopped() {
return this._stats.getLong(_numCqsStoppedId);
}
/**
* Increments the "numCqsStopped" stat by 1.
*/
- public void incCqsStopped() {
+ void incCqsStopped() {
this._stats.incLong(_numCqsStoppedId, 1);
}
/**
* Decrements the "numCqsStopped" stat by 1.
*/
- public void decCqsStopped() {
+ void decCqsStopped() {
this._stats.incLong(_numCqsStoppedId, -1);
}
@@ -232,14 +228,14 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsClosed" stat
*/
- public long getNumCqsClosed() {
+ long getNumCqsClosed() {
return this._stats.getLong(_numCqsClosedId);
}
/**
* Increments the "numCqsClosed" stat by 1.
*/
- public void incCqsClosed() {
+ void incCqsClosed() {
this._stats.incLong(_numCqsClosedId, 1);
}
@@ -248,21 +244,21 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsOnClient" stat
*/
- public long getNumCqsOnClient() {
+ long getNumCqsOnClient() {
return this._stats.getLong(_numCqsOnClientId);
}
/**
* Increments the "numCqsOnClient" stat by 1.
*/
- public void incCqsOnClient() {
+ void incCqsOnClient() {
this._stats.incLong(_numCqsOnClientId, 1);
}
/**
* Decrements the "numCqsOnClient" stat by 1.
*/
- public void decCqsOnClient() {
+ void decCqsOnClient() {
this._stats.incLong(_numCqsOnClientId, -1);
}
@@ -278,21 +274,21 @@ public class CqServiceVsdStats {
/**
* Increments the "numClientsWithCqs" stat by 1.
*/
- public void incClientsWithCqs() {
+ void incClientsWithCqs() {
this._stats.incLong(_numClientsWithCqsId, 1);
}
/**
* Decrements the "numCqsOnClient" stat by 1.
*/
- public void decClientsWithCqs() {
+ void decClientsWithCqs() {
this._stats.incLong(_numClientsWithCqsId, -1);
}
/**
* Start the CQ Query Execution time.
*/
- public long startCqQueryExecution() {
+ long startCqQueryExecution() {
this._stats.incInt(_cqQueryExecutionInProgressId, 1);
return NanoTimer.getTime();
}
@@ -302,7 +298,7 @@ public class CqServiceVsdStats {
*
* @param start long time value.
*/
- public void endCqQueryExecution(long start) {
+ void endCqQueryExecution(long start) {
long ts = NanoTimer.getTime();
this._stats.incLong(_cqQueryExecutionTimeId, ts - start);
this._stats.incInt(_cqQueryExecutionInProgressId, -1);
@@ -321,14 +317,14 @@ public class CqServiceVsdStats {
/**
* Increments number of Unique queries.
*/
- public void incUniqueCqQuery() {
+ void incUniqueCqQuery() {
this._stats.incInt(_numUniqueCqQuery, 1);
}
/**
* Decrements number of unique Queries.
*/
- public void decUniqueCqQuery() {
+ void decUniqueCqQuery() {
this._stats.incInt(_numUniqueCqQuery, -1);
}
@@ -338,11 +334,8 @@ public class CqServiceVsdStats {
* tests.
* <p>
* Returns the number of CQs (active + suspended) on the given region.
- *
- * @param regionName
*/
- public long numCqsOnRegion(String regionName) {
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ public long numCqsOnRegion(final InternalCache cache, String regionName) {
if (cache == null) {
return 0;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java
index ec6e984..c484105 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java
@@ -21,24 +21,18 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.logging.log4j.Logger;
-import org.apache.geode.CancelException;
import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
-import org.apache.geode.SystemFailure;
-import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EvictionAction;
-import org.apache.geode.cache.client.internal.UserAttributes;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqAttributesMutator;
import org.apache.geode.cache.query.CqClosedException;
import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.CqExistsException;
-import org.apache.geode.cache.query.CqListener;
import org.apache.geode.cache.query.CqResults;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryException;
@@ -49,6 +43,7 @@ import org.apache.geode.cache.query.internal.CompiledRegion;
import org.apache.geode.cache.query.internal.CompiledSelect;
import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQuery;
+import org.apache.geode.i18n.StringId;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.Token;
@@ -58,7 +53,6 @@ import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
-import org.apache.geode.i18n.StringId;
public class ServerCQImpl extends CqQueryImpl implements DataSerializable, ServerCQ {
private static final Logger logger = LogService.getLogger();
@@ -84,7 +78,7 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
public volatile boolean cqResultKeysInitialized = false;
/** Boolean flag to see if the CQ is on Partitioned Region */
- public volatile boolean isPR = false;
+ volatile boolean isPR = false;
private ClientProxyMembershipID clientProxyId = null;
@@ -92,7 +86,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
private String serverCqName;
-
/** identifier assigned to this query for FilterRoutingInfos */
private Long filterID;
@@ -106,21 +99,11 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
// For deserialization
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#getFilterID()
- */
@Override
public Long getFilterID() {
return this.filterID;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#setFilterID(java.lang.Long)
- */
@Override
public void setFilterID(Long filterID) {
this.filterID = filterID;
@@ -142,19 +125,12 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
CacheClientProxy clientProxy = null;
this.clientProxyId = p_clientProxyId;
- // servConnection = serverSideConnection;
if (p_ccn != null) {
this.ccn = p_ccn;
clientProxy = p_ccn.getClientProxy(p_clientProxyId, true);
}
- /*
- * try { initCq(); } catch (CqExistsException cqe) { // Should not happen. throw new
- * CqException(LocalizedStrings.CqQueryImpl_UNABLE_TO_CREATE_CQ_0_ERROR__1.toLocalizedString(new
- * Object[] { cqName, cqe.getMessage()})); }
- */
-
validateCq();
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -228,13 +204,11 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
throw new CqException(errMsg);
}
- // checkAndSetCqOnRegion();
-
// Can be null by the time we are here
if (clientProxy != null) {
clientProxy.incCqCount();
if (clientProxy.hasOneCq()) {
- cqService.stats.incClientsWithCqs();
+ cqService.stats().incClientsWithCqs();
}
if (isDebugEnabled) {
logger.debug("Added CQ to the base region: {} With key as: {}", cqBaseRegion.getFullPath(),
@@ -307,7 +281,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
* query.
*
* @return String modified query.
- * @throws CqException
*/
private Query constructServerSideQuery() throws QueryException {
GemFireCacheImpl cache = (GemFireCacheImpl) cqService.getCache();
@@ -328,7 +301,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
* Returns if the passed key is part of the CQs result set. This method needs to be called once
* the CQ result key caching is completed (cqResultsCacheInitialized is true).
*
- * @param key
* @return true if key is in the Results Cache.
*/
public boolean isPartOfCqResult(Object key) {
@@ -352,27 +324,18 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#addToCqResultKeys(java.lang.Object)
- */
@Override
public void addToCqResultKeys(Object key) {
if (!CqServiceProvider.MAINTAIN_KEYS) {
return;
}
- // this.logger.fine("Adding key to Results Cache For CQ :" +
- // this.cqName + " key :" + key);
if (this.cqResultKeys != null) {
synchronized (this.cqResultKeys) {
this.cqResultKeys.put(key, TOKEN);
if (!this.cqResultKeysInitialized) {
// This key could be coming after add, destroy.
// Remove this from destroy queue.
- // this.logger.fine("Removing key from Destroy Cache For CQ :" +
- // this.cqName + " key :" + key);
if (this.destroysWhileCqResultsInProgress != null) {
this.destroysWhileCqResultsInProgress.remove(key);
}
@@ -381,21 +344,11 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
}
-
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqQuery2#removeFromCqResultKeys(java.lang.Object,
- * boolean)
- */
@Override
public void removeFromCqResultKeys(Object key, boolean isTokenMode) {
if (!CqServiceProvider.MAINTAIN_KEYS) {
return;
}
- // this.logger.fine("Removing key from Results Cache For CQ :" +
- // this.cqName + " key :" + key);
if (this.cqResultKeys != null) {
synchronized (this.cqResultKeys) {
if (isTokenMode && this.cqResultKeys.get(key) != Token.DESTROYED) {
@@ -403,8 +356,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
this.cqResultKeys.remove(key);
if (!this.cqResultKeysInitialized) {
- // this.logger.fine("Adding key to Destroy Cache For CQ :" +
- // this.cqName + " key :" + key);
if (this.destroysWhileCqResultsInProgress != null) {
this.destroysWhileCqResultsInProgress.add(key);
}
@@ -415,10 +366,8 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
/**
* Marks the key as destroyed in the CQ Results key cache.
- *
- * @param key
*/
- public void markAsDestroyedInCqResultKeys(Object key) {
+ void markAsDestroyedInCqResultKeys(Object key) {
if (!CqServiceProvider.MAINTAIN_KEYS) {
return;
}
@@ -439,12 +388,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
}
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#setCqResultsCacheInitialized()
- */
@Override
public void setCqResultsCacheInitialized() {
if (CqServiceProvider.MAINTAIN_KEYS) {
@@ -466,13 +409,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqQuery2#isOldValueRequiredForQueryProcessing(
- * java.lang.Object)
- */
@Override
public boolean isOldValueRequiredForQueryProcessing(Object key) {
if (this.cqResultKeysInitialized && this.isPartOfCqResult(key)) {
@@ -484,18 +420,11 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
/**
* Closes the Query. On Client side, sends the cq close request to server. On Server side, takes
* care of repository cleanup.
- *
- * @throws CqException
*/
public void close() throws CqClosedException, CqException {
close(true);
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#close(boolean)
- */
@Override
public void close(boolean sendRequestToServer) throws CqClosedException, CqException {
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -523,9 +452,9 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
// Stat update.
if (stateBeforeClosing == CqStateImpl.RUNNING) {
- cqService.stats.decCqsActive();
+ cqService.stats().decCqsActive();
} else if (stateBeforeClosing == CqStateImpl.STOPPED) {
- cqService.stats.decCqsStopped();
+ cqService.stats().decCqsStopped();
}
// Clean-up the CQ Results Cache.
@@ -537,8 +466,8 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
// Set the state to close, and update stats
this.cqState.setState(CqStateImpl.CLOSED);
- cqService.stats.incCqsClosed();
- cqService.stats.decCqsOnClient();
+ cqService.stats().incCqsClosed();
+ cqService.stats().decCqsOnClient();
if (this.stats != null)
this.stats.close();
}
@@ -564,9 +493,8 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
/**
* Clears the resource used by CQ.
- *
- * @throws CqException
*/
+ @Override
protected void cleanup() throws CqException {
// CqBaseRegion
try {
@@ -575,7 +503,7 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
CacheClientProxy clientProxy = ccn.getClientProxy(clientProxyId);
clientProxy.decCqCount();
if (clientProxy.hasNoCq()) {
- cqService.stats.decClientsWithCqs();
+ cqService.stats().decClientsWithCqs();
}
}
} catch (Exception ex) {
@@ -587,16 +515,9 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
/**
- * @param serverCqName The serverCqName to set.
- */
- public void setServerCqName(String serverCqName) {
-
- this.serverCqName = serverCqName;
- }
-
- /**
* Stop or pause executing the query.
*/
+ @Override
public void stop() throws CqClosedException, CqException {
boolean isStopped = false;
synchronized (this.cqState) {
@@ -613,18 +534,16 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
// Change state and stats on the client side
this.cqState.setState(CqStateImpl.STOPPED);
- this.cqService.stats.incCqsStopped();
- this.cqService.stats.decCqsActive();
+ this.cqService.stats().incCqsStopped();
+ this.cqService.stats().decCqsActive();
if (logger.isDebugEnabled()) {
logger.debug("Successfully stopped the CQ. {}", cqName);
}
}
}
- /* DataSerializableFixedID methods ---------------------------------------- */
-
+ @Override
public void fromData(DataInput in) throws IOException, ClassNotFoundException {
- // this.cqName = DataSerializer.readString(in);
synchronized (cqState) {
this.cqState.setState(DataSerializer.readInteger(in));
}
@@ -633,23 +552,14 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
this.filterID = in.readLong();
}
- /*
- * public int getDSFID() { return CQ_QUERY; }
- */
-
+ @Override
public void toData(DataOutput out) throws IOException {
- // DataSerializer.writeString(this.cqName, out);
DataSerializer.writeInteger(this.cqState.getState(), out);
DataSerializer.writeBoolean(this.isDurable, out);
DataSerializer.writeString(this.queryString, out);
out.writeLong(this.filterID);
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#isPR()
- */
@Override
public boolean isPR() {
return isPR;
@@ -676,5 +586,4 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
throw new IllegalStateException("Execute cannot be called on a CQ on the server");
}
-
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
index bcf9806..9bddbc7 100644
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
@@ -27,7 +27,6 @@ import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
import org.apache.geode.distributed.internal.DistributionStats;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
@@ -85,8 +84,7 @@ public class ExecuteCQ extends BaseCQCommand {
ServerCQ cqQuery = null;
try {
- qService =
- (DefaultQueryService) ((GemFireCacheImpl) crHelper.getCache()).getLocalQueryService();
+ qService = (DefaultQueryService) crHelper.getCache().getLocalQueryService();
// Authorization check
AuthorizeRequest authzRequest = servConn.getAuthzRequest();
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
index f333b4b..de61445 100755
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
@@ -28,7 +28,6 @@ import org.apache.geode.cache.query.internal.cq.CqServiceImpl;
import org.apache.geode.cache.query.internal.cq.CqServiceProvider;
import org.apache.geode.cache.query.internal.cq.ServerCQImpl;
import org.apache.geode.distributed.internal.DistributionStats;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
@@ -111,8 +110,7 @@ public class ExecuteCQ61 extends BaseCQCommand {
ServerCQImpl cqQuery = null;
try {
- qService =
- (DefaultQueryService) ((GemFireCacheImpl) crHelper.getCache()).getLocalQueryService();
+ qService = (DefaultQueryService) crHelper.getCache().getLocalQueryService();
// Authorization check
AuthorizeRequest authzRequest = servConn.getAuthzRequest();
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
index eac9ed3..a2d201d 100755
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
@@ -22,7 +22,6 @@ import java.util.List;
import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.query.internal.cq.CqService;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
@@ -64,8 +63,7 @@ public class GetDurableCQs extends BaseCQCommand {
CqService cqServiceForExec = null;
try {
- qService =
- (DefaultQueryService) ((GemFireCacheImpl) crHelper.getCache()).getLocalQueryService();
+ qService = (DefaultQueryService) crHelper.getCache().getLocalQueryService();
this.securityService.authorizeClusterRead();
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java
index 7ace0e8..f4cd706 100644
--- a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java
+++ b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java
@@ -14,20 +14,15 @@
*/
package org.apache.geode.cache.query.cq.dunit;
-import org.junit.experimental.categories.Category;
-import org.junit.Test;
-
import static org.junit.Assert.*;
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
-
import java.util.Collection;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.query.CqException;
-import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.CqServiceStatistics;
import org.apache.geode.cache.query.CqStatistics;
import org.apache.geode.cache.query.QueryService;
@@ -40,7 +35,7 @@ import org.apache.geode.cache.query.internal.cq.CqServiceImpl;
import org.apache.geode.cache.query.internal.cq.CqServiceVsdStats;
import org.apache.geode.cache.query.internal.cq.InternalCqQuery;
import org.apache.geode.cache30.CacheSerializableRunnable;
-import org.apache.geode.cache30.CacheTestCase;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
@@ -48,27 +43,26 @@ import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
+import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
/**
- * This class tests the ContiunousQuery mechanism in GemFire. This includes the test with different
+ * This class tests the ContinuousQuery mechanism in GemFire. This includes the test with different
* data activities.
- *
*/
@Category(DistributedTest.class)
public class CqStatsDUnitTest extends JUnit4CacheTestCase {
+ // TODO: delete this use of CqQueryDUnitTest
private CqQueryDUnitTest cqDUnitTest = new CqQueryDUnitTest();
- public CqStatsDUnitTest() {
- super();
- }
-
@Override
public final void postSetUp() throws Exception {
// avoid IllegalStateException from HandShake by connecting all vms to
// system before creating pool
getSystem();
Invoke.invokeInEveryVM(new SerializableRunnable("getSystem") {
+ @Override
public void run() {
getSystem();
}
@@ -81,6 +75,7 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
public void validateCQStats(VM vm, final String cqName, final int creates, final int updates,
final int deletes, final int totalEvents, final int cqListenerInvocations) {
vm.invoke(new CacheSerializableRunnable("Validate CQs") {
+ @Override
public void run2() throws CacheException {
LogWriterUtils.getLogWriter().info("### Validating CQ Stats. ### " + cqName);
// Get CQ Service.
@@ -161,6 +156,7 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
final int stopped, final int closed, final int cqsOnClient, final int cqsOnRegion,
final int clientsWithCqs) {
vm.invoke(new CacheSerializableRunnable("Validate CQ Service Stats") {
+ @Override
public void run2() throws CacheException {
LogWriterUtils.getLogWriter().info("### Validating CQ Service Stats. ### ");
// Get CQ Service.
@@ -176,7 +172,7 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
CqServiceVsdStats cqServiceVsdStats = null;
try {
cqServiceVsdStats =
- ((CqServiceImpl) ((DefaultQueryService) qService).getCqService()).stats;
+ ((CqServiceImpl) ((DefaultQueryService) qService).getCqService()).stats();
} catch (CqException e) {
// TODO Auto-generated catch block
e.printStackTrace();
@@ -185,12 +181,14 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
fail("Failed to get CQ Service Stats");
}
- getCache().getLogger().info("#### CQ Service stats: " + " CQs created: "
- + cqServiceStats.numCqsCreated() + " CQs active: " + cqServiceStats.numCqsActive()
- + " CQs stopped: " + cqServiceStats.numCqsStopped() + " CQs closed: "
- + cqServiceStats.numCqsClosed() + " CQs on Client: " + cqServiceStats.numCqsOnClient()
- + " CQs on region /root/regionA : " + cqServiceVsdStats.numCqsOnRegion("/root/regionA")
- + " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
+ getCache().getLogger()
+ .info("#### CQ Service stats: " + " CQs created: " + cqServiceStats.numCqsCreated()
+ + " CQs active: " + cqServiceStats.numCqsActive() + " CQs stopped: "
+ + cqServiceStats.numCqsStopped() + " CQs closed: " + cqServiceStats.numCqsClosed()
+ + " CQs on Client: " + cqServiceStats.numCqsOnClient()
+ + " CQs on region /root/regionA : "
+ + cqServiceVsdStats.numCqsOnRegion(GemFireCacheImpl.getInstance(), "/root/regionA")
+ + " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
// Check for created count.
@@ -223,7 +221,7 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
// Check for CQs on region.
if (cqsOnRegion != CqQueryDUnitTest.noTest) {
assertEquals("Number of CQs on region /root/regionA mismatch", cqsOnRegion,
- cqServiceVsdStats.numCqsOnRegion("/root/regionA"));
+ cqServiceVsdStats.numCqsOnRegion(GemFireCacheImpl.getInstance(), "/root/regionA"));
}
// Check for clients with CQs count.
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java
index d6068f1..c03bb8b 100644
--- a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java
+++ b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java
@@ -14,21 +14,16 @@
*/
package org.apache.geode.cache.query.cq.dunit;
-import org.junit.experimental.categories.Category;
-import org.junit.Test;
-
import static org.junit.Assert.*;
-import org.apache.geode.distributed.*;
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
+import java.util.Collection;
+import java.util.Properties;
-import java.util.*;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.query.CqException;
-import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.CqServiceStatistics;
import org.apache.geode.cache.query.CqStatistics;
import org.apache.geode.cache.query.QueryService;
@@ -41,7 +36,8 @@ import org.apache.geode.cache.query.internal.cq.CqServiceImpl;
import org.apache.geode.cache.query.internal.cq.CqServiceVsdStats;
import org.apache.geode.cache.query.internal.cq.InternalCqQuery;
import org.apache.geode.cache30.CacheSerializableRunnable;
-import org.apache.geode.cache30.CacheTestCase;
+import org.apache.geode.distributed.ConfigurationProperties;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
@@ -49,21 +45,19 @@ import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
+import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
/**
- * This class tests the ContiunousQuery mechanism in GemFire. This includes the test with different
+ * This class tests the ContinuousQuery mechanism in GemFire. This includes the test with different
* data activities.
- *
*/
@Category(DistributedTest.class)
public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
+ // TODO: delete this use of CqQueryUsingPoolDUnitTest
private CqQueryUsingPoolDUnitTest cqDUnitTest = new CqQueryUsingPoolDUnitTest();
- public CqStatsUsingPoolDUnitTest() {
- super();
- }
-
@Override
public Properties getDistributedSystemProperties() {
Properties result = super.getDistributedSystemProperties();
@@ -77,6 +71,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
// system before creating pool
getSystem();
Invoke.invokeInEveryVM(new SerializableRunnable("getSystem") {
+ @Override
public void run() {
getSystem();
}
@@ -89,6 +84,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
private void validateCQStats(VM vm, final String cqName, final int creates, final int updates,
final int deletes, final int totalEvents, final int cqListenerInvocations) {
vm.invoke(new CacheSerializableRunnable("Validate CQs") {
+ @Override
public void run2() throws CacheException {
LogWriterUtils.getLogWriter().info("### Validating CQ Stats. ### " + cqName);
// Get CQ Service.
@@ -169,6 +165,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
final int stopped, final int closed, final int cqsOnClient, final int cqsOnRegion,
final int clientsWithCqs) {
vm.invoke(new CacheSerializableRunnable("Validate CQ Service Stats") {
+ @Override
public void run2() throws CacheException {
LogWriterUtils.getLogWriter().info("### Validating CQ Service Stats. ### ");
// Get CQ Service.
@@ -184,7 +181,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
CqServiceVsdStats cqServiceVsdStats = null;
try {
cqServiceVsdStats =
- ((CqServiceImpl) ((DefaultQueryService) qService).getCqService()).stats;
+ ((CqServiceImpl) ((DefaultQueryService) qService).getCqService()).stats();
} catch (CqException e) {
// TODO Auto-generated catch block
e.printStackTrace();
@@ -193,12 +190,14 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
fail("Failed to get CQ Service Stats");
}
- getCache().getLogger().info("#### CQ Service stats: " + " CQs created: "
- + cqServiceStats.numCqsCreated() + " CQs active: " + cqServiceStats.numCqsActive()
- + " CQs stopped: " + cqServiceStats.numCqsStopped() + " CQs closed: "
- + cqServiceStats.numCqsClosed() + " CQs on Client: " + cqServiceStats.numCqsOnClient()
- + " CQs on region /root/regionA : " + cqServiceVsdStats.numCqsOnRegion("/root/regionA")
- + " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
+ getCache().getLogger()
+ .info("#### CQ Service stats: " + " CQs created: " + cqServiceStats.numCqsCreated()
+ + " CQs active: " + cqServiceStats.numCqsActive() + " CQs stopped: "
+ + cqServiceStats.numCqsStopped() + " CQs closed: " + cqServiceStats.numCqsClosed()
+ + " CQs on Client: " + cqServiceStats.numCqsOnClient()
+ + " CQs on region /root/regionA : "
+ + cqServiceVsdStats.numCqsOnRegion(GemFireCacheImpl.getInstance(), "/root/regionA")
+ + " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
// Check for created count.
@@ -231,7 +230,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
// Check for CQs on region.
if (cqsOnRegion != CqQueryUsingPoolDUnitTest.noTest) {
assertEquals("Number of CQs on region /root/regionA mismatch", cqsOnRegion,
- cqServiceVsdStats.numCqsOnRegion("/root/regionA"));
+ cqServiceVsdStats.numCqsOnRegion(GemFireCacheImpl.getInstance(), "/root/regionA"));
}
// Check for clients with CQs count.
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java
index 66c4c0a..5dd0d24 100644
--- a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java
@@ -15,10 +15,8 @@
package org.apache.geode.cache.lucene.internal.distributed;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.Logger;
@@ -27,7 +25,7 @@ import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.lucene.LuceneQuery;
import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.logging.LogService;
/**
@@ -44,16 +42,15 @@ import org.apache.geode.internal.logging.LogService;
*/
public class TopEntriesFunctionCollector
implements ResultCollector<TopEntriesCollector, TopEntries> {
- // Use this instance to perform reduce operation
- final CollectorManager<TopEntriesCollector> manager;
+ private static final Logger logger = LogService.getLogger();
- final String id;
+ // Use this instance to perform reduce operation
+ private final CollectorManager<TopEntriesCollector> manager;
- // Instance of gemfire cache to check status and other utility methods
- final private GemFireCacheImpl cache;
- private static final Logger logger = LogService.getLogger();
+ private final String id;
private final Collection<TopEntriesCollector> subResults = new ArrayList<>();
+
private TopEntriesCollector mergedResults;
public TopEntriesFunctionCollector() {
@@ -65,8 +62,7 @@ public class TopEntriesFunctionCollector
}
public TopEntriesFunctionCollector(LuceneFunctionContext<TopEntriesCollector> context,
- GemFireCacheImpl cache) {
- this.cache = cache;
+ InternalCache cache) {
id = cache == null ? String.valueOf(this.hashCode()) : cache.getName();
int limit = context == null ? 0 : context.getLimit();
@@ -115,4 +111,8 @@ public class TopEntriesFunctionCollector
subResults.add(resultOfSingleExecution);
}
}
+
+ String id() {
+ return this.id;
+ }
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java
index 5313ced..6690850 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java
@@ -12,7 +12,6 @@
* 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;
@@ -41,6 +40,7 @@ import org.apache.geode.cache.lucene.internal.StringQueryProvider;
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.cache.lucene.test.LuceneTestUtilities;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.execute.InternalFunctionInvocationTargetException;
import org.apache.geode.internal.cache.execute.InternalRegionFunctionContext;
@@ -58,30 +58,30 @@ import org.mockito.Mockito;
@Category(UnitTest.class)
public class LuceneQueryFunctionJUnitTest {
- 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<TopEntriesCollector> mockResultSender;
- Region<Object, Object> mockRegion;
-
- RepositoryManager mockRepoManager;
- IndexRepository mockRepository1;
- IndexRepository mockRepository2;
- IndexResultCollector mockCollector;
- InternalLuceneService mockService;
- LuceneIndexImpl mockIndex;
- LuceneIndexStats mockStats;
-
- ArrayList<IndexRepository> repos;
- LuceneFunctionContext<IndexResultCollector> searchArgs;
- LuceneQueryProvider queryProvider;
- Query query;
+ private String regionPath = "/region";
+
+ private final EntryScore<String> r1_1 = new EntryScore<>("key-1-1", .5f);
+ private final EntryScore<String> r1_2 = new EntryScore<>("key-1-2", .4f);
+ private final EntryScore<String> r1_3 = new EntryScore<>("key-1-3", .3f);
+ private final EntryScore<String> r2_1 = new EntryScore<>("key-2-1", .45f);
+ private final EntryScore<String> r2_2 = new EntryScore<>("key-2-2", .35f);
+
+ private InternalRegionFunctionContext mockContext;
+ private ResultSender<TopEntriesCollector> mockResultSender;
+ private Region<Object, Object> mockRegion;
+
+ private RepositoryManager mockRepoManager;
+ private IndexRepository mockRepository1;
+ private IndexRepository mockRepository2;
+ private IndexResultCollector mockCollector;
+ private InternalLuceneService mockService;
+ private LuceneIndexImpl mockIndex;
+ private LuceneIndexStats mockStats;
+
+ private ArrayList<IndexRepository> repos;
+ private LuceneFunctionContext<IndexResultCollector> searchArgs;
+ private LuceneQueryProvider queryProvider;
+ private Query query;
private InternalCache mockCache;
@@ -120,7 +120,7 @@ public class LuceneQueryFunctionJUnitTest {
List<EntryScore> hits = result.getEntries().getHits();
assertEquals(5, hits.size());
- TopEntriesJUnitTest.verifyResultOrder(result.getEntries().getHits(), r1_1, r2_1, r1_2, r2_2,
+ LuceneTestUtilities.verifyResultOrder(result.getEntries().getHits(), r1_1, r2_1, r1_2, r2_2,
r1_3);
}
@@ -161,7 +161,7 @@ public class LuceneQueryFunctionJUnitTest {
List<EntryScore> hits = result.getEntries().getHits();
assertEquals(3, hits.size());
- TopEntriesJUnitTest.verifyResultOrder(result.getEntries().getHits(), r1_1, r2_1, r1_2);
+ LuceneTestUtilities.verifyResultOrder(result.getEntries().getHits(), r1_1, r2_1, r1_2);
}
@Test
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java
index 3bfebdf..5767390 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java
@@ -26,21 +26,22 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.CopyHelper;
import org.apache.geode.cache.lucene.internal.LuceneServiceImpl;
import org.apache.geode.cache.lucene.internal.distributed.TopEntriesCollectorManager.ListScanner;
+import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
import org.apache.geode.test.junit.categories.UnitTest;
@Category(UnitTest.class)
public class TopEntriesCollectorJUnitTest {
- private EntryScore<String> r1_1 = new EntryScore<String>("1-1", .9f);
- private EntryScore<String> r1_2 = new EntryScore<String>("1-2", .7f);
- private EntryScore<String> r1_3 = new EntryScore<String>("1-3", .5f);
+ private EntryScore<String> r1_1 = new EntryScore<>("1-1", .9f);
+ private EntryScore<String> r1_2 = new EntryScore<>("1-2", .7f);
+ private EntryScore<String> r1_3 = new EntryScore<>("1-3", .5f);
- private EntryScore<String> r2_1 = new EntryScore<String>("2-1", .85f);
- private EntryScore<String> r2_2 = new EntryScore<String>("2-2", .65f);
+ private EntryScore<String> r2_1 = new EntryScore<>("2-1", .85f);
+ private EntryScore<String> r2_2 = new EntryScore<>("2-2", .65f);
- private EntryScore<String> r3_1 = new EntryScore<String>("3-1", .8f);
- private EntryScore<String> r3_2 = new EntryScore<String>("3-2", .6f);
- private EntryScore<String> r3_3 = new EntryScore<String>("3-3", .4f);
+ private EntryScore<String> r3_1 = new EntryScore<>("3-1", .8f);
+ private EntryScore<String> r3_2 = new EntryScore<>("3-2", .6f);
+ private EntryScore<String> r3_3 = new EntryScore<>("3-3", .4f);
private TopEntriesCollectorManager manager;
@@ -72,7 +73,7 @@ public class TopEntriesCollectorJUnitTest {
TopEntriesCollector hits = manager.reduce(collectors);
assertEquals(8, hits.getEntries().getHits().size());
- TopEntriesJUnitTest.verifyResultOrder(hits.getEntries().getHits(), r1_1, r2_1, r3_1, r1_2, r2_2,
+ LuceneTestUtilities.verifyResultOrder(hits.getEntries().getHits(), r1_1, r2_1, r3_1, r1_2, r2_2,
r3_2, r1_3, r3_3);
// input collector should not change
@@ -116,7 +117,7 @@ public class TopEntriesCollectorJUnitTest {
c1.collect(r1_3.getKey(), r1_3.getScore());
assertEquals(3, c1.getEntries().getHits().size());
- TopEntriesJUnitTest.verifyResultOrder(c1.getEntries().getHits(), r1_1, r1_2, r1_3);
+ LuceneTestUtilities.verifyResultOrder(c1.getEntries().getHits(), r1_1, r1_2, r1_3);
ListScanner scanner = new ListScanner(c1.getEntries().getHits());
assertTrue(scanner.hasNext());
@@ -131,6 +132,6 @@ public class TopEntriesCollectorJUnitTest {
assertFalse(scanner.hasNext());
assertEquals(3, c1.getEntries().getHits().size());
- TopEntriesJUnitTest.verifyResultOrder(c1.getEntries().getHits(), r1_1, r1_2, r1_3);
+ LuceneTestUtilities.verifyResultOrder(c1.getEntries().getHits(), r1_1, r1_2, r1_3);
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java
index bf08877..5fd9e2d 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java
@@ -12,7 +12,6 @@
* 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.junit.Assert.*;
@@ -20,9 +19,7 @@ import static org.mockito.Matchers.any;
import static org.mockito.Mockito.*;
import java.util.Collection;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
import org.junit.Assert;
import org.junit.Before;
@@ -31,25 +28,26 @@ import org.junit.experimental.categories.Category;
import org.mockito.ArgumentMatcher;
import org.mockito.Mockito;
-import org.apache.geode.CancelCriterion;
-import org.apache.geode.cache.execute.FunctionException;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.test.junit.categories.UnitTest;
@Category(UnitTest.class)
public class TopEntriesFunctionCollectorJUnitTest {
- EntryScore<String> r1_1;
- EntryScore<String> r1_2;
- EntryScore<String> r2_1;
- EntryScore<String> r2_2;
- TopEntriesCollector result1, result2;
+
+ private EntryScore<String> r1_1;
+ private EntryScore<String> r1_2;
+ private EntryScore<String> r2_1;
+ private EntryScore<String> r2_2;
+ private TopEntriesCollector result1;
+ private TopEntriesCollector result2;
@Before
public void initializeCommonObjects() {
- r1_1 = new EntryScore<String>("3", .9f);
- r1_2 = new EntryScore<String>("1", .8f);
- r2_1 = new EntryScore<String>("2", 0.85f);
- r2_2 = new EntryScore<String>("4", 0.1f);
+ r1_1 = new EntryScore<>("3", .9f);
+ r1_2 = new EntryScore<>("1", .8f);
+ r2_1 = new EntryScore<>("2", 0.85f);
+ r2_2 = new EntryScore<>("4", 0.1f);
result1 = new TopEntriesCollector(null);
result1.collect(r1_1);
@@ -73,13 +71,9 @@ public class TopEntriesFunctionCollectorJUnitTest {
collector.addResult(null, result1);
collector.addResult(null, result2);
- final CountDownLatch insideThread = new CountDownLatch(1);
- final CountDownLatch resultReceived = new CountDownLatch(1);
-
- final AtomicReference<TopEntries> result = new AtomicReference<>();
TopEntries merged = collector.getResult(1, TimeUnit.SECONDS);
assertEquals(4, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
}
@Test
@@ -95,7 +89,7 @@ public class TopEntriesFunctionCollectorJUnitTest {
TopEntries merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(3, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2);
}
@Test
@@ -108,7 +102,7 @@ public class TopEntriesFunctionCollectorJUnitTest {
TopEntries merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(4, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
}
@Test
@@ -121,12 +115,12 @@ public class TopEntriesFunctionCollectorJUnitTest {
TopEntries merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(4, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(4, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
}
@Test
@@ -167,7 +161,7 @@ public class TopEntriesFunctionCollectorJUnitTest {
TopEntries merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(2, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r2_1, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r2_1, r2_2);
}
@Test(expected = RuntimeException.class)
@@ -184,10 +178,10 @@ public class TopEntriesFunctionCollectorJUnitTest {
@Test
public void testCollectorName() {
- GemFireCacheImpl mockCache = mock(GemFireCacheImpl.class);
+ InternalCache mockCache = mock(InternalCache.class);
Mockito.doReturn("server").when(mockCache).getName();
TopEntriesFunctionCollector function = new TopEntriesFunctionCollector(null, mockCache);
- assertEquals("server", function.id);
+ assertEquals("server", function.id());
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java
index fcfebbc..e21ac7f 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java
@@ -16,9 +16,6 @@ package org.apache.geode.cache.lucene.internal.distributed;
import static org.junit.Assert.*;
-import java.util.Collection;
-import java.util.Iterator;
-
import org.jmock.Mockery;
import org.jmock.lib.concurrent.Synchroniser;
import org.jmock.lib.legacy.ClassImposteriser;
@@ -30,6 +27,7 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.CopyHelper;
import org.apache.geode.cache.lucene.LuceneQueryFactory;
import org.apache.geode.cache.lucene.internal.LuceneServiceImpl;
+import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
import org.apache.geode.test.junit.categories.UnitTest;
@Category(UnitTest.class)
@@ -44,34 +42,34 @@ public class TopEntriesJUnitTest {
@Test
public void testPopulateTopEntries() {
- TopEntries<String> hits = new TopEntries<String>();
+ TopEntries<String> hits = new TopEntries<>();
hits.addHit(r1_1);
hits.addHit(r2_1);
hits.addHit(r1_2);
hits.addHit(r2_2);
assertEquals(4, hits.size());
- verifyResultOrder(hits.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(hits.getHits(), r1_1, r2_1, r1_2, r2_2);
}
@Test
public void putSameScoreEntries() {
- TopEntries<String> hits = new TopEntries<String>();
- EntryScore<String> r1 = new EntryScore<String>("1", .8f);
- EntryScore<String> r2 = new EntryScore<String>("2", .8f);
+ TopEntries<String> hits = new TopEntries<>();
+ EntryScore<String> r1 = new EntryScore<>("1", .8f);
+ EntryScore<String> r2 = new EntryScore<>("2", .8f);
hits.addHit(r1);
hits.addHit(r2);
assertEquals(2, hits.size());
- verifyResultOrder(hits.getHits(), r1, r2);
+ LuceneTestUtilities.verifyResultOrder(hits.getHits(), r1, r2);
}
@Test
public void testInitialization() {
- TopEntries<String> hits = new TopEntries<String>();
+ TopEntries<String> hits = new TopEntries<>();
assertEquals(LuceneQueryFactory.DEFAULT_LIMIT, hits.getLimit());
- hits = new TopEntries<String>(123);
+ hits = new TopEntries<>(123);
assertEquals(123, hits.getLimit());
}
@@ -82,47 +80,33 @@ public class TopEntriesJUnitTest {
@Test
public void enforceLimit() throws Exception {
- TopEntries<String> hits = new TopEntries<String>(3);
+ TopEntries<String> hits = new TopEntries<>(3);
hits.addHit(r1_1);
hits.addHit(r2_1);
hits.addHit(r1_2);
hits.addHit(r2_2);
assertEquals(3, hits.size());
- verifyResultOrder(hits.getHits(), r1_1, r2_1, r1_2);
+ LuceneTestUtilities.verifyResultOrder(hits.getHits(), r1_1, r2_1, r1_2);
}
@Test
public void testSerialization() {
LuceneServiceImpl.registerDataSerializables();
- TopEntries<String> hits = new TopEntries<String>(3);
+ TopEntries<String> hits = new TopEntries<>(3);
TopEntries<String> copy = CopyHelper.deepCopy(hits);
assertEquals(3, copy.getLimit());
assertEquals(0, copy.getHits().size());
- hits = new TopEntries<String>(3);
+ hits = new TopEntries<>(3);
hits.addHit(r1_1);
hits.addHit(r2_1);
hits.addHit(r1_2);
copy = CopyHelper.deepCopy(hits);
assertEquals(3, copy.size());
- verifyResultOrder(copy.getHits(), r1_1, r2_1, r1_2);
- }
-
- // TODO: extract to lucene test util class
- public static void verifyResultOrder(Collection<EntryScore<String>> list,
- EntryScore<String>... expectedEntries) {
- Iterator<EntryScore<String>> iter = list.iterator();
- for (EntryScore expectedEntry : expectedEntries) {
- if (!iter.hasNext()) {
- fail();
- }
- EntryScore toVerify = iter.next();
- assertEquals(expectedEntry.getKey(), toVerify.getKey());
- assertEquals(expectedEntry.getScore(), toVerify.getScore(), .0f);
- }
+ LuceneTestUtilities.verifyResultOrder(copy.getHits(), r1_1, r2_1, r1_2);
}
@Before
http://git-wip-us.apache.org/repos/asf/geode/blob/222ebf72/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 5563112..17f4dea 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
@@ -18,8 +18,10 @@ import static org.junit.Assert.*;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -44,6 +46,7 @@ import org.apache.geode.cache.lucene.LuceneService;
import org.apache.geode.cache.lucene.LuceneServiceProvider;
import org.apache.geode.cache.lucene.internal.LuceneIndexForPartitionedRegion;
import org.apache.geode.cache.lucene.internal.LuceneServiceImpl;
+import org.apache.geode.cache.lucene.internal.distributed.EntryScore;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.test.dunit.VM;
@@ -82,6 +85,19 @@ public class LuceneTestUtilities {
public static String Quarter3 = "Q3";
public static String Quarter4 = "Q4";
+ public static void verifyResultOrder(Collection<EntryScore<String>> list,
+ EntryScore<String>... expectedEntries) {
+ Iterator<EntryScore<String>> iter = list.iterator();
+ for (EntryScore expectedEntry : expectedEntries) {
+ if (!iter.hasNext()) {
+ fail();
+ }
+ EntryScore toVerify = iter.next();
+ assertEquals(expectedEntry.getKey(), toVerify.getKey());
+ assertEquals(expectedEntry.getScore(), toVerify.getScore(), .0f);
+ }
+ }
+
public static class IntRangeQueryProvider implements LuceneQueryProvider {
String fieldName;
int lowerValue;