SGF-660 - Spring Cache Abstraction annotations do not trigger CQ.

src/main/java/org/springframework/data/gemfire/config/annotation/ContinuousQueryConfiguration.java

@@ -111,11 +111,11 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
 			private List<ContinuousQueryDefinition> continuousQueryDefinitions = new ArrayList<>();
 
 			@Nullable @Override
-			public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+			public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
 
 				if (bean instanceof ContinuousQueryListenerContainer) {
 					this.container = (ContinuousQueryListenerContainer) bean;
-					this.continuousQueryDefinitions.forEach(definition -> this.container.addListener(definition));
+					this.continuousQueryDefinitions.forEach(this.container::addListener);
 					this.continuousQueryDefinitions.clear();
 				}
 				else if (isApplicationBean(bean)) {
@@ -182,6 +182,7 @@ public class ContinuousQueryConfiguration extends AbstractAnnotationConfigSuppor
 								true, true);
 
 						return nullSafeMap(beansOfType).values().stream().collect(Collectors.toList());
+
 					})
 					.orElseGet(Collections::emptyList)
 			);

src/main/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueries.java

@@ -27,7 +27,6 @@ import java.util.concurrent.Executor;
 import org.apache.geode.cache.client.Pool;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.data.gemfire.config.xml.GemfireConstants;
 import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
 import org.springframework.util.ErrorHandler;
 
@@ -75,9 +74,9 @@ public @interface EnableContinuousQueries {
 	/**
 	 * Refers to the name of the {@link Pool} over which CQs are registered and CQ events are received.
 	 *
-	 * Defaults to {@literal gemfirePool}.
+	 * Defaults to unset.
 	 */
-	String poolName() default GemfireConstants.DEFAULT_GEMFIRE_POOL_NAME;
+	String poolName() default "";
 
 	/**
 	 * Refers to the name of the {@link Executor} bean used to process CQ events asynchronously.

src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.java

@@ -117,7 +117,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 	private List<ContinuousQueryListenerContainerConfigurer> cqListenerContainerConfigurers = Collections.emptyList();
 
 	private ContinuousQueryListenerContainerConfigurer compositeCqListenerContainerConfigurer =
-		(beanName, container) -> nullSafeList(cqListenerContainerConfigurers).forEach(configurer ->
+		(beanName, container) -> nullSafeList(this.cqListenerContainerConfigurers).forEach(configurer ->
 			configurer.configure(beanName, container));
 
 	protected final Log logger = LogFactory.getLog(getClass());
@@ -147,14 +147,6 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 		applyContinuousQueryListenerContainerConfigurers(getCompositeContinuousQueryListenerContainerConfigurer());
 	}
 
-	/* (non-Javadoc) */
-	private QueryService validateQueryService(QueryService queryService) {
-
-		Assert.state(queryService != null, "QueryService was not properly initialized");
-
-		return queryService;
-	}
-
 	/**
 	 * Applies the array of {@link ContinuousQueryListenerContainerConfigurer} objects to customize the configuration
 	 * of this {@link ContinuousQueryListenerContainer}.
@@ -260,6 +252,14 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 		return getQueryService();
 	}
 
+	/* (non-Javadoc) */
+	private QueryService validateQueryService(QueryService queryService) {
+
+		Assert.state(queryService != null, "QueryService is required");
+
+		return queryService;
+	}
+
 	/**
 	 * Initialize the {@link Executor} used to process CQ events asynchronously.
 	 *
@@ -301,7 +301,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 	 * Initializes all the {@link CqQuery Continuous Queries} defined by the {@link ContinuousQueryDefinition defintions}.
 	 */
 	private void initContinuousQueries() {
-		initContinuousQueries(this.continuousQueryDefinitions);
+		initContinuousQueries(getContinuousQueryDefinitions());
 	}
 
 	/* (non-Javadoc) */
@@ -418,13 +418,24 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 	 * Returns a reference to all the configured/registered {@link CqQuery Continuous Queries}.
 	 *
 	 * @return a reference to all the configured/registered {@link CqQuery Continuous Queries}.
-	 * @see java.util.Queue
 	 * @see org.apache.geode.cache.query.CqQuery
+	 * @see java.util.Queue
 	 */
 	protected Queue<CqQuery> getContinuousQueries() {
 		return this.continuousQueries;
 	}
 
+	/**
+	 * Returns a reference to all the configured {@link ContinuousQueryDefinition ContinuousQueryDefinitions}.
+	 *
+	 * @return a reference to all the configured {@link ContinuousQueryDefinition ContinuousQueryDefinitions}.
+	 * @see org.springframework.data.gemfire.listener.ContinuousQueryDefinition
+	 * @see java.util.Set
+	 */
+	protected Set<ContinuousQueryDefinition> getContinuousQueryDefinitions() {
+		return this.continuousQueryDefinitions;
+	}
+
 	/**
 	 * Null-safe operation setting an array of {@link ContinuousQueryListenerContainerConfigurer} objects used to
 	 * customize the configuration of this {@link ContinuousQueryListenerContainer}.
@@ -504,6 +515,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 	 * @return the phase value of this CQ listener container.
 	 * @see org.springframework.context.Phased#getPhase()
 	 */
+	@Override
 	public int getPhase() {
 		return this.phase;
 	}
@@ -532,8 +544,8 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 	 * @param queries set of queries
 	 */
 	public void setQueryListeners(Set<ContinuousQueryDefinition> queries) {
-		this.continuousQueryDefinitions.clear();
-		this.continuousQueryDefinitions.addAll(nullSafeSet(queries));
+		getContinuousQueryDefinitions().clear();
+		getContinuousQueryDefinitions().addAll(nullSafeSet(queries));
 	}
 
 	/**
@@ -596,6 +608,10 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 		}
 	}
 
+	public boolean addContinuousQueryDefinition(ContinuousQueryDefinition definition) {
+		return Optional.ofNullable(definition).map(it -> getContinuousQueryDefinitions().add(it)).orElse(false);
+	}
+
 	/* (non-Javadoc) */
 	CqQuery addContinuousQuery(ContinuousQueryDefinition definition) {
 
@@ -641,6 +657,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 	public synchronized void start() {
 
 		if (!isRunning()) {
+
 			doStart();
 			this.running = true;
 
@@ -657,6 +674,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 
 	/* (non-Javadoc) */
 	private void execute(CqQuery query) {
+
 		try {
 			query.execute();
 		}
@@ -686,6 +704,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 	 * @see #handleListenerError(Throwable)
 	 */
 	private void notify(ContinuousQueryListener listener, CqEvent event) {
+
 		try {
 			listener.onEvent(event);
 		}
@@ -708,8 +727,7 @@ public class ContinuousQueryListenerContainer implements BeanFactoryAware, BeanN
 	 */
 	private void handleListenerError(Throwable cause) {
 
-		getErrorHandler()
-			.filter(errorHandler -> {
+		getErrorHandler().filter(errorHandler -> {
 
 				boolean active = this.isActive();
 

src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationIntegrationTests.java

@@ -48,6 +48,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.data.gemfire.PartitionedRegionFactoryBean;
 import org.springframework.data.gemfire.client.ClientRegionFactoryBean;
+import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
 import org.springframework.data.gemfire.listener.annotation.ContinuousQuery;
 import org.springframework.data.gemfire.process.ProcessWrapper;
 import org.springframework.data.gemfire.support.ConnectionEndpoint;
@@ -60,8 +61,8 @@ import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 
 /**
- * Integration tests for {@link EnableContinuousQueries}, {@link ContinuousQueryConfiguration}
- * and {@link ContinuousQuery}.
+ * Integration tests for {@link EnableContinuousQueries}, {@link ContinuousQueryConfiguration}, {@link ContinuousQuery}
+ * and {@link ContinuousQueryListenerContainer}.
  *
  * @author John Blum
  * @see org.junit.Test
@@ -216,7 +217,7 @@ public class EnableContinuousQueriesConfigurationIntegrationTests extends Client
 
 		@Bean
 		CacheServerConfigurer cacheServerPortConfigurer(
-			@Value("${" + GEMFIRE_CACHE_SERVER_PORT_PROPERTY + ":40404}") int port) {
+				@Value("${" + GEMFIRE_CACHE_SERVER_PORT_PROPERTY + ":40404}") int port) {
 
 			return (bean, cacheServerFactoryBean) -> cacheServerFactoryBean.setPort(port);
 		}
@@ -273,6 +274,7 @@ public class EnableContinuousQueriesConfigurationIntegrationTests extends Client
 					catch (InterruptedException ignore) {
 					}
 				}
+
 				@Override
 				public void close() {
 				}

src/test/java/org/springframework/data/gemfire/config/annotation/EnableContinuousQueriesConfigurationUnitTests.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.gemfire.config.annotation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import org.apache.geode.cache.GemFireCache;
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.cache.client.Pool;
+import org.apache.geode.cache.query.CqAttributes;
+import org.apache.geode.cache.query.CqEvent;
+import org.apache.geode.cache.query.CqQuery;
+import org.apache.geode.cache.query.QueryService;
+import org.junit.Test;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
+import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
+import org.springframework.data.gemfire.listener.annotation.ContinuousQuery;
+import org.springframework.data.gemfire.test.mock.annotation.EnableGemFireMockObjects;
+import org.springframework.stereotype.Component;
+
+/**
+ * Unit tests for {@link EnableContinuousQueries}, {@link ContinuousQueryConfiguration}, {@link ContinuousQuery}
+ * and {@link ContinuousQueryListenerContainer}.
+ *
+ * @author John Blum
+ * @see org.junit.Test
+ * @see org.springframework.data.gemfire.config.annotation.ContinuousQueryConfiguration
+ * @see org.springframework.data.gemfire.config.annotation.EnableContinuousQueries
+ * @see org.springframework.data.gemfire.listener.annotation.ContinuousQuery
+ * @since 2.0.1
+ */
+public class EnableContinuousQueriesConfigurationUnitTests {
+
+	private ConfigurableApplicationContext newApplicationContext(Class<?>... annotatedClasses) {
+		return new AnnotationConfigApplicationContext(annotatedClasses);
+	}
+
+	@Test
+	public void registersAndExecutesContinuousQueries() throws Exception {
+
+		ConfigurableApplicationContext applicationContext =
+			newApplicationContext(TestContinuousQueryRegistrationAndExecutionConfiguration.class);
+
+		assertThat(applicationContext).isNotNull();
+		assertThat(applicationContext.containsBean("DEFAULT")).isTrue();
+
+		GemFireCache gemfireCache = applicationContext.getBean(ClientCache.class);
+
+		assertThat(gemfireCache).isNotNull();
+
+		QueryService mockQueryService = gemfireCache.getQueryService();
+
+		assertThat(mockQueryService).isNotNull();
+		assertThat(mockQueryService.getCqs()).hasSize(1);
+
+		CqQuery mockCqQuery = mockQueryService.getCqs()[0];
+
+		assertThat(mockCqQuery).isNotNull();
+		assertThat(mockCqQuery.getName()).isEqualTo("TestQuery");
+		assertThat(mockCqQuery.getQueryString()).isEqualTo("SELECT * FROM /Example");
+		assertThat(mockCqQuery.isRunning()).isTrue();
+
+		verify(mockQueryService, times(1)).newCq(eq("TestQuery"),
+			eq("SELECT * FROM /Example"), any(CqAttributes.class), eq(false));
+
+		verify(mockCqQuery, times(1)).execute();
+	}
+
+	@ClientCacheApplication
+	@EnableContinuousQueries
+	@EnableGemFireMockObjects
+	@SuppressWarnings("unused")
+	static class TestContinuousQueryRegistrationAndExecutionConfiguration {
+
+		@Bean
+		BeanFactoryPostProcessor dependsOnBeanFactoryPostProcessor() {
+
+			return configurableListableBeanFactory -> {
+				configurableListableBeanFactory.getBeanDefinition("continuousQueryListenerContainer")
+					.setDependsOn("testComponent");
+			};
+		}
+
+		@Bean("DEFAULT")
+		Pool mockPool() {
+			return mock(Pool.class);
+		}
+
+		@Bean
+		@Order(Ordered.HIGHEST_PRECEDENCE)
+		TestContinuousQueryComponent testComponent() {
+			return new TestContinuousQueryComponent();
+		}
+	}
+
+	@Component
+	@SuppressWarnings("unused")
+	static class TestContinuousQueryComponent {
+
+		@ContinuousQuery(name = "TestQuery", query = "SELECT * FROM /Example")
+		public void handle(CqEvent event) {
+		}
+	}
+}

src/test/java/org/springframework/data/gemfire/test/mock/MockGemFireObjectsSupport.java

@@ -38,6 +38,7 @@ import java.io.InputStream;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -75,13 +76,22 @@ import org.apache.geode.cache.client.ClientRegionShortcut;
 import org.apache.geode.cache.client.Pool;
 import org.apache.geode.cache.client.PoolFactory;
 import org.apache.geode.cache.control.ResourceManager;
+import org.apache.geode.cache.execute.RegionFunctionContext;
+import org.apache.geode.cache.query.CqAttributes;
+import org.apache.geode.cache.query.CqQuery;
+import org.apache.geode.cache.query.Query;
+import org.apache.geode.cache.query.QueryService;
+import org.apache.geode.cache.query.QueryStatistics;
 import org.apache.geode.cache.server.CacheServer;
 import org.apache.geode.cache.server.ClientSubscriptionConfig;
 import org.apache.geode.compression.Compressor;
 import org.apache.geode.distributed.DistributedSystem;
+import org.apache.geode.internal.concurrent.ConcurrentHashSet;
 import org.apache.geode.pdx.PdxSerializer;
 import org.mockito.ArgumentMatchers;
+import org.mockito.stubbing.Answer;
 import org.springframework.data.gemfire.server.SubscriptionEvictionPolicy;
+import org.springframework.data.gemfire.test.mock.support.MockObjectInvocationException;
 import org.springframework.data.gemfire.test.support.FileSystemUtils;
 
 /**
@@ -117,6 +127,9 @@ public abstract class MockGemFireObjectsSupport extends MockObjectsSupport {
 
 	private static final Map<String, RegionAttributes<Object, Object>> regionAttributes = new ConcurrentHashMap<>();
 
+	private static final String FROM_KEYWORD = "FROM";
+	private static final String WHERE_KEYWORD = "WHERE";
+
 	private static final String REPEATING_REGION_SEPARATOR = Region.SEPARATOR + "{2,}";
 
 	public static void destroy() {
@@ -247,14 +260,17 @@ public abstract class MockGemFireObjectsSupport extends MockObjectsSupport {
 
 		doAnswer(newVoidAnswer(invocation -> mockClientCache.close())).when(mockClientCache).close(anyBoolean());
 
-		return mockClientRegionFactory(mockCacheApi(mockClientCache));
+		when(mockClientCache.createClientRegionFactory(any(ClientRegionShortcut.class)))
+			.thenAnswer(invocation -> mockClientRegionFactory(mockClientCache));
+
+		return mockQueryService(mockCacheApi(mockClientCache));
 	}
 
 	public static GemFireCache mockGemFireCache() {
 
 		GemFireCache mockGemFireCache = mock(GemFireCache.class);
 
-		return mockCacheApi(mockGemFireCache);
+		return mockQueryService(mockCacheApi(mockGemFireCache));
 	}
 
 	public static Cache mockPeerCache() {
@@ -290,7 +306,7 @@ public abstract class MockGemFireObjectsSupport extends MockObjectsSupport {
 		when(mockCache.getReconnectedCache()).thenAnswer(invocation -> mockPeerCache());
 		when(mockCache.getSearchTimeout()).thenAnswer(newGetter(searchTimeout));
 
-		return mockCacheApi(mockCache);
+		return mockQueryService(mockCacheApi(mockCache));
 	}
 
 	public static CacheServer mockCacheServer() {
@@ -365,14 +381,11 @@ public abstract class MockGemFireObjectsSupport extends MockObjectsSupport {
 	}
 
 	@SuppressWarnings("unchecked")
-	public static <K, V> ClientCache mockClientRegionFactory(ClientCache mockClientCache) {
+	public static <K, V> ClientRegionFactory<K, V> mockClientRegionFactory(ClientCache mockClientCache) {
 
 		ClientRegionFactory<K, V> mockClientRegionFactory =
 			mock(ClientRegionFactory.class, mockObjectIdentifier("MockClientRegionFactory"));
 
-		when(mockClientCache.<K, V>createClientRegionFactory(any(ClientRegionShortcut.class)))
-			.thenReturn(mockClientRegionFactory);
-
 		ExpirationAttributes DEFAULT_EXPIRATION_ATTRIBUTES =
 			new ExpirationAttributes(0, ExpirationAction.INVALIDATE);
 
@@ -486,7 +499,7 @@ public abstract class MockGemFireObjectsSupport extends MockObjectsSupport {
 		when(mockClientRegionFactory.createSubregion(any(Region.class), anyString())).thenAnswer(invocation ->
 			mockSubRegion(invocation.getArgument(0), invocation.getArgument(1), mockRegionAttributes));
 
-		return mockClientCache;
+		return mockClientRegionFactory;
 	}
 
 	public static ClientSubscriptionConfig mockClientSubscriptionConfig() {
@@ -775,6 +788,180 @@ public abstract class MockGemFireObjectsSupport extends MockObjectsSupport {
 		return mockPoolFactory;
 	}
 
+	public static Pool mockQueryService(Pool pool) {
+
+		QueryService mockQueryService = mockQueryService();
+
+		when(pool.getQueryService()).thenReturn(mockQueryService);
+
+		return pool;
+	}
+
+	public static <T extends RegionService> T mockQueryService(T regionService) {
+
+		QueryService mockQueryService = mockQueryService();
+
+		when(regionService.getQueryService()).thenReturn(mockQueryService);
+
+		if (regionService instanceof ClientCache) {
+			when(((ClientCache) regionService).getLocalQueryService()).thenReturn(mockQueryService);
+		}
+
+		return regionService;
+	}
+
+	// TODO write more mocking logic for the QueryService interface
+	public static QueryService mockQueryService() {
+
+		QueryService mockQueryService = mock(QueryService.class);
+
+		Set<CqQuery> cqQueries = new ConcurrentHashSet<>();
+
+		try {
+			when(mockQueryService.getCqs()).thenAnswer(invocation -> cqQueries.toArray(new CqQuery[cqQueries.size()]));
+
+			when(mockQueryService.getCq(anyString())).thenAnswer(invocation ->
+				cqQueries.stream().filter(cqQuery -> invocation.getArgument(0).equals(cqQuery.getName()))
+					.findFirst().orElse(null));
+
+			when(mockQueryService.getCqs(anyString())).thenAnswer(invocation -> {
+
+				List<CqQuery> cqQueriesByRegion = cqQueries.stream().filter(cqQuery -> {
+
+					String queryString = cqQuery.getQueryString();
+
+					int indexOfFromClause = queryString.indexOf(FROM_KEYWORD);
+					int indexOfWhereClause = queryString.indexOf(WHERE_KEYWORD);
+
+					queryString = (indexOfFromClause > -1
+						? queryString.substring(indexOfFromClause + FROM_KEYWORD.length()) : queryString);
+
+					queryString = (indexOfWhereClause > 0 ? queryString.substring(0, indexOfWhereClause) : queryString);
+
+					queryString = (queryString.startsWith(Region.SEPARATOR) ? queryString.substring(1) : queryString);
+
+					return invocation.getArgument(0).equals(queryString.trim());
+
+				}).collect(Collectors.toList());
+
+				return cqQueriesByRegion.toArray(new CqQuery[cqQueriesByRegion.size()]);
+			});
+
+			when(mockQueryService.newCq(anyString(), any(CqAttributes.class))).thenAnswer(invocation ->
+				add(cqQueries, mockCqQuery(null, invocation.getArgument(0), invocation.getArgument(1),
+					false)));
+
+			when(mockQueryService.newCq(anyString(), any(CqAttributes.class), anyBoolean())).thenAnswer(invocation ->
+				add(cqQueries, mockCqQuery(null, invocation.getArgument(0), invocation.getArgument(1),
+					invocation.getArgument(2))));
+
+			when(mockQueryService.newCq(anyString(), anyString(), any(CqAttributes.class))).thenAnswer(invocation ->
+				add(cqQueries, mockCqQuery(invocation.getArgument(0), invocation.getArgument(1),
+					invocation.getArgument(2), false)));
+
+			when(mockQueryService.newCq(anyString(), anyString(), any(CqAttributes.class), anyBoolean()))
+				.thenAnswer(invocation -> add(cqQueries, mockCqQuery(invocation.getArgument(0),
+					invocation.getArgument(1), invocation.getArgument(2), invocation.getArgument(3))));
+		}
+		catch (Exception cause) {
+			throw new MockObjectInvocationException(cause);
+		}
+
+		return mockQueryService;
+	}
+
+	private static CqQuery add(Collection<CqQuery> cqQueries, CqQuery cqQuery) {
+		cqQueries.add(cqQuery);
+		return cqQuery;
+	}
+
+	private static CqQuery mockCqQuery(String name, String queryString, CqAttributes cqAttributes, boolean durable) {
+
+		CqQuery mockCqQuery = mock(CqQuery.class);
+
+		Query mockQuery = mockQuery(queryString);
+
+		AtomicBoolean closed = new AtomicBoolean(false);
+		AtomicBoolean running = new AtomicBoolean(false);
+		AtomicBoolean stopped = new AtomicBoolean(true);
+
+		when(mockCqQuery.getCqAttributes()).thenReturn(cqAttributes);
+		when(mockCqQuery.getName()).thenReturn(name);
+		when(mockCqQuery.getQuery()).thenReturn(mockQuery);
+		when(mockCqQuery.getQueryString()).thenReturn(queryString);
+
+		try {
+			doAnswer(newSetter(closed, true, null)).when(mockCqQuery).close();
+
+			doAnswer(invocation -> {
+
+				running.set(true);
+				stopped.set(false);
+
+				return null;
+
+			}).when(mockCqQuery).execute();
+
+			doAnswer(invocation -> {
+
+				running.set(false);
+				stopped.set(true);
+
+				return null;
+
+			}).when(mockCqQuery).stop();
+		}
+		catch (Exception cause) {
+			throw new MockObjectInvocationException(cause);
+		}
+
+		when(mockCqQuery.isClosed()).thenAnswer(newGetter(closed));
+		when(mockCqQuery.isDurable()).thenReturn(durable);
+		when(mockCqQuery.isRunning()).thenAnswer(newGetter(running));
+		when(mockCqQuery.isStopped()).thenAnswer(newGetter(stopped));
+
+		return mockCqQuery;
+	}
+
+	private static Query mockQuery(String queryString) {
+
+		Query mockQuery = mock(Query.class);
+
+		QueryStatistics mockQueryStatistics = mockQueryStatistics(mockQuery);
+
+		when(mockQuery.getQueryString()).thenReturn(queryString);
+		when(mockQuery.getStatistics()).thenReturn(mockQueryStatistics);
+
+		return mockQuery;
+	}
+
+	private static QueryStatistics mockQueryStatistics(Query query) {
+
+		QueryStatistics mockQueryStatistics = mock(QueryStatistics.class);
+
+		AtomicLong numberOfExecutions = new AtomicLong(0L);
+
+		Answer<Object> executeAnswer = invocation -> {
+			numberOfExecutions.incrementAndGet();
+			return null;
+		};
+
+		try {
+			when(query.execute()).thenAnswer(executeAnswer);
+			when(query.execute(any(Object[].class))).thenAnswer(executeAnswer);
+			when(query.execute(any(RegionFunctionContext.class))).thenAnswer(executeAnswer);
+			when(query.execute(any(RegionFunctionContext.class), any(Object[].class))).thenAnswer(executeAnswer);
+		}
+		catch (Exception cause) {
+			throw new MockObjectInvocationException(cause);
+		}
+
+		when(mockQueryStatistics.getNumExecutions()).thenAnswer(newGetter(numberOfExecutions));
+		when(mockQueryStatistics.getTotalExecutionTime()).thenReturn(0L);
+
+		return mockQueryStatistics;
+	}
+
 	@SuppressWarnings("unchecked")
 	public static <K, V> Region<K, V> mockRegion(RegionService regionService, String name,
 			RegionAttributes<K, V> regionAttributes) {

src/test/java/org/springframework/data/gemfire/test/mock/MockObjectsSupport.java

@@ -35,6 +35,8 @@ import org.springframework.util.StringUtils;
  * used in mocking using Mockito.
  *
  * @author John Blum
+ * @see org.mockito.invocation.InvocationOnMock
+ * @see org.mockito.stubbing.Answer
  * @since 2.0.0
  */
 @SuppressWarnings("unused")

src/test/java/org/springframework/data/gemfire/test/mock/support/MockObjectInvocationException.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.gemfire.test.mock.support;
+
+import java.lang.reflect.Method;
+
+/**
+ * The {@link MockObjectInvocationException} class is an extension of {@link MockObjectsException} to categorize
+ * problems with {@link Method method} invocations on {@link Object Mock Objects}.
+ *
+ * @author John Blum
+ * @see org.springframework.data.gemfire.test.mock.support.MockObjectsException
+ * @since 2.0.1
+ */
+@SuppressWarnings("unused")
+public class MockObjectInvocationException extends MockObjectsException {
+
+	/**
+	 * Constructs a new instance of the {@link MockObjectInvocationException} class with no message or underlying cause.
+	 */
+	public MockObjectInvocationException() {
+	}
+
+	/**
+	 * Constructs a new instance of the {@link MockObjectInvocationException} class initialized with
+	 * the given {@link String message} describing the problem.
+	 *
+	 * @param message {@link String} describing the problem.
+	 * @see java.lang.String
+	 */
+	public MockObjectInvocationException(String message) {
+		super(message);
+	}
+
+	/**
+	 * Constructs a new instance of the {@link MockObjectInvocationException} class initialized with
+	 * the given {@link Throwable cause} of the underlying problem.
+	 *
+	 * @param cause {@link Throwable} object containing the cause of this exception.
+	 * @see java.lang.Throwable
+	 */
+	public MockObjectInvocationException(Throwable cause) {
+		super(cause);
+	}
+
+	/**
+	 * Constructs a new instance of the {@link MockObjectInvocationException} class initialized with
+	 * the given {@link String message} describing the underlying problem as well as the {@link Throwable cause}
+	 * of the underlying problem.
+	 *
+	 * @param message {@link String} describing the problem.
+	 * @param cause {@link Throwable} object containing the cause of this exception.
+	 * @see java.lang.Throwable
+	 * @see java.lang.String
+	 */
+	public MockObjectInvocationException(String message, Throwable cause) {
+		super(message, cause);
+	}
+}

src/test/java/org/springframework/data/gemfire/test/mock/support/MockObjectsException.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2017 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.gemfire.test.mock.support;
+
+/**
+ * The {@link MockObjectsException} class is a {@link RuntimeException} indicating a general problem
+ * with the Mock Objects infrastructure.
+ *
+ * @author John Blum
+ * @see java.lang.RuntimeException
+ * @since 2.0.1
+ */
+@SuppressWarnings("unused")
+public class MockObjectsException extends RuntimeException {
+
+	/**
+	 * Constructs a new instance of the {@link MockObjectsException} class with no message or underlying cause.
+	 */
+	public MockObjectsException() {
+	}
+
+	/**
+	 * Constructs a new instance of the {@link MockObjectsException} class initialized with
+	 * the given {@link String message} describing the problem.
+	 *
+	 * @param message {@link String} describing the problem.
+	 * @see java.lang.String
+	 */
+	public MockObjectsException(String message) {
+		super(message);
+	}
+
+	/**
+	 * Constructs a new instance of the {@link MockObjectsException} class initialized with
+	 * the given {@link Throwable cause} of the underlying problem.
+	 *
+	 * @param cause {@link Throwable} object containing the cause of this exception.
+	 * @see java.lang.Throwable
+	 */
+	public MockObjectsException(Throwable cause) {
+		super(cause);
+	}
+
+	/**
+	 * Constructs a new instance of the {@link MockObjectsException} class initialized with
+	 * the given {@link String message} describing the underlying problem as well as the {@link Throwable cause}
+	 * of the underlying problem.
+	 *
+	 * @param message {@link String} describing the problem.
+	 * @param cause {@link Throwable} object containing the cause of this exception.
+	 * @see java.lang.Throwable
+	 * @see java.lang.String
+	 */
+	public MockObjectsException(String message, Throwable cause) {
+		super(message, cause);
+	}
+}