You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by nt...@apache.org on 2016/04/22 17:42:47 UTC
[1/4] ignite git commit: IGNITE-2004 Fixed "Asynchronous execution of
ContinuousQuery's remote filter & local list".
Repository: ignite
Updated Branches:
refs/heads/master a6de16db8 -> 2ff64c2aa
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOrderingEventTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOrderingEventTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOrderingEventTest.java
new file mode 100644
index 0000000..7d975f2
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOrderingEventTest.java
@@ -0,0 +1,722 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.cache.configuration.FactoryBuilder;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryUpdatedListener;
+import javax.cache.processor.EntryProcessorException;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.CacheEntryProcessor;
+import org.apache.ignite.cache.CacheMemoryMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.PA;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.spi.eventstorage.memory.MemoryEventStorageSpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+
+import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.PRIMARY;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_TIERED;
+import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_VALUES;
+import static org.apache.ignite.cache.CacheMemoryMode.ONHEAP_TIERED;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC;
+
+/**
+ *
+ */
+public class CacheContinuousQueryOrderingEventTest extends GridCommonAbstractTest {
+ /** */
+ public static final int LISTENER_CNT = 3;
+
+ /** */
+ public static final int KEYS = 10;
+
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int NODES = 5;
+
+ /** */
+ public static final int ITERATION_CNT = 100;
+
+ /** */
+ private boolean client;
+
+ /** */
+ private static volatile boolean fail;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+ ((TcpCommunicationSpi)cfg.getCommunicationSpi()).setSharedMemoryPort(-1);
+
+ cfg.setClientMode(client);
+
+ MemoryEventStorageSpi storeSpi = new MemoryEventStorageSpi();
+ storeSpi.setExpireCount(100);
+
+ cfg.setEventStorageSpi(storeSpi);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGridsMultiThreaded(NODES - 1);
+
+ client = true;
+
+ startGrid(NODES - 1);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ super.afterTestsStopped();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ fail = false;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOnheapTwoBackup() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ ONHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOffheapTwoBackup() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ OFFHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOffheapValuesTwoBackup() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ OFFHEAP_VALUES, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicReplicatedOffheap() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 0, ATOMIC,
+ OFFHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxOnheapTwoBackup() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL,
+ ONHEAP_TIERED, FULL_SYNC);
+
+ doOrderingTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxOnheapWithoutBackup() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 0, TRANSACTIONAL,
+ ONHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxOnheapWithoutBackupFullSync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 0, TRANSACTIONAL,
+ ONHEAP_TIERED, FULL_SYNC);
+
+ doOrderingTest(ccfg, false);
+ }
+
+ // ASYNC
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOnheapTwoBackupAsync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ ONHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOnheapTwoBackupAsyncFullSync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ ONHEAP_TIERED, FULL_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOffheapTwoBackupAsync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ OFFHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOffheapTwoBackupAsyncFullSync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ OFFHEAP_TIERED, FULL_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOffheapValuesTwoBackupAsync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ OFFHEAP_VALUES, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOffheapValuesTwoBackupAsyncFullSync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC,
+ OFFHEAP_VALUES, FULL_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicReplicatedAsync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 0, ATOMIC,
+ ONHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicReplicatedAsyncFullSync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 0, ATOMIC,
+ ONHEAP_TIERED, FULL_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicReplicatedOffheapAsync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 0, ATOMIC,
+ OFFHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOnheapWithoutBackupAsync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 0, ATOMIC,
+ ONHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxOnheapTwoBackupAsync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL,
+ ONHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxOnheapAsync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 0, TRANSACTIONAL,
+ ONHEAP_TIERED, PRIMARY_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxOnheapAsyncFullSync() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 0, TRANSACTIONAL,
+ ONHEAP_TIERED, FULL_SYNC);
+
+ doOrderingTest(ccfg, true);
+ }
+
+ /**
+ * @param ccfg Cache configuration.
+ * @param async Async filter.
+ * @throws Exception If failed.
+ */
+ protected void doOrderingTest(
+ final CacheConfiguration ccfg,
+ final boolean async)
+ throws Exception {
+ ignite(0).createCache(ccfg);
+
+ List<QueryCursor<?>> qries = new ArrayList<>();
+
+ try {
+ List<BlockingQueue<CacheEntryEvent<QueryTestKey, QueryTestValue>>> rcvdEvts =
+ new ArrayList<>(LISTENER_CNT * NODES);
+
+ final AtomicInteger qryCntr = new AtomicInteger(0);
+
+ final int threadCnt = 20;
+
+ for (int idx = 0; idx < NODES; idx++) {
+ for (int i = 0; i < LISTENER_CNT; i++) {
+ BlockingQueue<CacheEntryEvent<QueryTestKey, QueryTestValue>> queue =
+ new ArrayBlockingQueue<>(ITERATION_CNT * threadCnt);
+
+ ContinuousQuery qry = new ContinuousQuery();
+
+ if (async) {
+ qry.setLocalListener(new TestCacheAsyncEventListener(queue, qryCntr));
+
+ qry.setRemoteFilterFactory(FactoryBuilder.factoryOf(
+ new CacheTestRemoteFilterAsync(ccfg.getName())));
+ }
+ else {
+ qry.setLocalListener(new TestCacheEventListener(queue, qryCntr));
+
+ qry.setRemoteFilterFactory(FactoryBuilder.factoryOf(
+ new CacheTestRemoteFilter(ccfg.getName())));
+ }
+
+ rcvdEvts.add(queue);
+
+ IgniteCache<Object, Object> cache = grid(idx).cache(ccfg.getName());
+
+ QueryCursor qryCursor = cache.query(qry);
+
+ qries.add(qryCursor);
+ }
+ }
+
+ IgniteInternalFuture<Long> f = GridTestUtils.runMultiThreadedAsync(new Runnable() {
+ @Override public void run() {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (int i = 0; i < ITERATION_CNT; i++) {
+ IgniteCache<QueryTestKey, QueryTestValue> cache =
+ grid(rnd.nextInt(NODES)).cache(ccfg.getName());
+
+ QueryTestKey key = new QueryTestKey(rnd.nextInt(KEYS));
+
+ boolean startTx = cache.getConfiguration(CacheConfiguration.class).getAtomicityMode() ==
+ TRANSACTIONAL && rnd.nextBoolean();
+
+ Transaction tx = null;
+
+ if (startTx)
+ tx = cache.unwrap(Ignite.class).transactions().txStart();
+
+ try {
+ if ((cache.get(key) == null) || rnd.nextBoolean()) {
+ cache.invoke(key, new CacheEntryProcessor<QueryTestKey, QueryTestValue, Object>() {
+ @Override public Object process(
+ MutableEntry<QueryTestKey, QueryTestValue> entry,
+ Object... arguments)
+ throws EntryProcessorException {
+ if (entry.exists())
+ entry.setValue(new QueryTestValue(entry.getValue().val1 + 1));
+ else
+ entry.setValue(new QueryTestValue(0));
+
+ return null;
+ }
+ });
+ }
+ else {
+ QueryTestValue val;
+ QueryTestValue newVal;
+
+ do {
+ val = cache.get(key);
+
+ newVal = val == null ?
+ new QueryTestValue(0) : new QueryTestValue(val.val1 + 1);
+ }
+ while (!cache.replace(key, val, newVal));
+ }
+ }
+ finally {
+ if (tx != null)
+ tx.commit();
+ }
+ }
+ }
+ }, threadCnt, "put-thread");
+
+ f.get(15, TimeUnit.SECONDS);
+
+ GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return qryCntr.get() >= ITERATION_CNT * threadCnt * LISTENER_CNT * NODES;
+ }
+ }, 1000L);
+
+ for (BlockingQueue<CacheEntryEvent<QueryTestKey, QueryTestValue>> queue : rcvdEvts)
+ checkEvents(queue, ITERATION_CNT * threadCnt);
+
+ assertFalse("Ordering invocations of filter broken.", fail);
+ }
+ finally {
+ for (QueryCursor<?> qry : qries)
+ qry.close();
+
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @param queue Event queue.
+ * @throws Exception If failed.
+ */
+ private void checkEvents(BlockingQueue<CacheEntryEvent<QueryTestKey, QueryTestValue>> queue, int expCnt)
+ throws Exception {
+ CacheEntryEvent<QueryTestKey, QueryTestValue> evt;
+ int cnt = 0;
+ Map<QueryTestKey, Integer> vals = new HashMap<>();
+
+ while ((evt = queue.poll(100, TimeUnit.MILLISECONDS)) != null) {
+ assertNotNull(evt);
+ assertNotNull(evt.getKey());
+
+ Integer preVal = vals.get(evt.getKey());
+
+ if (preVal == null)
+ assertEquals(new QueryTestValue(0), evt.getValue());
+ else {
+ if (!new QueryTestValue(preVal + 1).equals(evt.getValue()))
+ assertEquals("Key event: " + evt.getKey(), new QueryTestValue(preVal + 1), evt.getValue());
+ }
+
+ vals.put(evt.getKey(), evt.getValue().val1);
+
+ ++cnt;
+ }
+
+ assertEquals(expCnt, cnt);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return TimeUnit.MINUTES.toMillis(8);
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ private static class CacheTestRemoteFilterAsync extends CacheTestRemoteFilter {
+ /**
+ * @param cacheName Cache name.
+ */
+ public CacheTestRemoteFilterAsync(String cacheName) {
+ super(cacheName);
+ }
+ }
+
+ /**
+ *
+ */
+ private static class CacheTestRemoteFilter implements
+ CacheEntryEventSerializableFilter<QueryTestKey, QueryTestValue> {
+ /** */
+ private Map<QueryTestKey, QueryTestValue> prevVals = new ConcurrentHashMap<>();
+
+ /** */
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** */
+ private String cacheName;
+
+ /**
+ * @param cacheName Cache name.
+ */
+ public CacheTestRemoteFilter(String cacheName) {
+ this.cacheName = cacheName;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e) {
+ if (affinity(ignite.cache(cacheName)).isPrimary(ignite.cluster().localNode(), e.getKey())) {
+ QueryTestValue prevVal = prevVals.put(e.getKey(), e.getValue());
+
+ if (prevVal != null) {
+ if (!new QueryTestValue(prevVal.val1 + 1).equals(e.getValue()))
+ fail = true;
+ }
+ }
+
+ return true;
+ }
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ private static class TestCacheAsyncEventListener extends TestCacheEventListener {
+ /**
+ * @param queue Queue.
+ * @param cntr Received events counter.
+ */
+ public TestCacheAsyncEventListener(BlockingQueue<CacheEntryEvent<QueryTestKey, QueryTestValue>> queue,
+ AtomicInteger cntr) {
+ super(queue, cntr);
+ }
+ }
+
+ /**
+ *
+ */
+ private static class TestCacheEventListener implements CacheEntryUpdatedListener<QueryTestKey, QueryTestValue> {
+ /** */
+ private final BlockingQueue<CacheEntryEvent<QueryTestKey, QueryTestValue>> queue;
+
+ /** */
+ private final AtomicInteger cntr;
+
+ /**
+ * @param queue Queue.
+ * @param cntr Received events counter.
+ */
+ public TestCacheEventListener(BlockingQueue<CacheEntryEvent<QueryTestKey, QueryTestValue>> queue,
+ AtomicInteger cntr) {
+ this.queue = queue;
+ this.cntr = cntr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> evts) {
+ for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : evts) {
+ queue.add((CacheEntryEvent<QueryTestKey, QueryTestValue>)e);
+
+ cntr.incrementAndGet();
+ }
+ }
+ }
+
+ /**
+ * @param cacheMode Cache mode.
+ * @param backups Number of backups.
+ * @param atomicityMode Cache atomicity mode.
+ * @param memoryMode Cache memory mode.
+ * @param writeMode Cache write mode.
+ * @return Cache configuration.
+ */
+ protected CacheConfiguration<Object, Object> cacheConfiguration(
+ CacheMode cacheMode,
+ int backups,
+ CacheAtomicityMode atomicityMode,
+ CacheMemoryMode memoryMode,
+ CacheWriteSynchronizationMode writeMode) {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
+
+ ccfg.setName("test-cache-" + atomicityMode + "-" + cacheMode + "-" + memoryMode + "-" + memoryMode + "-"
+ + backups);
+ ccfg.setAtomicityMode(atomicityMode);
+ ccfg.setCacheMode(cacheMode);
+ ccfg.setMemoryMode(memoryMode);
+ ccfg.setWriteSynchronizationMode(writeMode);
+ ccfg.setAtomicWriteOrderMode(PRIMARY);
+
+ if (cacheMode == PARTITIONED)
+ ccfg.setBackups(backups);
+
+ return ccfg;
+ }
+
+ /**
+ *
+ */
+ public static class QueryTestKey implements Serializable, Comparable {
+ /** */
+ private final Integer key;
+
+ /**
+ * @param key Key.
+ */
+ public QueryTestKey(Integer key) {
+ this.key = key;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ QueryTestKey that = (QueryTestKey)o;
+
+ return key.equals(that.key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return key.hashCode();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(QueryTestKey.class, this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int compareTo(Object o) {
+ return key - ((QueryTestKey)o).key;
+ }
+ }
+
+ /**
+ *
+ */
+ public static class QueryTestValue implements Serializable {
+ /** */
+ @GridToStringInclude
+ protected final Integer val1;
+
+ /** */
+ @GridToStringInclude
+ protected final String val2;
+
+ /**
+ * @param val Value.
+ */
+ public QueryTestValue(Integer val) {
+ this.val1 = val;
+ this.val2 = String.valueOf(val);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ QueryTestValue that = (QueryTestValue) o;
+
+ return val1.equals(that.val1) && val2.equals(that.val2);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = val1.hashCode();
+
+ res = 31 * res + val2.hashCode();
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(QueryTestValue.class, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java
index e9fbf70..a6c33bb 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTest.java
@@ -62,6 +62,7 @@ import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.PA;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -118,6 +119,7 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
IgniteConfiguration cfg = super.getConfiguration(gridName);
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+ ((TcpCommunicationSpi)cfg.getCommunicationSpi()).setSharedMemoryPort(-1);
cfg.setClientMode(client);
@@ -598,6 +600,9 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
final List<CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> evts =
new CopyOnWriteArrayList<>();
+ if (noOpFilterFactory() != null)
+ qry.setRemoteFilterFactory(noOpFilterFactory());
+
qry.setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
@Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
? extends QueryTestValue>> events) throws CacheEntryListenerException {
@@ -684,9 +689,17 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
checkSingleEvent(evts.get(7), CREATED, new QueryTestValue(5), null);
checkSingleEvent(evts.get(8), EventType.UPDATED, new QueryTestValue(6), new QueryTestValue(5));
+ evts.clear();
+
cache.remove(key);
cache.remove(key);
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return evts.size() == 1;
+ }
+ }, 5_000);
+
evts.clear();
log.info("Finish iteration: " + i);
@@ -699,6 +712,13 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
}
/**
+ * @return No-op filter factory for batch operations.
+ */
+ protected Factory<? extends CacheEntryEventFilter<QueryTestKey, QueryTestValue>> noOpFilterFactory() {
+ return null;
+ }
+
+ /**
* @param ccfg Cache configuration.
* @throws Exception If failed.
*/
@@ -711,6 +731,9 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
final List<CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> evts =
new CopyOnWriteArrayList<>();
+ if (noOpFilterFactory() != null)
+ qry.setRemoteFilterFactory(noOpFilterFactory());
+
qry.setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
@Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
? extends QueryTestValue>> events) throws CacheEntryListenerException {
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
index b4c31be..e4afe73 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridTestKernalContext.java
@@ -60,6 +60,7 @@ public class GridTestKernalContext extends GridKernalContextImpl {
null,
null,
null,
+ null,
U.allPluginProviders());
GridTestUtils.setFieldValue(grid(), "cfg", config());
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
index 30625fe..761d4bd 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
@@ -44,7 +44,6 @@ public class IgniteBinaryCacheQueryTestSuite extends TestSuite {
suite.addTestSuite(BinarySerializationQuerySelfTest.class);
suite.addTestSuite(BinarySerializationQueryWithReflectiveSerializerSelfTest.class);
suite.addTestSuite(IgniteCacheBinaryObjectsScanSelfTest.class);
- suite.addTestSuite(CacheContinuousQueryLostPartitionTest.class);
//Should be adjusted. Not ready to be used with BinaryMarshaller.
//suite.addTestSuite(GridCacheBinarySwapScanQuerySelfTest.class);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite3.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite3.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite3.java
index fbb3091..e0e81b7 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite3.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite3.java
@@ -20,9 +20,14 @@ package org.apache.ignite.testsuites;
import junit.framework.TestSuite;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousBatchAckTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousBatchForceServerModeAckTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryAsyncFilterListenerTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryExecuteInPrimaryTest;
-import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFactoryFilterTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFactoryAsyncFilterRandomOperationTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFactoryFilterRandomOperationTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryLostPartitionTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryOperationFromCallbackTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryOperationP2PTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryOrderingEventTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryRandomOperationsTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryRandomOperationsTwoNodesTest;
import org.apache.ignite.internal.processors.cache.query.continuous.GridCacheContinuousQueryAtomicNearEnabledSelfTest;
@@ -87,12 +92,17 @@ public class IgniteCacheQuerySelfTestSuite3 extends TestSuite {
suite.addTestSuite(IgniteCacheContinuousQueryClientTxReconnectTest.class);
suite.addTestSuite(CacheContinuousQueryRandomOperationsTest.class);
suite.addTestSuite(CacheContinuousQueryRandomOperationsTwoNodesTest.class);
- suite.addTestSuite(CacheContinuousQueryFactoryFilterTest.class);
suite.addTestSuite(GridCacheContinuousQueryConcurrentTest.class);
+ suite.addTestSuite(CacheContinuousQueryAsyncFilterListenerTest.class);
+ suite.addTestSuite(CacheContinuousQueryFactoryFilterRandomOperationTest.class);
+ suite.addTestSuite(CacheContinuousQueryFactoryAsyncFilterRandomOperationTest.class);
+ suite.addTestSuite(CacheContinuousQueryOrderingEventTest.class);
+ suite.addTestSuite(CacheContinuousQueryOperationFromCallbackTest.class);
suite.addTestSuite(CacheContinuousQueryOperationP2PTest.class);
suite.addTestSuite(CacheContinuousBatchAckTest.class);
suite.addTestSuite(CacheContinuousBatchForceServerModeAckTest.class);
suite.addTestSuite(CacheContinuousQueryExecuteInPrimaryTest.class);
+ suite.addTestSuite(CacheContinuousQueryLostPartitionTest.class);
return suite;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite4.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite4.java
index fa4e642..c4fcdac 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite4.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite4.java
@@ -18,6 +18,9 @@
package org.apache.ignite.testsuites;
import junit.framework.TestSuite;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryAsyncFailoverAtomicPrimaryWriteOrderSelfTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryAsyncFailoverTxReplicatedSelfTest;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryAsyncFailoverTxSelfTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFailoverAtomicPrimaryWriteOrderOffheapTieredTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFailoverAtomicPrimaryWriteOrderSelfTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFailoverAtomicReplicatedSelfTest;
@@ -44,6 +47,10 @@ public class IgniteCacheQuerySelfTestSuite4 extends TestSuite {
suite.addTestSuite(CacheContinuousQueryFailoverAtomicPrimaryWriteOrderOffheapTieredTest.class);
suite.addTestSuite(CacheContinuousQueryFailoverTxOffheapTieredTest.class);
+ suite.addTestSuite(CacheContinuousQueryAsyncFailoverAtomicPrimaryWriteOrderSelfTest.class);
+ suite.addTestSuite(CacheContinuousQueryAsyncFailoverTxReplicatedSelfTest.class);
+ suite.addTestSuite(CacheContinuousQueryAsyncFailoverTxSelfTest.class);
+
return suite;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventAsyncProbe.java
----------------------------------------------------------------------
diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventAsyncProbe.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventAsyncProbe.java
new file mode 100644
index 0000000..0ea66d4
--- /dev/null
+++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventAsyncProbe.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.yardstick.cache;
+
+import java.util.concurrent.atomic.AtomicLong;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryUpdatedListener;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+
+/**
+ * Probe which calculate continuous query events.
+ */
+public class CacheEntryEventAsyncProbe extends CacheEntryEventProbe {
+ /** */
+ @Override protected CacheEntryUpdatedListener<Integer, Integer> localListener(AtomicLong cntr) {
+ return new CacheEntryEventListener(cntr);
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ private static final class CacheEntryEventListener implements CacheEntryUpdatedListener<Integer, Integer> {
+ /** */
+ private AtomicLong cnt;
+
+ /**
+ * @param cnt Counter.
+ */
+ public CacheEntryEventListener(AtomicLong cnt) {
+ this.cnt = cnt;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> events)
+ throws CacheEntryListenerException {
+ int size = 0;
+
+ for (CacheEntryEvent<? extends Integer, ? extends Integer> e : events)
+ ++size;
+
+ cnt.addAndGet(size);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventProbe.java
----------------------------------------------------------------------
diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventProbe.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventProbe.java
index e42479a..a25f975 100644
--- a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventProbe.java
+++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/CacheEntryEventProbe.java
@@ -45,9 +45,6 @@ public class CacheEntryEventProbe implements BenchmarkProbe {
/** */
private BenchmarkConfiguration cfg;
- /** Counter. */
- private AtomicLong cnt = new AtomicLong(0);
-
/** Collected points. */
private Collection<BenchmarkProbePoint> collected = new ArrayList<>();
@@ -67,17 +64,9 @@ public class CacheEntryEventProbe implements BenchmarkProbe {
if (drv0.cache() != null) {
ContinuousQuery<Integer, Integer> qry = new ContinuousQuery<>();
- qry.setLocalListener(new CacheEntryUpdatedListener<Integer, Integer>() {
- @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>>
- events) throws CacheEntryListenerException {
- int size = 0;
-
- for (CacheEntryEvent<? extends Integer, ? extends Integer> e : events)
- ++size;
+ final AtomicLong cnt = new AtomicLong();
- cnt.addAndGet(size);
- }
- });
+ qry.setLocalListener(localListener(cnt));
qryCur = drv0.cache().query(qry);
@@ -113,6 +102,24 @@ public class CacheEntryEventProbe implements BenchmarkProbe {
+ " probe. Probably, the driver doesn't provide \"cache()\" method.");
}
+ /**
+ * @param cntr Received event counter.
+ * @return Local listener.
+ */
+ protected CacheEntryUpdatedListener<Integer, Integer> localListener(final AtomicLong cntr) {
+ return new CacheEntryUpdatedListener<Integer, Integer>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends Integer>> events)
+ throws CacheEntryListenerException {
+ int size = 0;
+
+ for (CacheEntryEvent<? extends Integer, ? extends Integer> e : events)
+ ++size;
+
+ cntr.addAndGet(size);
+ }
+ };
+ }
+
/** {@inheritDoc} */
@Override public void stop() throws Exception {
if (qryCur != null) {
[4/4] ignite git commit: IGNITE-2004 Fixed "Asynchronous execution of
ContinuousQuery's remote filter & local list".
Posted by nt...@apache.org.
IGNITE-2004 Fixed "Asynchronous execution of ContinuousQuery's remote filter & local list".
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2ff64c2a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2ff64c2a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2ff64c2a
Branch: refs/heads/master
Commit: 2ff64c2aa8f142f2918200e5ee85e6b7e83afbfa
Parents: a6de16d
Author: nikolay_tikhonov <nt...@gridgain.com>
Authored: Fri Apr 22 18:41:58 2016 +0300
Committer: nikolay_tikhonov <nt...@gridgain.com>
Committed: Fri Apr 22 18:41:58 2016 +0300
----------------------------------------------------------------------
.../CacheContinuousAsyncQueryExample.java | 138 +++
.../datagrid/CacheContinuousQueryExample.java | 13 +-
.../ignite/cache/query/ContinuousQuery.java | 27 +
.../configuration/IgniteConfiguration.java | 32 +
.../ignite/internal/GridKernalContext.java | 8 +
.../ignite/internal/GridKernalContextImpl.java | 12 +
.../apache/ignite/internal/IgniteKernal.java | 3 +
.../org/apache/ignite/internal/IgnitionEx.java | 16 +-
.../processors/cache/GridCacheEntryEx.java | 5 +-
.../processors/cache/GridCacheMapEntry.java | 49 +-
.../dht/atomic/GridDhtAtomicCache.java | 98 +-
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 121 +--
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 33 +-
.../distributed/near/GridNearAtomicCache.java | 3 +-
.../continuous/CacheContinuousQueryEvent.java | 7 +
.../continuous/CacheContinuousQueryHandler.java | 446 +++++++--
.../CacheContinuousQueryListener.java | 6 +-
.../continuous/CacheContinuousQueryManager.java | 62 +-
.../apache/ignite/lang/IgniteAsyncCallback.java | 111 +++
.../thread/IgniteStripedThreadPoolExecutor.java | 164 +--
.../processors/cache/GridCacheTestEntryEx.java | 4 +-
...FailoverAtomicPrimaryWriteOrderSelfTest.java | 50 +
...sQueryAsyncFailoverTxReplicatedSelfTest.java | 37 +
...eContinuousQueryAsyncFailoverTxSelfTest.java | 44 +
...eContinuousQueryAsyncFilterListenerTest.java | 986 +++++++++++++++++++
...ryFactoryAsyncFilterRandomOperationTest.java | 131 +++
...usQueryFactoryFilterRandomOperationTest.java | 725 ++++++++++++++
.../CacheContinuousQueryFactoryFilterTest.java | 714 --------------
...ContinuousQueryFailoverAbstractSelfTest.java | 63 +-
.../CacheContinuousQueryLostPartitionTest.java | 14 +
...ontinuousQueryOperationFromCallbackTest.java | 627 ++++++++++++
.../CacheContinuousQueryOrderingEventTest.java | 722 ++++++++++++++
...acheContinuousQueryRandomOperationsTest.java | 23 +
.../junits/GridTestKernalContext.java | 1 +
.../IgniteBinaryCacheQueryTestSuite.java | 1 -
.../IgniteCacheQuerySelfTestSuite3.java | 14 +-
.../IgniteCacheQuerySelfTestSuite4.java | 7 +
.../cache/CacheEntryEventAsyncProbe.java | 61 ++
.../yardstick/cache/CacheEntryEventProbe.java | 33 +-
39 files changed, 4408 insertions(+), 1203 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousAsyncQueryExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousAsyncQueryExample.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousAsyncQueryExample.java
new file mode 100644
index 0000000..4ac7ecb
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousAsyncQueryExample.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.datagrid;
+
+import javax.cache.Cache;
+import javax.cache.configuration.Factory;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryUpdatedListener;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.examples.ExampleNodeStartup;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.resources.IgniteInstanceResource;
+
+/**
+ * This examples demonstrates asynchronous continuous query API.
+ * <p>
+ * Remote nodes should always be started with special configuration file which
+ * enables P2P class loading: {@code 'ignite.{sh|bat} examples/config/example-ignite.xml'}.
+ * <p>
+ * Alternatively you can run {@link ExampleNodeStartup} in another JVM which will
+ * start node with {@code examples/config/example-ignite.xml} configuration.
+ */
+public class CacheContinuousAsyncQueryExample {
+ /** Cache name. */
+ private static final String CACHE_NAME = CacheContinuousAsyncQueryExample.class.getSimpleName();
+
+ /**
+ * Executes example.
+ *
+ * @param args Command line arguments, none required.
+ * @throws Exception If example execution failed.
+ */
+ public static void main(String[] args) throws Exception {
+ try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
+ System.out.println();
+ System.out.println(">>> Cache continuous query example started.");
+
+ // Auto-close cache at the end of the example.
+ try (IgniteCache<Integer, String> cache = ignite.getOrCreateCache(CACHE_NAME)) {
+ int keyCnt = 20;
+
+ // These entries will be queried by initial predicate.
+ for (int i = 0; i < keyCnt; i++)
+ cache.put(i, Integer.toString(i));
+
+ // Create new continuous query.
+ ContinuousQuery<Integer, String> qry = new ContinuousQuery<>();
+
+ qry.setInitialQuery(new ScanQuery<>(new IgniteBiPredicate<Integer, String>() {
+ @Override public boolean apply(Integer key, String val) {
+ return key > 10;
+ }
+ }));
+
+ // Callback that is called locally when update notifications are received.
+ qry.setLocalListener(new CacheEntryUpdatedListener<Integer, String>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends Integer, ? extends String>> evts) {
+ for (CacheEntryEvent<? extends Integer, ? extends String> e : evts)
+ System.out.println("Updated entry [key=" + e.getKey() + ", val=" + e.getValue() + ']');
+ }
+ });
+
+ // This filter will be evaluated remotely on all nodes.
+ // Entry that pass this filter will be sent to the caller.
+ qry.setRemoteFilterFactory(new Factory<CacheEntryEventFilter<Integer, String>>() {
+ @Override public CacheEntryEventFilter<Integer, String> create() {
+ return new CacheEntryFilter();
+ }
+ });
+
+ // Execute query.
+ try (QueryCursor<Cache.Entry<Integer, String>> cur = cache.query(qry)) {
+ // Iterate through existing data.
+ for (Cache.Entry<Integer, String> e : cur)
+ System.out.println("Queried existing entry [key=" + e.getKey() + ", val=" + e.getValue() + ']');
+
+ // Add a few more keys and watch more query notifications.
+ for (int i = 0; i < keyCnt; i++)
+ cache.put(i, Integer.toString(i));
+
+ // Wait for a while while callback is notified about remaining puts.
+ Thread.sleep(2000);
+ }
+
+ // Iterate through entries which was updated from filter.
+ for (int i = 0; i < 10; i++)
+ System.out.println("Entry updated from filter [key=" + i + ", val=" + cache.get(i) + ']');
+ }
+ finally {
+ // Distributed cache could be removed from cluster only by #destroyCache() call.
+ ignite.destroyCache(CACHE_NAME);
+ }
+ }
+ }
+
+ /**
+ * Filter returns {@code true} for entries which have key bigger than 10.
+ */
+ @IgniteAsyncCallback
+ private static class CacheEntryFilter implements CacheEntryEventFilter<Integer, String> {
+ /** Ignite instance. */
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends String> e)
+ throws CacheEntryListenerException {
+ // This cache operation is safe because filter has Ignite AsyncCallback annotation.
+ if (e.getKey() < 10 && String.valueOf(e.getKey()).equals(e.getValue()))
+ ignite.cache(CACHE_NAME).put(e.getKey(), e.getValue() + "_less_than_10");
+
+ return e.getKey() > 10;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousQueryExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousQueryExample.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousQueryExample.java
index 59759af..aad5b5d 100644
--- a/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousQueryExample.java
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousQueryExample.java
@@ -18,12 +18,13 @@
package org.apache.ignite.examples.datagrid;
import javax.cache.Cache;
+import javax.cache.configuration.Factory;
import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
import javax.cache.event.CacheEntryUpdatedListener;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
-import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
import org.apache.ignite.cache.query.ContinuousQuery;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.query.ScanQuery;
@@ -81,9 +82,13 @@ public class CacheContinuousQueryExample {
// This filter will be evaluated remotely on all nodes.
// Entry that pass this filter will be sent to the caller.
- qry.setRemoteFilter(new CacheEntryEventSerializableFilter<Integer, String>() {
- @Override public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends String> e) {
- return e.getKey() > 10;
+ qry.setRemoteFilterFactory(new Factory<CacheEntryEventFilter<Integer, String>>() {
+ @Override public CacheEntryEventFilter<Integer, String> create() {
+ return new CacheEntryEventFilter<Integer, String>() {
+ @Override public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends String> e) {
+ return e.getKey() > 10;
+ }
+ };
}
});
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java b/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java
index 3ea8f93..bbfe8cc 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/query/ContinuousQuery.java
@@ -23,6 +23,8 @@ import javax.cache.event.CacheEntryEventFilter;
import javax.cache.event.CacheEntryUpdatedListener;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.lang.IgniteAsyncCallback;
/**
* API for configuring continuous cache queries.
@@ -92,6 +94,16 @@ import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
* Note that this works even if you didn't provide initial query. Cursor will
* be empty in this case, but it will still unregister listeners when {@link QueryCursor#close()}
* is called.
+ * <p>
+ * {@link IgniteAsyncCallback} annotation is supported for {@link CacheEntryEventFilter}
+ * (see {@link #setRemoteFilterFactory(Factory)}) and {@link CacheEntryUpdatedListener}
+ * (see {@link #setLocalListener(CacheEntryUpdatedListener)}).
+ * If filter and/or listener are annotated with {@link IgniteAsyncCallback} then annotated callback
+ * is executed in async callback pool (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()})
+ * and notification order is kept the same as update order for given cache key.
+ *
+ * @see IgniteAsyncCallback
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize()
*/
public final class ContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
/** */
@@ -173,9 +185,14 @@ public final class ContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
* <b>WARNING:</b> all operations that involve any kind of JVM-local or distributed locking (e.g.,
* synchronization or transactional cache operations), should be executed asynchronously without
* blocking the thread that called the callback. Otherwise, you can get deadlocks.
+ * <p>
+ * If local listener are annotated with {@link IgniteAsyncCallback} then it is executed in async callback pool
+ * (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()}) that allow to perform a cache operations.
*
* @param locLsnr Local callback.
* @return {@code this} for chaining.
+ * @see IgniteAsyncCallback
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize()
*/
public ContinuousQuery<K, V> setLocalListener(CacheEntryUpdatedListener<K, V> locLsnr) {
this.locLsnr = locLsnr;
@@ -198,11 +215,16 @@ public final class ContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
* <b>WARNING:</b> all operations that involve any kind of JVM-local or distributed locking
* (e.g., synchronization or transactional cache operations), should be executed asynchronously
* without blocking the thread that called the filter. Otherwise, you can get deadlocks.
+ * <p>
+ * If remote filter are annotated with {@link IgniteAsyncCallback} then it is executed in async callback
+ * pool (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()}) that allow to perform a cache operations.
*
* @param rmtFilter Key-value filter.
* @return {@code this} for chaining.
*
* @deprecated Use {@link #setRemoteFilterFactory(Factory)} instead.
+ * @see IgniteAsyncCallback
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize()
*/
@Deprecated
public ContinuousQuery<K, V> setRemoteFilter(CacheEntryEventSerializableFilter<K, V> rmtFilter) {
@@ -227,9 +249,14 @@ public final class ContinuousQuery<K, V> extends Query<Cache.Entry<K, V>> {
* <b>WARNING:</b> all operations that involve any kind of JVM-local or distributed locking
* (e.g., synchronization or transactional cache operations), should be executed asynchronously
* without blocking the thread that called the filter. Otherwise, you can get deadlocks.
+ * <p>
+ * If remote filter are annotated with {@link IgniteAsyncCallback} then it is executed in async callback
+ * pool (see {@link IgniteConfiguration#getAsyncCallbackPoolSize()}) that allow to perform a cache operations.
*
* @param rmtFilterFactory Key-value filter factory.
* @return {@code this} for chaining.
+ * @see IgniteAsyncCallback
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize()
*/
public ContinuousQuery<K, V> setRemoteFilterFactory(
Factory<? extends CacheEntryEventFilter<K, V>> rmtFilterFactory) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
index e5bd05c..19b9a4d 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
@@ -43,6 +43,7 @@ import org.apache.ignite.events.Event;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteAsyncCallback;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lifecycle.LifecycleBean;
@@ -223,6 +224,9 @@ public class IgniteConfiguration {
/** Public pool size. */
private int pubPoolSize = DFLT_PUBLIC_THREAD_CNT;
+ /** Async Callback pool size. */
+ private int callbackPoolSize = DFLT_PUBLIC_THREAD_CNT;
+
/** System pool size. */
private int sysPoolSize = DFLT_SYSTEM_CORE_THREAD_CNT;
@@ -723,6 +727,20 @@ public class IgniteConfiguration {
}
/**
+ * Size of thread pool that is in charge of processing asynchronous callbacks.
+ * <p>
+ * This pool is used for callbacks annotated with {@link IgniteAsyncCallback}.
+ * <p>
+ * If not provided, executor service will have size {@link #DFLT_PUBLIC_THREAD_CNT}.
+ *
+ * @return Thread pool size to be used.
+ * @see IgniteAsyncCallback
+ */
+ public int getAsyncCallbackPoolSize() {
+ return callbackPoolSize;
+ }
+
+ /**
* Size of thread pool that is in charge of processing internal and Visor
* {@link ComputeJob GridJobs}.
* <p>
@@ -831,6 +849,20 @@ public class IgniteConfiguration {
}
/**
+ * Sets async callback thread pool size to use within grid.
+ *
+ * @param poolSize Thread pool size to use within grid.
+ * @return {@code this} for chaining.
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize()
+ * @see IgniteAsyncCallback
+ */
+ public IgniteConfiguration setAsyncCallbackPoolSize(int poolSize) {
+ this.callbackPoolSize = poolSize;
+
+ return this;
+ }
+
+ /**
* Sets management thread pool size to use within grid.
*
* @param poolSize Thread pool size to use within grid.
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
index b95d595..3eaef1e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
@@ -67,6 +67,7 @@ import org.apache.ignite.internal.util.IgniteExceptionRegistry;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.plugin.PluginNotFoundException;
import org.apache.ignite.plugin.PluginProvider;
+import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor;
/**
*
@@ -298,6 +299,13 @@ public interface GridKernalContext extends Iterable<GridComponent> {
public ExecutorService marshallerCachePool();
/**
+ * Gets async callback pool.
+ *
+ * @return Async callback pool.
+ */
+ public IgniteStripedThreadPoolExecutor asyncCallbackPool();
+
+ /**
* Gets cache object processor.
*
* @return Cache object processor.
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
index 753dbe8..79d67df 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
@@ -89,6 +89,7 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.PluginNotFoundException;
import org.apache.ignite.plugin.PluginProvider;
+import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_DAEMON;
@@ -305,6 +306,10 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
/** */
@GridToStringExclude
+ protected IgniteStripedThreadPoolExecutor callbackExecSvc;
+
+ /** */
+ @GridToStringExclude
private Map<String, Object> attrs = new HashMap<>();
/** */
@@ -379,6 +384,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
ExecutorService mgmtExecSvc,
ExecutorService igfsExecSvc,
ExecutorService restExecSvc,
+ IgniteStripedThreadPoolExecutor callbackExecSvc,
List<PluginProvider> plugins) throws IgniteCheckedException {
assert grid != null;
assert cfg != null;
@@ -395,6 +401,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
this.mgmtExecSvc = mgmtExecSvc;
this.igfsExecSvc = igfsExecSvc;
this.restExecSvc = restExecSvc;
+ this.callbackExecSvc = callbackExecSvc;
marshCtx = new MarshallerContextImpl(plugins);
@@ -746,6 +753,11 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
}
/** {@inheritDoc} */
+ @Override public IgniteStripedThreadPoolExecutor asyncCallbackPool() {
+ return callbackExecSvc;
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteCacheObjectProcessor cacheObjects() {
return cacheObjProc;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 20795fc..d6655d9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -167,6 +167,7 @@ import org.apache.ignite.plugin.PluginProvider;
import org.apache.ignite.spi.IgniteSpi;
import org.apache.ignite.spi.IgniteSpiVersionCheckException;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
+import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_CONFIG_URL;
@@ -667,6 +668,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
ExecutorService mgmtExecSvc,
ExecutorService igfsExecSvc,
ExecutorService restExecSvc,
+ IgniteStripedThreadPoolExecutor callbackExecSvc,
GridAbsClosure errHnd)
throws IgniteCheckedException
{
@@ -771,6 +773,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
mgmtExecSvc,
igfsExecSvc,
restExecSvc,
+ callbackExecSvc,
plugins);
cfg.getMarshaller().setContext(ctx.marshallerContext());
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
index 9a83826..f47b7ff 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
@@ -95,6 +95,7 @@ import org.apache.ignite.spi.indexing.noop.NoopIndexingSpi;
import org.apache.ignite.spi.loadbalancing.roundrobin.RoundRobinLoadBalancingSpi;
import org.apache.ignite.spi.swapspace.file.FileSwapSpaceSpi;
import org.apache.ignite.spi.swapspace.noop.NoopSwapSpaceSpi;
+import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor;
import org.apache.ignite.thread.IgniteThread;
import org.apache.ignite.thread.IgniteThreadPoolExecutor;
import org.jetbrains.annotations.Nullable;
@@ -1473,6 +1474,9 @@ public class IgnitionEx {
/** Marshaller cache executor service. */
private ExecutorService marshCacheExecSvc;
+ /** Continuous query executor service. */
+ private IgniteStripedThreadPoolExecutor callbackExecSvc;
+
/** Grid state. */
private volatile IgniteState state = STOPPED;
@@ -1685,6 +1689,12 @@ public class IgnitionEx {
0,
new LinkedBlockingQueue<Runnable>());
+ // Note that we do not pre-start threads here as this pool may not be needed.
+ callbackExecSvc = new IgniteStripedThreadPoolExecutor(
+ cfg.getAsyncCallbackPoolSize(),
+ cfg.getGridName(),
+ "callback");
+
if (myCfg.getConnectorConfiguration() != null) {
restExecSvc = new IgniteThreadPoolExecutor(
"rest",
@@ -1724,7 +1734,7 @@ public class IgnitionEx {
grid = grid0;
grid0.start(myCfg, utilityCacheExecSvc, marshCacheExecSvc, execSvc, sysExecSvc, p2pExecSvc, mgmtExecSvc,
- igfsExecSvc, restExecSvc,
+ igfsExecSvc, restExecSvc, callbackExecSvc,
new CA() {
@Override public void apply() {
startLatch.countDown();
@@ -2327,6 +2337,10 @@ public class IgnitionEx {
U.shutdownNow(getClass(), marshCacheExecSvc, log);
marshCacheExecSvc = null;
+
+ U.shutdownNow(getClass(), callbackExecSvc, log);
+
+ callbackExecSvc = null;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
index d6d7335..e679dfd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
@@ -27,6 +27,7 @@ import org.apache.ignite.cache.eviction.EvictableEntry;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateFuture;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -462,6 +463,7 @@ public interface GridCacheEntryEx {
* @param subjId Subject ID initiated this update.
* @param taskName Task name.
* @param updateCntr Update counter.
+ * @param fut Dht atomic future.
* @return Tuple where first value is flag showing whether operation succeeded,
* second value is old entry value if return value is requested, third is updated entry value,
* fourth is the version to enqueue for deferred delete the fifth is DR conflict context
@@ -497,7 +499,8 @@ public interface GridCacheEntryEx {
@Nullable UUID subjId,
String taskName,
@Nullable CacheObject prevVal,
- @Nullable Long updateCntr
+ @Nullable Long updateCntr,
+ @Nullable GridDhtAtomicUpdateFuture fut
) throws IgniteCheckedException, GridCacheEntryRemovedException;
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index 735e20a..75d96d8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -40,6 +40,7 @@ import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheEntry;
import org.apache.ignite.internal.processors.cache.extras.GridCacheEntryExtras;
import org.apache.ignite.internal.processors.cache.extras.GridCacheMvccEntryExtras;
@@ -1248,6 +1249,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
tx.local(),
false,
updateCntr0,
+ null,
topVer);
}
@@ -1445,6 +1447,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
tx.local(),
false,
updateCntr0,
+ null,
topVer);
}
@@ -1821,6 +1824,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
true,
false,
updateCntr,
+ null,
AffinityTopologyVersion.NONE);
}
@@ -1870,7 +1874,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
@Nullable final UUID subjId,
final String taskName,
@Nullable final CacheObject prevVal,
- @Nullable final Long updateCntr
+ @Nullable final Long updateCntr,
+ @Nullable GridDhtAtomicUpdateFuture fut
) throws IgniteCheckedException, GridCacheEntryRemovedException, GridClosureException {
assert cctx.atomic();
@@ -1899,7 +1904,12 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
Long updateCntr0 = null;
synchronized (this) {
- boolean needVal = intercept || retval || op == GridCacheOperation.TRANSFORM || !F.isEmptyOrNulls(filter);
+ boolean internal = isInternal() || !context().userCache();
+
+ Map<UUID, CacheContinuousQueryListener> lsnrs = cctx.continuousQueries().updateListeners(internal, false);
+
+ boolean needVal = lsnrs != null || intercept || retval || op == GridCacheOperation.TRANSFORM
+ || !F.isEmptyOrNulls(filter);
checkObsolete();
@@ -2093,6 +2103,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
primary,
false,
updateCntr0,
+ null,
topVer);
}
@@ -2501,13 +2512,42 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (res)
updateMetrics(op, metrics);
+ // Continuous query filter should be perform under lock.
+ if (lsnrs != null) {
+ CacheObject evtVal = updated;
+ CacheObject evtOldVal = oldVal;
+
+ if (isOffHeapValuesOnly()) {
+ evtVal = cctx.toCacheObject(cctx.unwrapTemporary(evtVal));
+
+ evtOldVal = cctx.toCacheObject(cctx.unwrapTemporary(evtOldVal));
+ }
+
+ cctx.continuousQueries().onEntryUpdated(lsnrs, key, evtVal, evtOldVal, internal,
+ partition(), primary, false, updateCntr0, fut, topVer);
+ }
+
cctx.dataStructures().onEntryUpdated(key, op == GridCacheOperation.DELETE, keepBinary);
if (intercept) {
if (op == GridCacheOperation.UPDATE)
- cctx.config().getInterceptor().onAfterPut(new CacheLazyEntry(cctx, key, key0, updated, updated0, keepBinary, updateCntr0));
+ cctx.config().getInterceptor().onAfterPut(new CacheLazyEntry(
+ cctx,
+ key,
+ key0,
+ updated,
+ updated0,
+ keepBinary,
+ updateCntr0));
else
- cctx.config().getInterceptor().onAfterRemove(new CacheLazyEntry(cctx, key, key0, oldVal, old0, keepBinary, updateCntr0));
+ cctx.config().getInterceptor().onAfterRemove(new CacheLazyEntry(
+ cctx,
+ key,
+ key0,
+ oldVal,
+ old0,
+ keepBinary,
+ updateCntr0));
if (interceptRes != null)
oldVal = cctx.toCacheObject(cctx.unwrapTemporary(interceptRes.get2()));
@@ -3302,6 +3342,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
true,
preload,
updateCntr,
+ null,
topVer);
cctx.dataStructures().onEntryUpdated(key, false, true);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 013184b..d28aaaa 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -76,7 +76,6 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSing
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
-import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
@@ -2141,10 +2140,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
boolean intercept = ctx.config().getInterceptor() != null;
- boolean initLsnrs = false;
- Map<UUID, CacheContinuousQueryListener> lsnrs = null;
- boolean internal = false;
-
// Avoid iterator creation.
for (int i = 0; i < keys.size(); i++) {
KeyCacheObject k = keys.get(i);
@@ -2159,14 +2154,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (entry == null)
continue;
- if (!initLsnrs) {
- internal = entry.isInternal() || !context().userCache();
-
- lsnrs = ctx.continuousQueries().updateListeners(internal, false);
-
- initLsnrs = true;
- }
-
GridCacheVersion newConflictVer = req.conflictVersion(i);
long newConflictTtl = req.conflictTtl(i);
long newConflictExpireTime = req.conflictExpireTime(i);
@@ -2195,7 +2182,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
req.invokeArguments(),
primary && writeThrough() && !req.skipStore(),
!req.skipStore(),
- lsnrs != null || sndPrevVal || req.returnValue(),
+ sndPrevVal || req.returnValue(),
req.keepBinary(),
expiry,
true,
@@ -2213,7 +2200,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
req.subjectId(),
taskName,
null,
- null);
+ null,
+ dhtFut);
if (dhtFut == null && !F.isEmpty(filteredReaders)) {
dhtFut = createDhtFuture(ver, req, res, completionCb, true);
@@ -2222,8 +2210,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
if (dhtFut != null) {
- dhtFut.listeners(lsnrs);
-
if (updRes.sendToDht()) { // Send to backups even in case of remove-remove scenarios.
GridCacheVersionConflictContext<?, ?> conflictCtx = updRes.conflictResolveResult();
@@ -2260,19 +2246,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
"[entry=" + entry + ", filter=" + Arrays.toString(req.filter()) + ']');
}
}
- else if (lsnrs != null && updRes.updateCounter() != 0) {
- ctx.continuousQueries().onEntryUpdated(
- lsnrs,
- entry.key(),
- updRes.newValue(),
- updRes.oldValue(),
- internal,
- entry.partition(),
- primary,
- false,
- updRes.updateCounter(),
- topVer);
- }
if (hasNear) {
if (primary && updRes.sendToDht()) {
@@ -2446,9 +2419,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
boolean intercept = ctx.config().getInterceptor() != null;
- boolean initLsnrs = false;
- Map<UUID, CacheContinuousQueryListener> lsnrs = null;
-
// Avoid iterator creation.
for (int i = 0; i < entries.size(); i++) {
GridDhtCacheEntry entry = entries.get(i);
@@ -2482,14 +2452,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
filteredReaders = F.view(entry.readers(), F.notEqualTo(node.id()));
}
- if (!initLsnrs) {
- lsnrs = ctx.continuousQueries().updateListeners(
- entry.isInternal() || !context().userCache(),
- false);
-
- initLsnrs = true;
- }
-
GridCacheUpdateAtomicResult updRes = entry.innerUpdate(
ver,
node.id(),
@@ -2499,7 +2461,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
null,
/*write-through*/false,
/*read-through*/false,
- /*retval*/sndPrevVal || lsnrs != null,
+ /*retval*/sndPrevVal,
req.keepBinary(),
expiry,
/*event*/true,
@@ -2517,7 +2479,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
req.subjectId(),
taskName,
null,
- null);
+ null,
+ dhtFut);
assert !updRes.success() || updRes.newTtl() == CU.TTL_NOT_CHANGED || expiry != null :
"success=" + updRes.success() + ", newTtl=" + updRes.newTtl() + ", expiry=" + expiry;
@@ -2548,12 +2511,10 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
if (dhtFut != null) {
- dhtFut.listeners(lsnrs);
-
EntryProcessor<Object, Object, Object> entryProcessor =
entryProcessorMap == null ? null : entryProcessorMap.get(entry.key());
- if (!batchRes.readersOnly())
+ if (!batchRes.readersOnly()) {
dhtFut.addWriteEntry(entry,
writeVal,
entryProcessor,
@@ -2563,6 +2524,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
sndPrevVal,
updRes.oldValue(),
updRes.updateCounter());
+ }
if (!F.isEmpty(filteredReaders))
dhtFut.addNearWriteEntries(filteredReaders,
@@ -2572,19 +2534,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
updRes.newTtl(),
CU.EXPIRE_TIME_CALCULATE);
}
- else if (lsnrs != null && updRes.updateCounter() != 0) {
- ctx.continuousQueries().onEntryUpdated(
- lsnrs,
- entry.key(),
- updRes.newValue(),
- updRes.oldValue(),
- entry.isInternal() || !context().userCache(),
- entry.partition(),
- primary,
- false,
- updRes.updateCounter(),
- topVer);
- }
if (hasNear) {
if (primary) {
@@ -2965,10 +2914,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
String taskName = ctx.kernalContext().task().resolveTaskName(req.taskNameHash());
- boolean initLsnrs = false;
- Map<UUID, CacheContinuousQueryListener> lsnrs = null;
- boolean internal = false;
-
for (int i = 0; i < req.size(); i++) {
KeyCacheObject key = req.key(i);
@@ -2991,14 +2936,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
long ttl = req.ttl(i);
long expireTime = req.conflictExpireTime(i);
- if (!initLsnrs) {
- internal = entry.isInternal() || !context().userCache();
-
- lsnrs = ctx.continuousQueries().updateListeners(internal, false);
-
- initLsnrs = true;
- }
-
GridCacheUpdateAtomicResult updRes = entry.innerUpdate(
ver,
nodeId,
@@ -3008,7 +2945,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
op == TRANSFORM ? req.invokeArguments() : null,
/*write-through*/false,
/*read-through*/false,
- /*retval*/lsnrs != null,
+ /*retval*/false,
req.keepBinary(),
/*expiry policy*/null,
/*event*/true,
@@ -3026,25 +2963,12 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
req.subjectId(),
taskName,
prevVal,
- updateIdx);
+ updateIdx,
+ null);
if (updRes.removeVersion() != null)
ctx.onDeferredDelete(entry, updRes.removeVersion());
- if (lsnrs != null && updRes.updateCounter() != 0) {
- ctx.continuousQueries().onEntryUpdated(
- lsnrs,
- entry.key(),
- updRes.newValue(),
- updRes.oldValue(),
- internal,
- entry.partition(),
- false,
- false,
- updRes.updateCounter(),
- req.topologyVersion());
- }
-
entry.onUnlock();
break; // While.
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index 4721d6e..5760596 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -37,11 +37,11 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
-import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.CI2;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -97,15 +97,15 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
/** Future keys. */
private final Collection<KeyCacheObject> keys;
+ /** Continuous query closures. */
+ private Collection<CI1<Boolean>> cntQryClsrs;
+
/** */
private final boolean waitForExchange;
/** Response count. */
private volatile int resCnt;
- /** */
- private Map<UUID, CacheContinuousQueryListener> lsnrs;
-
/**
* @param cctx Cache context.
* @param completionCb Callback to invoke when future is completed.
@@ -138,13 +138,6 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
waitForExchange = !(updateReq.topologyLocked() || (updateReq.fastMap() && !updateReq.clientRequest()));
}
- /**
- * @param lsnrs Continuous query listeners.
- */
- void listeners(@Nullable Map<UUID, CacheContinuousQueryListener> lsnrs) {
- this.lsnrs = lsnrs;
- }
-
/** {@inheritDoc} */
@Override public IgniteUuid futureId() {
return futVer.asGridUuid();
@@ -276,27 +269,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
addPrevVal,
entry.partition(),
prevVal,
- updateCntr,
- lsnrs != null);
- }
- else if (lsnrs != null && dhtNodes.size() == 1) {
- try {
- cctx.continuousQueries().onEntryUpdated(
- lsnrs,
- entry.key(),
- val,
- prevVal,
- entry.key().internal() || !cctx.userCache(),
- entry.partition(),
- true,
- false,
- updateCntr,
- updateReq.topologyVersion());
- }
- catch (IgniteCheckedException e) {
- U.warn(log, "Failed to send continuous query message. [key=" + entry.key() + ", newVal="
- + val + ", err=" + e + "]");
- }
+ updateCntr);
}
}
}
@@ -361,77 +334,33 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
}
}
+ /**
+ * @param clsr Continuous query closure.
+ */
+ public void addContinuousQueryClosure(CI1<Boolean> clsr){
+ assert !isDone() : this;
+
+ if (cntQryClsrs == null)
+ cntQryClsrs = new ArrayList<>(10);
+
+ cntQryClsrs.add(clsr);
+ }
+
/** {@inheritDoc} */
@Override public boolean onDone(@Nullable Void res, @Nullable Throwable err) {
if (super.onDone(res, err)) {
cctx.mvcc().removeAtomicFuture(version());
- if (err != null) {
- if (!mappings.isEmpty() && lsnrs != null) {
- Collection<KeyCacheObject> hndKeys = new ArrayList<>(keys.size());
-
- exit: for (GridDhtAtomicUpdateRequest req : mappings.values()) {
- for (int i = 0; i < req.size(); i++) {
- KeyCacheObject key = req.key(i);
-
- if (!hndKeys.contains(key)) {
- updateRes.addFailedKey(key, err);
+ boolean suc = err == null;
- cctx.continuousQueries().skipUpdateEvent(
- lsnrs,
- key,
- req.partitionId(i),
- req.updateCounter(i),
- updateReq.topologyVersion());
-
- hndKeys.add(key);
-
- if (hndKeys.size() == keys.size())
- break exit;
- }
- }
- }
- }
- else
- for (KeyCacheObject key : keys)
- updateRes.addFailedKey(key, err);
+ if (!suc) {
+ for (KeyCacheObject key : keys)
+ updateRes.addFailedKey(key, err);
}
- else {
- if (lsnrs != null) {
- Collection<KeyCacheObject> hndKeys = new ArrayList<>(keys.size());
-
- exit: for (GridDhtAtomicUpdateRequest req : mappings.values()) {
- for (int i = 0; i < req.size(); i++) {
- KeyCacheObject key = req.key(i);
-
- if (!hndKeys.contains(key)) {
- try {
- cctx.continuousQueries().onEntryUpdated(
- lsnrs,
- key,
- req.value(i),
- req.localPreviousValue(i),
- key.internal() || !cctx.userCache(),
- req.partitionId(i),
- true,
- false,
- req.updateCounter(i),
- updateReq.topologyVersion());
- }
- catch (IgniteCheckedException e) {
- U.warn(log, "Failed to send continuous query message. [key=" + key +
- ", newVal=" + req.value(i) +
- ", err=" + e + "]");
- }
-
- hndKeys.add(key);
-
- if (hndKeys.size() == keys.size())
- break exit;
- }
- }
- }
- }
+
+ if (cntQryClsrs != null) {
+ for (CI1<Boolean> clsr : cntQryClsrs)
+ clsr.apply(suc);
}
if (updateReq.writeSynchronizationMode() == FULL_SYNC)
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index c8e33c2..b5e2835 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -155,10 +155,6 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
@GridDirectTransient
private List<Integer> partIds;
- /** */
- @GridDirectTransient
- private List<CacheObject> locPrevVals;
-
/** Keep binary flag. */
private boolean keepBinary;
@@ -242,7 +238,6 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
* @param partId Partition.
* @param prevVal Previous value.
* @param updateCntr Update counter.
- * @param storeLocPrevVal If {@code true} stores previous value.
*/
public void addWriteValue(KeyCacheObject key,
@Nullable CacheObject val,
@@ -253,19 +248,12 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
boolean addPrevVal,
int partId,
@Nullable CacheObject prevVal,
- @Nullable Long updateCntr,
- boolean storeLocPrevVal) {
+ @Nullable Long updateCntr
+ ) {
keys.add(key);
partIds.add(partId);
- if (storeLocPrevVal) {
- if (locPrevVals == null)
- locPrevVals = new ArrayList<>();
-
- locPrevVals.add(prevVal);
- }
-
if (forceTransformBackups) {
assert entryProcessor != null;
@@ -526,16 +514,6 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
/**
* @param idx Key index.
- * @return Value.
- */
- @Nullable public CacheObject localPreviousValue(int idx) {
- assert locPrevVals != null;
-
- return locPrevVals.get(idx);
- }
-
- /**
- * @param idx Key index.
* @return Entry processor.
*/
@Nullable public EntryProcessor<Object, Object, Object> entryProcessor(int idx) {
@@ -1069,13 +1047,6 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
private void cleanup() {
nearVals = null;
prevVals = null;
-
- // Do not keep values if they are not needed for continuous query notification.
- if (locPrevVals == null) {
- keys = null;
- vals = null;
- locPrevVals = null;
- }
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
index a7481d3..3e0e392 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
@@ -37,7 +37,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
-import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
@@ -271,6 +270,7 @@ public class GridNearAtomicCache<K, V> extends GridNearCacheAdapter<K, V> {
subjId,
taskName,
null,
+ null,
null);
if (updRes.removeVersion() != null)
@@ -372,6 +372,7 @@ public class GridNearAtomicCache<K, V> extends GridNearCacheAdapter<K, V> {
req.subjectId(),
taskName,
null,
+ null,
null);
if (updRes.removeVersion() != null)
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEvent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEvent.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEvent.java
index eab5dbd..db70e2e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEvent.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEvent.java
@@ -56,6 +56,13 @@ class CacheContinuousQueryEvent<K, V> extends CacheQueryEntryEvent<K, V> {
return e;
}
+ /**
+ * @return Partition ID.
+ */
+ public int partitionId() {
+ return e.partition();
+ }
+
/** {@inheritDoc} */
@Override public K getKey() {
return (K)cctx.cacheObjectContext().unwrapBinaryIfNeeded(e.key(), e.isKeepBinary(), false);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
index 16513b0..9ae2972 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
@@ -48,6 +48,7 @@ import org.apache.ignite.events.CacheQueryExecutedEvent;
import org.apache.ignite.events.CacheQueryReadEvent;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteDeploymentCheckedException;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
import org.apache.ignite.internal.managers.deployment.GridDeployment;
@@ -58,7 +59,10 @@ import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheAffinityManager;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheDeploymentManager;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateFuture;
import org.apache.ignite.internal.processors.cache.query.CacheQueryType;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryManager.JCacheQueryLocalListener;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryManager.JCacheQueryRemoteFilter;
import org.apache.ignite.internal.processors.continuous.GridContinuousBatch;
import org.apache.ignite.internal.processors.continuous.GridContinuousBatchAdapter;
import org.apache.ignite.internal.processors.continuous.GridContinuousHandler;
@@ -66,13 +70,14 @@ import org.apache.ignite.internal.processors.platform.cache.query.PlatformContin
import org.apache.ignite.internal.util.GridConcurrentSkipListSet;
import org.apache.ignite.internal.util.GridLongList;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
-import org.apache.ignite.internal.util.typedef.C1;
+import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteAsyncCallback;
import org.apache.ignite.lang.IgniteBiTuple;
-import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentLinkedDeque8;
@@ -159,6 +164,21 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
/** */
private transient boolean ignoreClsNotFound;
+ /** */
+ private transient boolean asyncCallback;
+
+ /** */
+ private transient UUID nodeId;
+
+ /** */
+ private transient UUID routineId;
+
+ /** */
+ private transient GridKernalContext ctx;
+
+ /** */
+ private transient IgniteLogger log;
+
/**
* Required by {@link Externalizable}.
*/
@@ -283,13 +303,36 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
assert routineId != null;
assert ctx != null;
- if (locLsnr != null)
- ctx.resource().injectGeneric(locLsnr);
+ if (locLsnr != null) {
+ if (locLsnr instanceof JCacheQueryLocalListener) {
+ ctx.resource().injectGeneric(((JCacheQueryLocalListener)locLsnr).impl);
+
+ asyncCallback = ((JCacheQueryLocalListener)locLsnr).async();
+ }
+ else {
+ ctx.resource().injectGeneric(locLsnr);
+
+ asyncCallback = U.hasAnnotation(locLsnr, IgniteAsyncCallback.class);
+ }
+ }
final CacheEntryEventFilter filter = getEventFilter();
- if (filter != null)
- ctx.resource().injectGeneric(filter);
+ if (filter != null) {
+ if (filter instanceof JCacheQueryRemoteFilter) {
+ if (((JCacheQueryRemoteFilter)filter).impl != null)
+ ctx.resource().injectGeneric(((JCacheQueryRemoteFilter)filter).impl);
+
+ if (!asyncCallback)
+ asyncCallback = ((JCacheQueryRemoteFilter)filter).async();
+ }
+ else {
+ ctx.resource().injectGeneric(filter);
+
+ if (!asyncCallback)
+ asyncCallback = U.hasAnnotation(filter, IgniteAsyncCallback.class);
+ }
+ }
entryBufs = new ConcurrentHashMap<>();
@@ -299,10 +342,18 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
rcvs = new ConcurrentHashMap<>();
+ this.nodeId = nodeId;
+
+ this.routineId = routineId;
+
+ this.ctx = ctx;
+
final boolean loc = nodeId.equals(ctx.localNodeId());
assert !skipPrimaryCheck || loc;
+ log = ctx.log(CacheContinuousQueryHandler.class);
+
CacheContinuousQueryListener<K, V> lsnr = new CacheContinuousQueryListener<K, V>() {
@Override public void onExecution() {
if (ctx.event().isRecordable(EVT_CACHE_QUERY_EXECUTED)) {
@@ -324,15 +375,16 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
}
- /** {@inheritDoc} */
@Override public boolean keepBinary() {
return keepBinary;
}
- @Override public void onEntryUpdated(CacheContinuousQueryEvent<K, V> evt, boolean primary,
- boolean recordIgniteEvt) {
+ @Override public void onEntryUpdated(final CacheContinuousQueryEvent<K, V> evt,
+ boolean primary,
+ final boolean recordIgniteEvt,
+ GridDhtAtomicUpdateFuture fut) {
if (ignoreExpired && evt.getEventType() == EventType.EXPIRED)
- return;
+ return ;
final GridCacheContext<K, V> cctx = cacheContext(ctx);
@@ -343,93 +395,33 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
// skipPrimaryCheck is set only when listen locally for replicated cache events.
assert !skipPrimaryCheck || (cctx.isReplicated() && ctx.localNodeId().equals(nodeId));
- boolean notify = !evt.entry().isFiltered();
+ if (asyncCallback) {
+ ContinuousQueryAsyncClosure clsr = new ContinuousQueryAsyncClosure(
+ primary,
+ evt,
+ recordIgniteEvt,
+ fut);
- if (notify && filter != null) {
- try {
- notify = filter.evaluate(evt);
- }
- catch (Exception e) {
- U.error(cctx.logger(CacheContinuousQueryHandler.class), "CacheEntryEventFilter failed: " + e);
- }
+ ctx.asyncCallbackPool().execute(clsr, evt.partitionId());
}
-
- try {
- final CacheContinuousQueryEntry entry = evt.entry();
-
- if (!notify)
- entry.markFiltered();
+ else {
+ final boolean notify = filter(evt, primary);
if (primary || skipPrimaryCheck) {
- if (loc) {
- if (!locCache) {
- Collection<CacheEntryEvent<? extends K, ? extends V>> entries = handleEvent(ctx, entry);
-
- if (!entries.isEmpty()) {
- locLsnr.onUpdated(entries);
-
- if (!internal && !skipPrimaryCheck)
- sendBackupAcknowledge(ackBuf.onAcknowledged(entry), routineId, ctx);
- }
- }
- else {
- if (!entry.isFiltered())
- locLsnr.onUpdated(F.<CacheEntryEvent<? extends K, ? extends V>>asList(evt));
- }
- }
+ if (fut == null)
+ onEntryUpdate(evt, notify, loc, recordIgniteEvt);
else {
- if (!entry.isFiltered())
- prepareEntry(cctx, nodeId, entry);
-
- CacheContinuousQueryEntry e = handleEntry(entry);
-
- if (e != null)
- ctx.continuous().addNotification(nodeId, routineId, entry, topic, sync, true);
- }
- }
- else {
- if (!internal) {
- // Skip init query and expire entries.
- if (entry.updateCounter() != -1L) {
- entry.markBackup();
+ fut.addContinuousQueryClosure(new CI1<Boolean>() {
+ @Override public void apply(Boolean suc) {
+ if (!suc)
+ evt.entry().markFiltered();
- backupQueue.add(entry);
- }
+ onEntryUpdate(evt, notify, loc, recordIgniteEvt);
+ }
+ });
}
}
}
- catch (ClusterTopologyCheckedException ex) {
- IgniteLogger log = ctx.log(getClass());
-
- if (log.isDebugEnabled())
- log.debug("Failed to send event notification to node, node left cluster " +
- "[node=" + nodeId + ", err=" + ex + ']');
- }
- catch (IgniteCheckedException ex) {
- U.error(ctx.log(getClass()), "Failed to send event notification to node: " + nodeId, ex);
- }
-
- if (recordIgniteEvt && notify) {
- ctx.event().record(new CacheQueryReadEvent<>(
- ctx.discovery().localNode(),
- "Continuous query executed.",
- EVT_CACHE_QUERY_OBJECT_READ,
- CacheQueryType.CONTINUOUS.name(),
- cacheName,
- null,
- null,
- null,
- filter instanceof CacheEntryEventSerializableFilter ?
- (CacheEntryEventSerializableFilter)filter : null,
- null,
- nodeId,
- taskName(),
- evt.getKey(),
- evt.getValue(),
- evt.getOldValue(),
- null
- ));
- }
}
@Override public void onUnregister() {
@@ -475,15 +467,15 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
sendBackupAcknowledge(ackBuf.acknowledgeOnTimeout(), routineId, ctx);
}
- @Override public void skipUpdateEvent(CacheContinuousQueryEvent<K, V> evt, AffinityTopologyVersion topVer,
- boolean primary) {
+ @Override public void skipUpdateEvent(CacheContinuousQueryEvent<K, V> evt,
+ AffinityTopologyVersion topVer, boolean primary) {
assert evt != null;
CacheContinuousQueryEntry e = evt.entry();
e.markFiltered();
- onEntryUpdated(evt, primary, false);
+ onEntryUpdated(evt, primary, false, null);
}
@Override public void onPartitionEvicted(int part) {
@@ -580,17 +572,73 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
- @Override public void notifyCallback(UUID nodeId, UUID routineId, Collection<?> objs, GridKernalContext ctx) {
+ @Override public void notifyCallback(final UUID nodeId,
+ final UUID routineId,
+ Collection<?> objs,
+ final GridKernalContext ctx) {
assert nodeId != null;
assert routineId != null;
assert objs != null;
assert ctx != null;
- Collection<CacheContinuousQueryEntry> entries = (Collection<CacheContinuousQueryEntry>)objs;
+ final List<CacheContinuousQueryEntry> entries = (List<CacheContinuousQueryEntry>)objs;
+
+ if (entries.isEmpty())
+ return;
+
+ if (asyncCallback) {
+ IgniteStripedThreadPoolExecutor asyncPool = ctx.asyncCallbackPool();
+
+ int threadId = asyncPool.threadId(entries.get(0).partition());
+
+ int startIdx = 0;
+
+ if (entries.size() != 1) {
+ for (int i = 1; i < entries.size(); i++) {
+ int curThreadId = asyncPool.threadId(entries.get(i).partition());
+
+ // If all entries from one partition avoid creation new collections.
+ if (curThreadId == threadId)
+ continue;
+
+ final int i0 = i;
+ final int startIdx0 = startIdx;
+
+ asyncPool.execute(new Runnable() {
+ @Override public void run() {
+ notifyCallback0(nodeId, ctx, entries.subList(startIdx0, i0));
+ }
+ }, threadId);
+
+ startIdx = i0;
+ threadId = curThreadId;
+ }
+ }
+
+ final int startIdx0 = startIdx;
+
+ asyncPool.execute(new Runnable() {
+ @Override public void run() {
+ notifyCallback0(nodeId, ctx,
+ startIdx0 == 0 ? entries : entries.subList(startIdx0, entries.size()));
+ }
+ }, threadId);
+ }
+ else
+ notifyCallback0(nodeId, ctx, entries);
+ }
+ /**
+ * @param nodeId Node id.
+ * @param ctx Kernal context.
+ * @param entries Entries.
+ */
+ private void notifyCallback0(UUID nodeId,
+ final GridKernalContext ctx,
+ Collection<CacheContinuousQueryEntry> entries) {
final GridCacheContext cctx = cacheContext(ctx);
- Collection<CacheEntryEvent<? extends K, ? extends V>> entries0 = new ArrayList<>();
+ final Collection<CacheEntryEvent<? extends K, ? extends V>> entries0 = new ArrayList<>(entries.size());
for (CacheContinuousQueryEntry e : entries) {
GridCacheDeploymentManager depMgr = cctx.deploy();
@@ -609,7 +657,10 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
try {
e.unmarshal(cctx, ldr);
- entries0.addAll(handleEvent(ctx, e));
+ Collection<CacheEntryEvent<? extends K, ? extends V>> evts = handleEvent(ctx, e);
+
+ if (evts != null && !evts.isEmpty())
+ entries0.addAll(evts);
}
catch (IgniteCheckedException ex) {
if (ignoreClsNotFound)
@@ -640,8 +691,8 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
if (e.isFiltered())
return Collections.emptyList();
else
- return F.<CacheEntryEvent<? extends K, ? extends V>>asList(
- new CacheContinuousQueryEvent<K, V>(cache, cctx, e));
+ return F.<CacheEntryEvent<? extends K, ? extends V>>
+ asList(new CacheContinuousQueryEvent<K, V>(cache, cctx, e));
}
// Initial query entry or evicted entry. These events should be fired immediately.
@@ -653,7 +704,117 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
PartitionRecovery rec = getOrCreatePartitionRecovery(ctx, e.partition());
- return rec.collectEntries(cctx, cache, e);
+ return rec.collectEntries(e, cctx, cache);
+ }
+
+ /**
+ * @param primary Primary.
+ * @param evt Query event.
+ * @return {@code True} if event passed filter otherwise {@code true}.
+ */
+ public boolean filter(CacheContinuousQueryEvent evt, boolean primary) {
+ CacheContinuousQueryEntry entry = evt.entry();
+
+ boolean notify = !entry.isFiltered();
+
+ try {
+ if (notify && getEventFilter() != null)
+ notify = getEventFilter().evaluate(evt);
+ }
+ catch (Exception e) {
+ U.error(log, "CacheEntryEventFilter failed: " + e);
+ }
+
+ if (!notify)
+ entry.markFiltered();
+
+ if (!primary && !internal && entry.updateCounter() != -1L /* Skip init query and expire entries */) {
+ entry.markBackup();
+
+ backupQueue.add(entry);
+ }
+
+ return notify;
+ }
+
+ /**
+ * @param evt Continuous query event.
+ * @param notify Notify flag.
+ * @param loc Listener deployed on this node.
+ * @param recordIgniteEvt Record ignite event.
+ */
+ private void onEntryUpdate(CacheContinuousQueryEvent evt, boolean notify, boolean loc, boolean recordIgniteEvt) {
+ try {
+ GridCacheContext<K, V> cctx = cacheContext(ctx);
+
+ if (cctx == null)
+ return;
+
+ final CacheContinuousQueryEntry entry = evt.entry();
+
+ if (loc) {
+ if (!locCache) {
+ Collection<CacheEntryEvent<? extends K, ? extends V>> evts = handleEvent(ctx, entry);
+
+ if (!evts.isEmpty()) {
+ locLsnr.onUpdated(evts);
+
+ if (!internal && !skipPrimaryCheck)
+ sendBackupAcknowledge(ackBuf.onAcknowledged(entry), routineId, ctx);
+ }
+ }
+ else {
+ if (!entry.isFiltered())
+ locLsnr.onUpdated(F.<CacheEntryEvent<? extends K, ? extends V>>asList(evt));
+ }
+ }
+ else {
+ if (!entry.isFiltered())
+ prepareEntry(cctx, nodeId, entry);
+
+ CacheContinuousQueryEntry e = handleEntry(entry);
+
+ if (e != null)
+ ctx.continuous().addNotification(nodeId, routineId, entry, topic, sync, true);
+ }
+ }
+ catch (ClusterTopologyCheckedException ex) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to send event notification to node, node left cluster " +
+ "[node=" + nodeId + ", err=" + ex + ']');
+ }
+ catch (IgniteCheckedException ex) {
+ U.error(ctx.log(getClass()), "Failed to send event notification to node: " + nodeId, ex);
+ }
+
+ if (recordIgniteEvt && notify) {
+ ctx.event().record(new CacheQueryReadEvent<>(
+ ctx.discovery().localNode(),
+ "Continuous query executed.",
+ EVT_CACHE_QUERY_OBJECT_READ,
+ CacheQueryType.CONTINUOUS.name(),
+ cacheName,
+ null,
+ null,
+ null,
+ getEventFilter() instanceof CacheEntryEventSerializableFilter ?
+ (CacheEntryEventSerializableFilter)getEventFilter() : null,
+ null,
+ nodeId,
+ taskName(),
+ evt.getKey(),
+ evt.getValue(),
+ evt.getOldValue(),
+ null
+ ));
+ }
+ }
+
+ /**
+ * @return Task name.
+ */
+ private String taskName() {
+ return ctx.security().enabled() ? ctx.task().resolveTaskName(taskHash) : null;
}
/**
@@ -781,9 +942,11 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
* @param entry Cache continuous query entry.
* @return Collection entries which will be fired. This collection should contains only non-filtered events.
*/
- public <K, V> Collection<CacheEntryEvent<? extends K, ? extends V>> collectEntries(GridCacheContext cctx,
- IgniteCache cache,
- CacheContinuousQueryEntry entry) {
+ <K, V> Collection<CacheEntryEvent<? extends K, ? extends V>> collectEntries(
+ CacheContinuousQueryEntry entry,
+ GridCacheContext cctx,
+ IgniteCache cache
+ ) {
assert entry != null;
if (entry.topologyVersion() == null) { // Possible if entry is sent from old node.
@@ -1241,6 +1404,87 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
/**
+ *
+ */
+ private class ContinuousQueryAsyncClosure implements Runnable {
+ /** */
+ private final CacheContinuousQueryEvent<K, V> evt;
+
+ /** */
+ private final boolean primary;
+
+ /** */
+ private final boolean recordIgniteEvt;
+
+ /** */
+ private final IgniteInternalFuture<?> fut;
+
+ /**
+ * @param primary Primary flag.
+ * @param evt Event.
+ * @param recordIgniteEvt Fired event.
+ * @param fut Dht future.
+ */
+ ContinuousQueryAsyncClosure(
+ boolean primary,
+ CacheContinuousQueryEvent<K, V> evt,
+ boolean recordIgniteEvt,
+ IgniteInternalFuture<?> fut) {
+ this.primary = primary;
+ this.evt = evt;
+ this.recordIgniteEvt = recordIgniteEvt;
+ this.fut = fut;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void run() {
+ final boolean notify = filter(evt, primary);
+
+ if (!primary())
+ return;
+
+ if (fut == null) {
+ onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
+
+ return;
+ }
+
+ if (fut.isDone()) {
+ if (fut.error() != null)
+ evt.entry().markFiltered();
+
+ onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
+ }
+ else {
+ fut.listen(new CI1<IgniteInternalFuture<?>>() {
+ @Override public void apply(IgniteInternalFuture<?> f) {
+ if (f.error() != null)
+ evt.entry().markFiltered();
+
+ ctx.asyncCallbackPool().execute(new Runnable() {
+ @Override public void run() {
+ onEntryUpdate(evt, notify, nodeId.equals(ctx.localNodeId()), recordIgniteEvt);
+ }
+ }, evt.entry().partition());
+ }
+ });
+ }
+ }
+
+ /**
+ * @return {@code True} if event fired on this node.
+ */
+ private boolean primary() {
+ return primary || skipPrimaryCheck;
+ }
+
+ /** {@inheritDoc} */
+ public String toString() {
+ return S.toString(ContinuousQueryAsyncClosure.class, this);
+ }
+ }
+
+ /**
* Deployable object.
*/
protected static class DeployableObject implements Externalizable {
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java
index 83ff32c..8eca81c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryListener.java
@@ -20,6 +20,8 @@ package org.apache.ignite.internal.processors.cache.query.continuous;
import java.util.Map;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateFuture;
+import org.jetbrains.annotations.Nullable;
/**
* Continuous query listener.
@@ -36,8 +38,10 @@ public interface CacheContinuousQueryListener<K, V> {
* @param evt Event
* @param primary Primary flag.
* @param recordIgniteEvt Whether to record event.
+ * @param fut Dht atomic future.
*/
- public void onEntryUpdated(CacheContinuousQueryEvent<K, V> evt, boolean primary, boolean recordIgniteEvt);
+ public void onEntryUpdated(CacheContinuousQueryEvent<K, V> evt, boolean primary,
+ boolean recordIgniteEvt, @Nullable GridDhtAtomicUpdateFuture fut);
/**
* Listener unregistered callback.
[2/4] ignite git commit: IGNITE-2004 Fixed "Asynchronous execution of
ContinuousQuery's remote filter & local list".
Posted by nt...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryFilterRandomOperationTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryFilterRandomOperationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryFilterRandomOperationTest.java
new file mode 100644
index 0000000..dbaafe1
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryFilterRandomOperationTest.java
@@ -0,0 +1,725 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
+import javax.cache.configuration.CacheEntryListenerConfiguration;
+import javax.cache.configuration.Factory;
+import javax.cache.configuration.FactoryBuilder;
+import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
+import javax.cache.event.CacheEntryCreatedListener;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.CacheEntryExpiredListener;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryRemovedListener;
+import javax.cache.event.CacheEntryUpdatedListener;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.cache.query.CacheQueryEntryEvent;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.jetbrains.annotations.NotNull;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMemoryMode.ONHEAP_TIERED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFactoryFilterRandomOperationTest.NonSerializableFilter.isAccepted;
+import static org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryRandomOperationsTest.ContinuousDeploy.CLIENT;
+import static org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryRandomOperationsTest.ContinuousDeploy.SERVER;
+import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
+
+/**
+ *
+ */
+public class CacheContinuousQueryFactoryFilterRandomOperationTest extends CacheContinuousQueryRandomOperationsTest {
+ /** */
+ private static final int NODES = 5;
+
+ /** */
+ private static final int KEYS = 50;
+
+ /** */
+ private static final int VALS = 10;
+
+ /** */
+ public static final int ITERATION_CNT = 40;
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testInternalQuery() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED,
+ 1,
+ ATOMIC,
+ ONHEAP_TIERED,
+ false);
+
+ final IgniteCache<Object, Object> cache = grid(0).createCache(ccfg);
+
+ UUID uuid = null;
+
+ try {
+ for (int i = 0; i < 10; i++)
+ cache.put(i, i);
+
+ final CountDownLatch latch = new CountDownLatch(5);
+
+ CacheEntryUpdatedListener lsnr = new CacheEntryUpdatedListener() {
+ @Override public void onUpdated(Iterable iterable) throws CacheEntryListenerException {
+ for (Object evt : iterable) {
+ latch.countDown();
+
+ log.info("Received event: " + evt);
+ }
+ }
+ };
+
+ uuid = grid(0).context().cache().cache(cache.getName()).context().continuousQueries()
+ .executeInternalQuery(lsnr, new SerializableFilter(), false, true, true);
+
+ for (int i = 10; i < 20; i++)
+ cache.put(i, i);
+
+ assertTrue(latch.await(3, SECONDS));
+ }
+ finally {
+ if (uuid != null)
+ grid(0).context().cache().cache(cache.getName()).context().continuousQueries()
+ .cancelInternalQuery(uuid);
+
+ grid(0).destroyCache(ccfg.getName());
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void doTestContinuousQuery(CacheConfiguration<Object, Object> ccfg, ContinuousDeploy deploy)
+ throws Exception {
+ ignite(0).createCache(ccfg);
+
+ try {
+ long seed = System.currentTimeMillis();
+
+ Random rnd = new Random(seed);
+
+ log.info("Random seed: " + seed);
+
+ List<BlockingQueue<CacheEntryEvent<?, ?>>> evtsQueues = new ArrayList<>();
+
+ Collection<QueryCursor<?>> curs = new ArrayList<>();
+
+ Collection<T2<Integer, MutableCacheEntryListenerConfiguration>> lsnrCfgs = new ArrayList<>();
+
+ if (deploy == CLIENT)
+ evtsQueues.add(registerListener(ccfg.getName(), NODES - 1, curs, lsnrCfgs, rnd.nextBoolean()));
+ else if (deploy == SERVER)
+ evtsQueues.add(registerListener(ccfg.getName(), rnd.nextInt(NODES - 1), curs, lsnrCfgs,
+ rnd.nextBoolean()));
+ else {
+ boolean isSync = rnd.nextBoolean();
+
+ for (int i = 0; i < NODES - 1; i++)
+ evtsQueues.add(registerListener(ccfg.getName(), i, curs, lsnrCfgs, isSync));
+ }
+
+ ConcurrentMap<Object, Object> expData = new ConcurrentHashMap<>();
+
+ Map<Integer, Long> partCntr = new ConcurrentHashMap<>();
+
+ try {
+ for (int i = 0; i < ITERATION_CNT; i++) {
+ if (i % 10 == 0)
+ log.info("Iteration: " + i);
+
+ for (int idx = 0; idx < NODES; idx++)
+ randomUpdate(rnd, evtsQueues, expData, partCntr, grid(idx).cache(ccfg.getName()));
+ }
+ }
+ finally {
+ for (QueryCursor<?> cur : curs)
+ cur.close();
+
+ for (T2<Integer, MutableCacheEntryListenerConfiguration> e : lsnrCfgs)
+ grid(e.get1()).cache(ccfg.getName()).deregisterCacheEntryListener(e.get2());
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @param cacheName Cache name.
+ * @param nodeIdx Node index.
+ * @param curs Cursors.
+ * @param lsnrCfgs Listener configurations.
+ * @return Event queue
+ */
+ private BlockingQueue<CacheEntryEvent<?, ?>> registerListener(String cacheName,
+ int nodeIdx,
+ Collection<QueryCursor<?>> curs,
+ Collection<T2<Integer, MutableCacheEntryListenerConfiguration>> lsnrCfgs,
+ boolean sync) {
+ final BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue = new ArrayBlockingQueue<>(50_000);
+
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ MutableCacheEntryListenerConfiguration<QueryTestKey, QueryTestValue> lsnrCfg =
+ new MutableCacheEntryListenerConfiguration<>(
+ FactoryBuilder.factoryOf(new LocalNonSerialiseListener() {
+ @Override protected void onEvents(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> evts) {
+ for (CacheEntryEvent<?, ?> evt : evts)
+ evtsQueue.add(evt);
+ }
+ }),
+ createFilterFactory(),
+ true,
+ sync
+ );
+
+ grid(nodeIdx).cache(cacheName).registerCacheEntryListener((CacheEntryListenerConfiguration)lsnrCfg);
+
+ lsnrCfgs.add(new T2<Integer, MutableCacheEntryListenerConfiguration>(nodeIdx, lsnrCfg));
+ }
+ else {
+ ContinuousQuery<QueryTestKey, QueryTestValue> qry = new ContinuousQuery<>();
+
+ qry.setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
+ for (CacheEntryEvent<?, ?> evt : evts)
+ evtsQueue.add(evt);
+ }
+ });
+
+ qry.setRemoteFilterFactory(createFilterFactory());
+
+ QueryCursor<?> cur = grid(nodeIdx).cache(cacheName).query(qry);
+
+ curs.add(cur);
+ }
+
+ return evtsQueue;
+ }
+
+ /**
+ * @return Filter factory.
+ */
+ @NotNull protected Factory<? extends CacheEntryEventFilter<QueryTestKey, QueryTestValue>> createFilterFactory() {
+ return new FilterFactory();
+ }
+
+ /**
+ * @param rnd Random generator.
+ * @param evtsQueues Events queue.
+ * @param expData Expected cache data.
+ * @param partCntr Partition counter.
+ * @param cache Cache.
+ * @throws Exception If failed.
+ */
+ private void randomUpdate(
+ Random rnd,
+ List<BlockingQueue<CacheEntryEvent<?, ?>>> evtsQueues,
+ ConcurrentMap<Object, Object> expData,
+ Map<Integer, Long> partCntr,
+ IgniteCache<Object, Object> cache)
+ throws Exception {
+ Object key = new QueryTestKey(rnd.nextInt(KEYS));
+ Object newVal = value(rnd);
+ Object oldVal = expData.get(key);
+
+ int op = rnd.nextInt(11);
+
+ Ignite ignite = cache.unwrap(Ignite.class);
+
+ Transaction tx = null;
+
+ if (cache.getConfiguration(CacheConfiguration.class).getAtomicityMode() == TRANSACTIONAL && rnd.nextBoolean())
+ tx = ignite.transactions().txStart(txRandomConcurrency(rnd), txRandomIsolation(rnd));
+
+ try {
+ // log.info("Random operation [key=" + key + ", op=" + op + ']');
+
+ switch (op) {
+ case 0: {
+ cache.put(key, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
+
+ expData.put(key, newVal);
+
+ break;
+ }
+
+ case 1: {
+ cache.getAndPut(key, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
+
+ expData.put(key, newVal);
+
+ break;
+ }
+
+ case 2: {
+ cache.remove(key);
+
+ if (tx != null)
+ tx.commit();
+
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, null, oldVal);
+
+ expData.remove(key);
+
+ break;
+ }
+
+ case 3: {
+ cache.getAndRemove(key);
+
+ if (tx != null)
+ tx.commit();
+
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, null, oldVal);
+
+ expData.remove(key);
+
+ break;
+ }
+
+ case 4: {
+ cache.invoke(key, new EntrySetValueProcessor(newVal, rnd.nextBoolean()));
+
+ if (tx != null)
+ tx.commit();
+
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
+
+ expData.put(key, newVal);
+
+ break;
+ }
+
+ case 5: {
+ cache.invoke(key, new EntrySetValueProcessor(null, rnd.nextBoolean()));
+
+ if (tx != null)
+ tx.commit();
+
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, null, oldVal);
+
+ expData.remove(key);
+
+ break;
+ }
+
+ case 6: {
+ cache.putIfAbsent(key, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ if (oldVal == null) {
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, null);
+
+ expData.put(key, newVal);
+ }
+ else
+ checkNoEvent(evtsQueues);
+
+ break;
+ }
+
+ case 7: {
+ cache.getAndPutIfAbsent(key, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ if (oldVal == null) {
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, null);
+
+ expData.put(key, newVal);
+ }
+ else
+ checkNoEvent(evtsQueues);
+
+ break;
+ }
+
+ case 8: {
+ cache.replace(key, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ if (oldVal != null) {
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
+
+ expData.put(key, newVal);
+ }
+ else
+ checkNoEvent(evtsQueues);
+
+ break;
+ }
+
+ case 9: {
+ cache.getAndReplace(key, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ if (oldVal != null) {
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
+
+ expData.put(key, newVal);
+ }
+ else
+ checkNoEvent(evtsQueues);
+
+ break;
+ }
+
+ case 10: {
+ if (oldVal != null) {
+ Object replaceVal = value(rnd);
+
+ boolean success = replaceVal.equals(oldVal);
+
+ if (success) {
+ cache.replace(key, replaceVal, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ updatePartitionCounter(cache, key, partCntr);
+
+ waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
+
+ expData.put(key, newVal);
+ }
+ else {
+ cache.replace(key, replaceVal, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ checkNoEvent(evtsQueues);
+ }
+ }
+ else {
+ cache.replace(key, value(rnd), newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ checkNoEvent(evtsQueues);
+ }
+
+ break;
+ }
+
+ default:
+ fail("Op:" + op);
+ }
+ }
+ finally {
+ if (tx != null)
+ tx.close();
+ }
+ }
+
+ /**
+ * @param rnd {@link Random}.
+ * @return {@link TransactionIsolation}.
+ */
+ private TransactionIsolation txRandomIsolation(Random rnd) {
+ int val = rnd.nextInt(3);
+
+ if (val == 0)
+ return READ_COMMITTED;
+ else if (val == 1)
+ return REPEATABLE_READ;
+ else
+ return SERIALIZABLE;
+ }
+
+ /**
+ * @param rnd {@link Random}.
+ * @return {@link TransactionConcurrency}.
+ */
+ private TransactionConcurrency txRandomConcurrency(Random rnd) {
+ return rnd.nextBoolean() ? TransactionConcurrency.OPTIMISTIC : TransactionConcurrency.PESSIMISTIC;
+ }
+
+ /**
+ * @param cache Cache.
+ * @param key Key
+ * @param cntrs Partition counters.
+ */
+ private void updatePartitionCounter(IgniteCache<Object, Object> cache, Object key, Map<Integer, Long> cntrs) {
+ Affinity<Object> aff = cache.unwrap(Ignite.class).affinity(cache.getName());
+
+ int part = aff.partition(key);
+
+ Long partCntr = cntrs.get(part);
+
+ if (partCntr == null)
+ partCntr = 0L;
+
+ cntrs.put(part, ++partCntr);
+ }
+
+ /**
+ * @param rnd Random generator.
+ * @return Cache value.
+ */
+ private static Object value(Random rnd) {
+ return new QueryTestValue(rnd.nextInt(VALS));
+ }
+
+ /**
+ * @param evtsQueues Event queue.
+ * @param partCntrs Partition counters.
+ * @param aff Affinity function.
+ * @param key Key.
+ * @param val Value.
+ * @param oldVal Old value.
+ * @throws Exception If failed.
+ */
+ private void waitAndCheckEvent(List<BlockingQueue<CacheEntryEvent<?, ?>>> evtsQueues,
+ Map<Integer, Long> partCntrs,
+ Affinity<Object> aff,
+ Object key,
+ Object val,
+ Object oldVal)
+ throws Exception {
+ if ((val == null && oldVal == null
+ || (val != null && !isAccepted((QueryTestValue)val)))) {
+ checkNoEvent(evtsQueues);
+
+ return;
+ }
+
+ for (BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue : evtsQueues) {
+ CacheEntryEvent<?, ?> evt = evtsQueue.poll(5, SECONDS);
+
+ assertNotNull("Failed to wait for event [key=" + key + ", val=" + val + ", oldVal=" + oldVal + ']', evt);
+ assertEquals(key, evt.getKey());
+ assertEquals(val, evt.getValue());
+ assertEquals(oldVal, evt.getOldValue());
+
+ long cntr = partCntrs.get(aff.partition(key));
+ CacheQueryEntryEvent qryEntryEvt = evt.unwrap(CacheQueryEntryEvent.class);
+
+ assertNotNull(cntr);
+ assertNotNull(qryEntryEvt);
+
+ assertEquals(cntr, qryEntryEvt.getPartitionUpdateCounter());
+ }
+ }
+
+ /**
+ * @param evtsQueues Event queue.
+ * @throws Exception If failed.
+ */
+ private void checkNoEvent(List<BlockingQueue<CacheEntryEvent<?, ?>>> evtsQueues) throws Exception {
+ for (BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue : evtsQueues) {
+ CacheEntryEvent<?, ?> evt = evtsQueue.poll(50, MILLISECONDS);
+
+ assertNull(evt);
+ }
+ }
+
+ /**
+ *
+ */
+ protected static class NonSerializableFilter
+ implements CacheEntryEventSerializableFilter<CacheContinuousQueryRandomOperationsTest.QueryTestKey,
+ CacheContinuousQueryRandomOperationsTest.QueryTestValue>, Externalizable {
+ /** */
+ public NonSerializableFilter() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> evt) {
+ return isAccepted(evt.getValue());
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ fail("Entry filter should not be marshaled.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ fail("Entry filter should not be marshaled.");
+ }
+
+ /**
+ * @param val Value.
+ * @return {@code True} if value is even.
+ */
+ public static boolean isAccepted(QueryTestValue val) {
+ return val == null || val.val1 % 2 == 0;
+ }
+ }
+
+ /**
+ *
+ */
+ protected static class SerializableFilter implements CacheEntryEventSerializableFilter<Integer, Integer> {
+ /** */
+ public SerializableFilter() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends Integer> evt)
+ throws CacheEntryListenerException {
+ return isAccepted(evt.getValue());
+ }
+
+ /**
+ * @return {@code True} if value is even.
+ */
+ public static boolean isAccepted(Integer val) {
+ return val == null || val % 2 == 0;
+ }
+ }
+
+ /**
+ *
+ */
+ protected static class FilterFactory implements Factory<NonSerializableFilter> {
+ /** {@inheritDoc} */
+ @Override public NonSerializableFilter create() {
+ return new NonSerializableFilter();
+ }
+ }
+
+ /**
+ *
+ */
+ public abstract class LocalNonSerialiseListener implements
+ CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>,
+ CacheEntryCreatedListener<QueryTestKey, QueryTestValue>,
+ CacheEntryExpiredListener<QueryTestKey, QueryTestValue>,
+ CacheEntryRemovedListener<QueryTestKey, QueryTestValue>,
+ Externalizable {
+ /** */
+ public LocalNonSerialiseListener() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onCreated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
+ onEvents(evts);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onExpired(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
+ onEvents(evts);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onRemoved(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
+ onEvents(evts);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
+ onEvents(evts);
+ }
+
+ /**
+ * @param evts Events.
+ */
+ protected abstract void onEvents(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> evts);
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ throw new UnsupportedOperationException("Failed. Listener should not be marshaled.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ throw new UnsupportedOperationException("Failed. Listener should not be unmarshaled.");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryFilterTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryFilterTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryFilterTest.java
deleted file mode 100644
index 55340d5..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryFilterTest.java
+++ /dev/null
@@ -1,714 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache.query.continuous;
-
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-import java.util.UUID;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ThreadLocalRandom;
-import javax.cache.configuration.CacheEntryListenerConfiguration;
-import javax.cache.configuration.Factory;
-import javax.cache.configuration.FactoryBuilder;
-import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
-import javax.cache.event.CacheEntryCreatedListener;
-import javax.cache.event.CacheEntryEvent;
-import javax.cache.event.CacheEntryExpiredListener;
-import javax.cache.event.CacheEntryListenerException;
-import javax.cache.event.CacheEntryRemovedListener;
-import javax.cache.event.CacheEntryUpdatedListener;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
-import org.apache.ignite.cache.affinity.Affinity;
-import org.apache.ignite.cache.query.CacheQueryEntryEvent;
-import org.apache.ignite.cache.query.ContinuousQuery;
-import org.apache.ignite.cache.query.QueryCursor;
-import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.internal.util.typedef.T2;
-import org.apache.ignite.transactions.Transaction;
-import org.apache.ignite.transactions.TransactionConcurrency;
-import org.apache.ignite.transactions.TransactionIsolation;
-
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
-import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
-import static org.apache.ignite.cache.CacheMemoryMode.ONHEAP_TIERED;
-import static org.apache.ignite.cache.CacheMode.REPLICATED;
-import static org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFactoryFilterTest.NonSerializableFilter.isAccepted;
-import static org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryRandomOperationsTest.ContinuousDeploy.CLIENT;
-import static org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryRandomOperationsTest.ContinuousDeploy.SERVER;
-import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
-import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
-import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
-
-/**
- *
- */
-public class CacheContinuousQueryFactoryFilterTest extends CacheContinuousQueryRandomOperationsTest {
- /** */
- private static final int NODES = 5;
-
- /** */
- private static final int KEYS = 50;
-
- /** */
- private static final int VALS = 10;
-
- /** */
- public static final int ITERATION_CNT = 40;
-
- /**
- * @throws Exception If failed.
- */
- public void testInternalQuery() throws Exception {
- CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED,
- 1,
- ATOMIC,
- ONHEAP_TIERED,
- false);
-
- final IgniteCache<Object, Object> cache = grid(0).createCache(ccfg);
-
- UUID uuid = null;
-
- try {
- for (int i = 0; i < 10; i++)
- cache.put(i, i);
-
- final CountDownLatch latch = new CountDownLatch(5);
-
- CacheEntryUpdatedListener lsnr = new CacheEntryUpdatedListener() {
- @Override public void onUpdated(Iterable iterable) throws CacheEntryListenerException {
- for (Object evt : iterable) {
- latch.countDown();
-
- log.info("Received event: " + evt);
- }
- }
- };
-
- uuid = grid(0).context().cache().cache(cache.getName()).context().continuousQueries()
- .executeInternalQuery(lsnr, new SerializableFilter(), false, true, true);
-
- for (int i = 10; i < 20; i++)
- cache.put(i, i);
-
- assertTrue(latch.await(3, SECONDS));
- }
- finally {
- if (uuid != null)
- grid(0).context().cache().cache(cache.getName()).context().continuousQueries()
- .cancelInternalQuery(uuid);
-
- grid(0).destroyCache(ccfg.getName());
- }
- }
-
- /** {@inheritDoc} */
- @Override protected void doTestContinuousQuery(CacheConfiguration<Object, Object> ccfg, ContinuousDeploy deploy)
- throws Exception {
- ignite(0).createCache(ccfg);
-
- try {
- long seed = System.currentTimeMillis();
-
- Random rnd = new Random(seed);
-
- log.info("Random seed: " + seed);
-
- List<BlockingQueue<CacheEntryEvent<?, ?>>> evtsQueues = new ArrayList<>();
-
- Collection<QueryCursor<?>> curs = new ArrayList<>();
-
- Collection<T2<Integer, MutableCacheEntryListenerConfiguration>> lsnrCfgs = new ArrayList<>();
-
- if (deploy == CLIENT)
- evtsQueues.add(registerListener(ccfg.getName(), NODES - 1, curs, lsnrCfgs, rnd.nextBoolean()));
- else if (deploy == SERVER)
- evtsQueues.add(registerListener(ccfg.getName(), rnd.nextInt(NODES - 1), curs, lsnrCfgs,
- rnd.nextBoolean()));
- else {
- boolean isSync = rnd.nextBoolean();
-
- for (int i = 0; i < NODES - 1; i++)
- evtsQueues.add(registerListener(ccfg.getName(), i, curs, lsnrCfgs, isSync));
- }
-
- ConcurrentMap<Object, Object> expData = new ConcurrentHashMap<>();
-
- Map<Integer, Long> partCntr = new ConcurrentHashMap<>();
-
- try {
- for (int i = 0; i < ITERATION_CNT; i++) {
- if (i % 10 == 0)
- log.info("Iteration: " + i);
-
- for (int idx = 0; idx < NODES; idx++)
- randomUpdate(rnd, evtsQueues, expData, partCntr, grid(idx).cache(ccfg.getName()));
- }
- }
- finally {
- for (QueryCursor<?> cur : curs)
- cur.close();
-
- for (T2<Integer, MutableCacheEntryListenerConfiguration> e : lsnrCfgs)
- grid(e.get1()).cache(ccfg.getName()).deregisterCacheEntryListener(e.get2());
- }
- }
- finally {
- ignite(0).destroyCache(ccfg.getName());
- }
- }
-
- /**
- * @param cacheName Cache name.
- * @param nodeIdx Node index.
- * @param curs Cursors.
- * @param lsnrCfgs Listener configurations.
- * @return Event queue
- */
- private BlockingQueue<CacheEntryEvent<?, ?>> registerListener(String cacheName,
- int nodeIdx,
- Collection<QueryCursor<?>> curs,
- Collection<T2<Integer, MutableCacheEntryListenerConfiguration>> lsnrCfgs,
- boolean sync) {
- final BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue = new ArrayBlockingQueue<>(50_000);
-
- if (ThreadLocalRandom.current().nextBoolean()) {
- MutableCacheEntryListenerConfiguration<QueryTestKey, QueryTestValue> lsnrCfg =
- new MutableCacheEntryListenerConfiguration<>(
- FactoryBuilder.factoryOf(new LocalNonSerialiseListener() {
- @Override protected void onEvents(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> evts) {
- for (CacheEntryEvent<?, ?> evt : evts)
- evtsQueue.add(evt);
- }
- }),
- new FilterFactory(),
- true,
- sync
- );
-
- grid(nodeIdx).cache(cacheName).registerCacheEntryListener((CacheEntryListenerConfiguration)lsnrCfg);
-
- lsnrCfgs.add(new T2<Integer, MutableCacheEntryListenerConfiguration>(nodeIdx, lsnrCfg));
- }
- else {
- ContinuousQuery<QueryTestKey, QueryTestValue> qry = new ContinuousQuery<>();
-
- qry.setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
- @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
- for (CacheEntryEvent<?, ?> evt : evts)
- evtsQueue.add(evt);
- }
- });
-
- qry.setRemoteFilterFactory(new FilterFactory());
-
- QueryCursor<?> cur = grid(nodeIdx).cache(cacheName).query(qry);
-
- curs.add(cur);
- }
-
- return evtsQueue;
- }
-
- /**
- * @param rnd Random generator.
- * @param evtsQueues Events queue.
- * @param expData Expected cache data.
- * @param partCntr Partition counter.
- * @param cache Cache.
- * @throws Exception If failed.
- */
- private void randomUpdate(
- Random rnd,
- List<BlockingQueue<CacheEntryEvent<?, ?>>> evtsQueues,
- ConcurrentMap<Object, Object> expData,
- Map<Integer, Long> partCntr,
- IgniteCache<Object, Object> cache)
- throws Exception {
- Object key = new QueryTestKey(rnd.nextInt(KEYS));
- Object newVal = value(rnd);
- Object oldVal = expData.get(key);
-
- int op = rnd.nextInt(11);
-
- Ignite ignite = cache.unwrap(Ignite.class);
-
- Transaction tx = null;
-
- if (cache.getConfiguration(CacheConfiguration.class).getAtomicityMode() == TRANSACTIONAL && rnd.nextBoolean())
- tx = ignite.transactions().txStart(txRandomConcurrency(rnd), txRandomIsolation(rnd));
-
- try {
- // log.info("Random operation [key=" + key + ", op=" + op + ']');
-
- switch (op) {
- case 0: {
- cache.put(key, newVal);
-
- if (tx != null)
- tx.commit();
-
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
-
- expData.put(key, newVal);
-
- break;
- }
-
- case 1: {
- cache.getAndPut(key, newVal);
-
- if (tx != null)
- tx.commit();
-
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
-
- expData.put(key, newVal);
-
- break;
- }
-
- case 2: {
- cache.remove(key);
-
- if (tx != null)
- tx.commit();
-
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, null, oldVal);
-
- expData.remove(key);
-
- break;
- }
-
- case 3: {
- cache.getAndRemove(key);
-
- if (tx != null)
- tx.commit();
-
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, null, oldVal);
-
- expData.remove(key);
-
- break;
- }
-
- case 4: {
- cache.invoke(key, new EntrySetValueProcessor(newVal, rnd.nextBoolean()));
-
- if (tx != null)
- tx.commit();
-
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
-
- expData.put(key, newVal);
-
- break;
- }
-
- case 5: {
- cache.invoke(key, new EntrySetValueProcessor(null, rnd.nextBoolean()));
-
- if (tx != null)
- tx.commit();
-
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, null, oldVal);
-
- expData.remove(key);
-
- break;
- }
-
- case 6: {
- cache.putIfAbsent(key, newVal);
-
- if (tx != null)
- tx.commit();
-
- if (oldVal == null) {
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, null);
-
- expData.put(key, newVal);
- }
- else
- checkNoEvent(evtsQueues);
-
- break;
- }
-
- case 7: {
- cache.getAndPutIfAbsent(key, newVal);
-
- if (tx != null)
- tx.commit();
-
- if (oldVal == null) {
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, null);
-
- expData.put(key, newVal);
- }
- else
- checkNoEvent(evtsQueues);
-
- break;
- }
-
- case 8: {
- cache.replace(key, newVal);
-
- if (tx != null)
- tx.commit();
-
- if (oldVal != null) {
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
-
- expData.put(key, newVal);
- }
- else
- checkNoEvent(evtsQueues);
-
- break;
- }
-
- case 9: {
- cache.getAndReplace(key, newVal);
-
- if (tx != null)
- tx.commit();
-
- if (oldVal != null) {
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
-
- expData.put(key, newVal);
- }
- else
- checkNoEvent(evtsQueues);
-
- break;
- }
-
- case 10: {
- if (oldVal != null) {
- Object replaceVal = value(rnd);
-
- boolean success = replaceVal.equals(oldVal);
-
- if (success) {
- cache.replace(key, replaceVal, newVal);
-
- if (tx != null)
- tx.commit();
-
- updatePartitionCounter(cache, key, partCntr);
-
- waitAndCheckEvent(evtsQueues, partCntr, affinity(cache), key, newVal, oldVal);
-
- expData.put(key, newVal);
- }
- else {
- cache.replace(key, replaceVal, newVal);
-
- if (tx != null)
- tx.commit();
-
- checkNoEvent(evtsQueues);
- }
- }
- else {
- cache.replace(key, value(rnd), newVal);
-
- if (tx != null)
- tx.commit();
-
- checkNoEvent(evtsQueues);
- }
-
- break;
- }
-
- default:
- fail("Op:" + op);
- }
- } finally {
- if (tx != null)
- tx.close();
- }
- }
-
- /**
- * @param rnd {@link Random}.
- * @return {@link TransactionIsolation}.
- */
- private TransactionIsolation txRandomIsolation(Random rnd) {
- int val = rnd.nextInt(3);
-
- if (val == 0)
- return READ_COMMITTED;
- else if (val == 1)
- return REPEATABLE_READ;
- else
- return SERIALIZABLE;
- }
-
- /**
- * @param rnd {@link Random}.
- * @return {@link TransactionConcurrency}.
- */
- private TransactionConcurrency txRandomConcurrency(Random rnd) {
- return rnd.nextBoolean() ? TransactionConcurrency.OPTIMISTIC : TransactionConcurrency.PESSIMISTIC;
- }
-
- /**
- * @param cache Cache.
- * @param key Key
- * @param cntrs Partition counters.
- */
- private void updatePartitionCounter(IgniteCache<Object, Object> cache, Object key, Map<Integer, Long> cntrs) {
- Affinity<Object> aff = cache.unwrap(Ignite.class).affinity(cache.getName());
-
- int part = aff.partition(key);
-
- Long partCntr = cntrs.get(part);
-
- if (partCntr == null)
- partCntr = 0L;
-
- cntrs.put(part, ++partCntr);
- }
-
- /**
- * @param rnd Random generator.
- * @return Cache value.
- */
- private static Object value(Random rnd) {
- return new QueryTestValue(rnd.nextInt(VALS));
- }
-
- /**
- * @param evtsQueues Event queue.
- * @param partCntrs Partition counters.
- * @param aff Affinity function.
- * @param key Key.
- * @param val Value.
- * @param oldVal Old value.
- * @throws Exception If failed.
- */
- private void waitAndCheckEvent(List<BlockingQueue<CacheEntryEvent<?, ?>>> evtsQueues,
- Map<Integer, Long> partCntrs,
- Affinity<Object> aff,
- Object key,
- Object val,
- Object oldVal)
- throws Exception {
- if ((val == null && oldVal == null
- || (val != null && !isAccepted((QueryTestValue)val)))) {
- checkNoEvent(evtsQueues);
-
- return;
- }
-
- for (BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue : evtsQueues) {
- CacheEntryEvent<?, ?> evt = evtsQueue.poll(5, SECONDS);
-
- assertNotNull("Failed to wait for event [key=" + key + ", val=" + val + ", oldVal=" + oldVal + ']', evt);
- assertEquals(key, evt.getKey());
- assertEquals(val, evt.getValue());
- assertEquals(oldVal, evt.getOldValue());
-
- long cntr = partCntrs.get(aff.partition(key));
- CacheQueryEntryEvent qryEntryEvt = evt.unwrap(CacheQueryEntryEvent.class);
-
- assertNotNull(cntr);
- assertNotNull(qryEntryEvt);
-
- assertEquals(cntr, qryEntryEvt.getPartitionUpdateCounter());
- }
- }
-
- /**
- * @param evtsQueues Event queue.
- * @throws Exception If failed.
- */
- private void checkNoEvent(List<BlockingQueue<CacheEntryEvent<?, ?>>> evtsQueues) throws Exception {
- for (BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue : evtsQueues) {
- CacheEntryEvent<?, ?> evt = evtsQueue.poll(50, MILLISECONDS);
-
- assertNull(evt);
- }
- }
-
- /**
- *
- */
- protected static class NonSerializableFilter
- implements CacheEntryEventSerializableFilter<CacheContinuousQueryRandomOperationsTest.QueryTestKey,
- CacheContinuousQueryRandomOperationsTest.QueryTestValue>, Externalizable {
- /** */
- public NonSerializableFilter() {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public boolean evaluate(CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> event)
- throws CacheEntryListenerException {
- return isAccepted(event.getValue());
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- fail("Entry filter should not be marshaled.");
- }
-
- /** {@inheritDoc} */
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- fail("Entry filter should not be marshaled.");
- }
-
- /**
- * @return {@code True} if value is even.
- */
- public static boolean isAccepted(QueryTestValue val) {
- return val == null || val.val1 % 2 == 0;
- }
- }
-
- /**
- *
- */
- protected static class SerializableFilter implements CacheEntryEventSerializableFilter<Integer, Integer>{
- /** */
- public SerializableFilter() {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends Integer> event)
- throws CacheEntryListenerException {
- return isAccepted(event.getValue());
- }
-
- /**
- * @return {@code True} if value is even.
- */
- public static boolean isAccepted(Integer val) {
- return val == null || val % 2 == 0;
- }
- }
-
- /**
- *
- */
- protected static class FilterFactory implements Factory<NonSerializableFilter> {
- @Override public NonSerializableFilter create() {
- return new NonSerializableFilter();
- }
- }
-
- /**
- *
- */
- public abstract class LocalNonSerialiseListener implements
- CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>,
- CacheEntryCreatedListener<QueryTestKey, QueryTestValue>,
- CacheEntryExpiredListener<QueryTestKey, QueryTestValue>,
- CacheEntryRemovedListener<QueryTestKey, QueryTestValue>,
- Externalizable {
- /** */
- public LocalNonSerialiseListener() {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void onCreated(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
- onEvents(evts);
- }
-
- /** {@inheritDoc} */
- @Override public void onExpired(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
- onEvents(evts);
- }
-
- /** {@inheritDoc} */
- @Override public void onRemoved(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
- onEvents(evts);
- }
-
- /** {@inheritDoc} */
- @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> evts) throws CacheEntryListenerException {
- onEvents(evts);
- }
-
- /**
- * @param evts Events.
- */
- protected abstract void onEvents(Iterable<CacheEntryEvent<? extends QueryTestKey,
- ? extends QueryTestValue>> evts);
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- throw new UnsupportedOperationException("Failed. Listener should not be marshaled.");
- }
-
- /** {@inheritDoc} */
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- throw new UnsupportedOperationException("Failed. Listener should not be unmarshaled.");
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
index 4226537..083367c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
@@ -81,6 +81,7 @@ import org.apache.ignite.internal.util.typedef.PA;
import org.apache.ignite.internal.util.typedef.PAX;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.T3;
+import org.apache.ignite.lang.IgniteAsyncCallback;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -99,7 +100,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.PRIMARY;
-import static org.apache.ignite.cache.CacheMemoryMode.*;
+import static org.apache.ignite.cache.CacheMemoryMode.ONHEAP_TIERED;
import static org.apache.ignite.cache.CacheMode.REPLICATED;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
@@ -167,6 +168,13 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
}
/**
+ * @return Async callback flag.
+ */
+ protected boolean asyncCallback() {
+ return false;
+ }
+
+ /**
* @return Near cache configuration.
*/
protected NearCacheConfiguration nearCacheConfiguration() {
@@ -476,7 +484,8 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
for (int j = 0; j < 50; ++j) {
ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
- final CacheEventListener3 lsnr = new CacheEventListener3();
+ final CacheEventListener3 lsnr = asyncCallback() ? new CacheEventAsyncListener3()
+ : new CacheEventListener3();
qry.setLocalListener(lsnr);
@@ -560,7 +569,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
- final CacheEventListener3 lsnr = new CacheEventListener3();
+ final CacheEventListener3 lsnr = asyncCallback() ? new CacheEventAsyncListener3() : new CacheEventListener3();
qry.setLocalListener(lsnr);
@@ -721,7 +730,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
- final CacheEventListener3 lsnr = new CacheEventListener3();
+ final CacheEventListener3 lsnr = asyncCallback() ? new CacheEventAsyncListener3() : new CacheEventListener3();
qry.setLocalListener(lsnr);
@@ -841,7 +850,8 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
Affinity<Object> aff = qryClient.affinity(null);
- CacheEventListener1 lsnr = new CacheEventListener1(false);
+ CacheEventListener1 lsnr = asyncCallback() ? new CacheEventAsyncListener1(false)
+ : new CacheEventListener1(false);
ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
@@ -1545,7 +1555,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
qry.setLocalListener(lsnr);
- qry.setRemoteFilter(new CacheEventFilter());
+ qry.setRemoteFilter(asyncCallback() ? new CacheEventAsyncFilter() : new CacheEventFilter());
QueryCursor<?> cur = qryClnCache.query(qry);
@@ -1639,7 +1649,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
newQry.setLocalListener(dinLsnr);
- newQry.setRemoteFilter(new CacheEventFilter());
+ newQry.setRemoteFilter(asyncCallback() ? new CacheEventAsyncFilter() : new CacheEventFilter());
dinQry = qryClnCache.query(newQry);
@@ -1786,7 +1796,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
final IgniteCache<Object, Object> qryClnCache = qryCln.cache(null);
- final CacheEventListener2 lsnr = new CacheEventListener2();
+ final CacheEventListener2 lsnr = asyncCallback() ? new CacheEventAsyncListener2() : new CacheEventListener2();
ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
@@ -2144,6 +2154,19 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
/**
*
*/
+ @IgniteAsyncCallback
+ private static class CacheEventAsyncListener1 extends CacheEventListener1 {
+ /**
+ * @param saveAll Save all events flag.
+ */
+ CacheEventAsyncListener1(boolean saveAll) {
+ super(saveAll);
+ }
+ }
+
+ /**
+ *
+ */
private static class CacheEventListener1 implements CacheEntryUpdatedListener<Object, Object> {
/** */
private volatile CountDownLatch latch;
@@ -2208,6 +2231,14 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
/**
*
*/
+ @IgniteAsyncCallback
+ private static class CacheEventAsyncListener2 extends CacheEventListener2 {
+ // No-op.
+ }
+
+ /**
+ *
+ */
private static class CacheEventListener2 implements CacheEntryUpdatedListener<Object, Object> {
/** */
@LoggerResource
@@ -2275,6 +2306,14 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
/**
*
*/
+ @IgniteAsyncCallback
+ public static class CacheEventAsyncListener3 extends CacheEventListener3 {
+ // No-op.
+ }
+
+ /**
+ *
+ */
public static class CacheEventListener3 implements CacheEntryUpdatedListener<Object, Object>,
CacheEntryEventSerializableFilter<Object, Object> {
/** Keys. */
@@ -2303,6 +2342,14 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
/**
*
*/
+ @IgniteAsyncCallback
+ private static class CacheEventAsyncFilter extends CacheEventFilter {
+ // No-op.
+ }
+
+ /**
+ *
+ */
public static class CacheEventFilter implements CacheEntryEventSerializableFilter<Object, Object> {
/** {@inheritDoc} */
@Override public boolean evaluate(CacheEntryEvent<?, ?> evt) throws CacheEntryListenerException {
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java
index 025dd80..b469a86 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryLostPartitionTest.java
@@ -60,6 +60,12 @@ public class CacheContinuousQueryLostPartitionTest extends GridCommonAbstractTes
super.beforeTest();
startGridsMultiThreaded(2);
+
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return grid(0).cluster().nodes().size() == 2;
+ }
+ }, 10000L);
}
/** {@inheritDoc} */
@@ -140,6 +146,14 @@ public class CacheContinuousQueryLostPartitionTest extends GridCommonAbstractTes
// node2 now becomes the primary for the key.
stopGrid(0);
+ final int prevSize = grid(1).cluster().nodes().size();
+
+ GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return prevSize - 1 == grid(1).cluster().nodes().size();
+ }
+ }, 5000L);
+
cache2.put(key, "2");
// Sanity check.
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java
new file mode 100644
index 0000000..0d027a9
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java
@@ -0,0 +1,627 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.cache.configuration.FactoryBuilder;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryUpdatedListener;
+import javax.cache.processor.EntryProcessorException;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.CacheEntryProcessor;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.PA;
+import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.spi.eventstorage.memory.MemoryEventStorageSpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.PRIMARY;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC;
+
+/**
+ *
+ */
+public class CacheContinuousQueryOperationFromCallbackTest extends GridCommonAbstractTest {
+ /** */
+ public static final int KEYS = 10;
+
+ /** */
+ public static final int KEYS_FROM_CALLBACK = 20;
+
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int NODES = 5;
+
+ /** */
+ public static final int ITERATION_CNT = 20;
+
+ /** */
+ public static final int SYSTEM_POOL_SIZE = 10;
+
+ /** */
+ private boolean client;
+
+ /** */
+ private static AtomicInteger filterCbCntr = new AtomicInteger(0);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ cfg.setSystemThreadPoolSize(SYSTEM_POOL_SIZE);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+ ((TcpCommunicationSpi)cfg.getCommunicationSpi()).setSharedMemoryPort(-1);
+
+ cfg.setClientMode(client);
+
+ MemoryEventStorageSpi storeSpi = new MemoryEventStorageSpi();
+ storeSpi.setExpireCount(100);
+
+ cfg.setEventStorageSpi(storeSpi);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGridsMultiThreaded(NODES - 1);
+
+ client = true;
+
+ startGrid(NODES - 1);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ super.afterTestsStopped();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ filterCbCntr.set(0);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicTwoBackups() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, ATOMIC, FULL_SYNC);
+
+ doTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxTwoBackupsFilter() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, FULL_SYNC);
+
+ doTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxTwoBackupsFilterPrimary() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, PRIMARY_SYNC);
+
+ doTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxReplicatedFilter() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 0, TRANSACTIONAL, FULL_SYNC);
+
+ doTest(ccfg, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxTwoBackup() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, FULL_SYNC);
+
+ doTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxReplicated() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 2, TRANSACTIONAL, FULL_SYNC);
+
+ doTest(ccfg, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxReplicatedPrimary() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 2, TRANSACTIONAL, PRIMARY_SYNC);
+
+ doTest(ccfg, true);
+ }
+
+ /**
+ * @param ccfg Cache configuration.
+ * @throws Exception If failed.
+ */
+ protected void doTest(final CacheConfiguration ccfg, boolean fromLsnr) throws Exception {
+ ignite(0).createCache(ccfg);
+
+ List<QueryCursor<?>> qries = new ArrayList<>();
+
+ assertEquals(0, filterCbCntr.get());
+
+ try {
+ List<Set<T2<QueryTestKey, QueryTestValue>>> rcvdEvts = new ArrayList<>(NODES);
+ List<Set<T2<QueryTestKey, QueryTestValue>>> evtsFromCallbacks = new ArrayList<>(NODES);
+
+ final AtomicInteger qryCntr = new AtomicInteger(0);
+
+ final AtomicInteger cbCntr = new AtomicInteger(0);
+
+ final int threadCnt = SYSTEM_POOL_SIZE * 2;
+
+ for (int idx = 0; idx < NODES; idx++) {
+ Set<T2<QueryTestKey, QueryTestValue>> evts = Collections.
+ newSetFromMap(new ConcurrentHashMap<T2<QueryTestKey, QueryTestValue>, Boolean>());
+ Set<T2<QueryTestKey, QueryTestValue>> evtsFromCb = Collections.
+ newSetFromMap(new ConcurrentHashMap<T2<QueryTestKey, QueryTestValue>, Boolean>());
+
+ IgniteCache<Object, Object> cache = grid(idx).getOrCreateCache(ccfg.getName());
+
+ ContinuousQuery qry = new ContinuousQuery();
+
+ qry.setLocalListener(new TestCacheAsyncEventListener(evts, evtsFromCb,
+ fromLsnr ? cache : null, qryCntr, cbCntr));
+
+ if (!fromLsnr)
+ qry.setRemoteFilterFactory(
+ FactoryBuilder.factoryOf(new CacheTestRemoteFilterAsync(ccfg.getName())));
+
+ rcvdEvts.add(evts);
+ evtsFromCallbacks.add(evtsFromCb);
+
+ QueryCursor qryCursor = cache.query(qry);
+
+ qries.add(qryCursor);
+ }
+
+ IgniteInternalFuture<Long> f = GridTestUtils.runMultiThreadedAsync(new Runnable() {
+ @Override public void run() {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (int i = 0; i < ITERATION_CNT; i++) {
+ IgniteCache<QueryTestKey, QueryTestValue> cache =
+ grid(rnd.nextInt(NODES)).cache(ccfg.getName());
+
+ QueryTestKey key = new QueryTestKey(rnd.nextInt(KEYS));
+
+ boolean startTx = cache.getConfiguration(CacheConfiguration.class).getAtomicityMode() ==
+ TRANSACTIONAL && rnd.nextBoolean();
+
+ Transaction tx = null;
+
+ if (startTx)
+ tx = cache.unwrap(Ignite.class).transactions().txStart();
+
+ try {
+ if ((cache.get(key) == null) || rnd.nextBoolean())
+ cache.invoke(key, new IncrementTestEntryProcessor());
+ else {
+ QueryTestValue val;
+ QueryTestValue newVal;
+
+ do {
+ val = cache.get(key);
+
+ newVal = val == null ?
+ new QueryTestValue(0) : new QueryTestValue(val.val1 + 1);
+ }
+ while (!cache.replace(key, val, newVal));
+ }
+ }
+ finally {
+ if (tx != null)
+ tx.commit();
+ }
+ }
+ }
+ }, threadCnt, "put-thread");
+
+ f.get(30, TimeUnit.SECONDS);
+
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return qryCntr.get() >= ITERATION_CNT * threadCnt * NODES;
+ }
+ }, TimeUnit.MINUTES.toMillis(2));
+
+ for (Set<T2<QueryTestKey, QueryTestValue>> set : rcvdEvts)
+ checkEvents(set, ITERATION_CNT * threadCnt, grid(0).cache(ccfg.getName()), false);
+
+ if (fromLsnr) {
+ final int expCnt = qryCntr.get() * NODES * KEYS_FROM_CALLBACK;
+
+ boolean res = GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return cbCntr.get() >= expCnt;
+ }
+ }, TimeUnit.SECONDS.toMillis(60));
+
+ assertTrue("Failed to wait events [exp=" + expCnt + ", act=" + cbCntr.get() + "]", res);
+
+ assertEquals(expCnt, cbCntr.get());
+
+ for (Set<T2<QueryTestKey, QueryTestValue>> set : evtsFromCallbacks)
+ checkEvents(set, qryCntr.get() * KEYS_FROM_CALLBACK, grid(0).cache(ccfg.getName()), true);
+ }
+ else {
+ final int expInvkCnt = ITERATION_CNT * threadCnt *
+ (ccfg.getCacheMode() != REPLICATED ? (ccfg.getBackups() + 1) : NODES - 1) * NODES;
+
+ GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return filterCbCntr.get() >= expInvkCnt;
+ }
+ }, TimeUnit.SECONDS.toMillis(60));
+
+ assertEquals(expInvkCnt, filterCbCntr.get());
+
+ for (Set<T2<QueryTestKey, QueryTestValue>> set : evtsFromCallbacks)
+ checkEvents(set, expInvkCnt * KEYS_FROM_CALLBACK, grid(0).cache(ccfg.getName()), true);
+ }
+ }
+ finally {
+ for (QueryCursor<?> qry : qries)
+ qry.close();
+
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @param expCnt Expected count.
+ * @param cache Cache.
+ * @param set Received events.
+ * @throws Exception If failed.
+ */
+ private void checkEvents(final Set<T2<QueryTestKey, QueryTestValue>> set, final int expCnt, IgniteCache cache,
+ boolean cb) throws Exception {
+ assertTrue("Expected size: " + expCnt + ", actual: " + set.size(), GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return set.size() >= expCnt;
+ }
+ }, 10000L));
+
+ int startKey = cb ? KEYS : 0;
+ int endKey = cb ? KEYS + KEYS_FROM_CALLBACK : KEYS;
+
+ for (int i = startKey; i < endKey; i++) {
+ QueryTestKey key = new QueryTestKey(i);
+
+ QueryTestValue maxVal = (QueryTestValue)cache.get(key);
+
+ for (int val = 0; val <= maxVal.val1; val++)
+ assertTrue(set.remove(new T2<>(key, new QueryTestValue(val))));
+ }
+
+ assertTrue(set.isEmpty());
+ }
+
+ /**
+ *
+ */
+ private static class IncrementTestEntryProcessor implements
+ CacheEntryProcessor<QueryTestKey, QueryTestValue, Object> {
+ /** {@inheritDoc} */
+ @Override public Object process(MutableEntry<QueryTestKey, QueryTestValue> entry, Object... arguments)
+ throws EntryProcessorException {
+ if (entry.exists())
+ entry.setValue(new QueryTestValue(entry.getValue().val1 + 1));
+ else
+ entry.setValue(new QueryTestValue(0));
+
+ return null;
+ }
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ private static class CacheTestRemoteFilterAsync implements
+ CacheEntryEventSerializableFilter<QueryTestKey, QueryTestValue> {
+ /** */
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** */
+ private String cacheName;
+
+ /**
+ * @param cacheName Cache name.
+ */
+ public CacheTestRemoteFilterAsync(String cacheName) {
+ this.cacheName = cacheName;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e)
+ throws CacheEntryListenerException {
+ if (e.getKey().compareTo(new QueryTestKey(KEYS)) < 0) {
+ IgniteCache<QueryTestKey, QueryTestValue> cache = ignite.cache(cacheName);
+
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ Set<QueryTestKey> keys = new LinkedHashSet<>();
+
+ for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
+ keys.add(new QueryTestKey(key));
+
+ cache.invokeAll(keys, new IncrementTestEntryProcessor());
+ }
+ else {
+ for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
+ cache.invoke(new QueryTestKey(key), new IncrementTestEntryProcessor());
+ }
+
+ filterCbCntr.incrementAndGet();
+ }
+
+ return true;
+ }
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ private static class TestCacheAsyncEventListener
+ implements CacheEntryUpdatedListener<QueryTestKey, QueryTestValue> {
+ /** */
+ private final Set<T2<QueryTestKey, QueryTestValue>> rcvsEvts;
+
+ /** */
+ private final AtomicInteger cntr;
+
+ /** */
+ private final AtomicInteger cbCntr;
+
+ /** */
+ private final Set<T2<QueryTestKey, QueryTestValue>> evtsFromCb;
+
+ /** */
+ private IgniteCache<QueryTestKey, QueryTestValue> cache;
+
+ /**
+ * @param rcvsEvts Set for received events.
+ * @param evtsFromCb Set for received events.
+ * @param cache Ignite cache.
+ * @param cntr Received events counter.
+ * @param cbCntr Received events counter from callbacks.
+ */
+ public TestCacheAsyncEventListener(Set<T2<QueryTestKey, QueryTestValue>> rcvsEvts,
+ Set<T2<QueryTestKey, QueryTestValue>> evtsFromCb,
+ @Nullable IgniteCache cache,
+ AtomicInteger cntr,
+ AtomicInteger cbCntr) {
+ this.rcvsEvts = rcvsEvts;
+ this.evtsFromCb = evtsFromCb;
+ this.cache = cache;
+ this.cntr = cntr;
+ this.cbCntr = cbCntr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> evts)
+ throws CacheEntryListenerException {
+ for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : evts) {
+ if (e.getKey().compareTo(new QueryTestKey(KEYS)) < 0) {
+ rcvsEvts.add(new T2<>(e.getKey(), e.getValue()));
+
+ cntr.incrementAndGet();
+
+ if (cache != null) {
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ Set<QueryTestKey> keys = new LinkedHashSet<>();
+
+ for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
+ keys.add(new QueryTestKey(key));
+
+ cache.invokeAll(keys, new IncrementTestEntryProcessor());
+ }
+ else {
+ for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
+ cache.invoke(new QueryTestKey(key), new IncrementTestEntryProcessor());
+ }
+ }
+ }
+ else {
+ evtsFromCb.add(new T2<>(e.getKey(), e.getValue()));
+
+ cbCntr.incrementAndGet();
+ }
+ }
+ }
+ }
+
+ /**
+ * @param cacheMode Cache mode.
+ * @param backups Number of backups.
+ * @param atomicityMode Cache atomicity mode.
+ * @param writeMode Write sync mode.
+ * @return Cache configuration.
+ */
+ protected CacheConfiguration<Object, Object> cacheConfiguration(
+ CacheMode cacheMode,
+ int backups,
+ CacheAtomicityMode atomicityMode,
+ CacheWriteSynchronizationMode writeMode) {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
+
+ ccfg.setName("test-cache-" + atomicityMode + "-" + cacheMode + "-" + writeMode + "-" + backups);
+ ccfg.setAtomicityMode(atomicityMode);
+ ccfg.setCacheMode(cacheMode);
+ ccfg.setWriteSynchronizationMode(writeMode);
+ ccfg.setAtomicWriteOrderMode(PRIMARY);
+
+ if (cacheMode == PARTITIONED)
+ ccfg.setBackups(backups);
+
+ return ccfg;
+ }
+
+ /**
+ *
+ */
+ public static class QueryTestKey implements Serializable, Comparable {
+ /** */
+ private final Integer key;
+
+ /**
+ * @param key Key.
+ */
+ public QueryTestKey(Integer key) {
+ this.key = key;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ QueryTestKey that = (QueryTestKey)o;
+
+ return key.equals(that.key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return key.hashCode();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(QueryTestKey.class, this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int compareTo(Object o) {
+ return key - ((QueryTestKey)o).key;
+ }
+ }
+
+ /**
+ *
+ */
+ public static class QueryTestValue implements Serializable {
+ /** */
+ @GridToStringInclude
+ protected final Integer val1;
+
+ /** */
+ @GridToStringInclude
+ protected final String val2;
+
+ /**
+ * @param val Value.
+ */
+ public QueryTestValue(Integer val) {
+ this.val1 = val;
+ this.val2 = String.valueOf(val);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ QueryTestValue that = (QueryTestValue) o;
+
+ return val1.equals(that.val1) && val2.equals(that.val2);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = val1.hashCode();
+
+ res = 31 * res + val2.hashCode();
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(QueryTestValue.class, this);
+ }
+ }
+}
[3/4] ignite git commit: IGNITE-2004 Fixed "Asynchronous execution of
ContinuousQuery's remote filter & local list".
Posted by nt...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
index c01f636..9efc456 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
@@ -54,11 +54,13 @@ import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheManagerAdapter;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateFuture;
import org.apache.ignite.internal.processors.continuous.GridContinuousHandler;
import org.apache.ignite.internal.processors.continuous.GridContinuousProcessor;
import org.apache.ignite.internal.util.typedef.CI2;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteAsyncCallback;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.plugin.security.SecurityPermission;
@@ -166,23 +168,8 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
* @param key Entry key.
* @param partId Partition id.
* @param updCntr Updated counter.
- * @param topVer Topology version.
- */
- public void skipUpdateEvent(Map<UUID, CacheContinuousQueryListener> lsnrs,
- KeyCacheObject key,
- int partId,
- long updCntr,
- AffinityTopologyVersion topVer) {
- skipUpdateEvent(lsnrs, key, partId, updCntr, true, topVer);
- }
-
- /**
- * @param lsnrs Listeners to notify.
- * @param key Entry key.
- * @param partId Partition id.
- * @param updCntr Updated counter.
- * @param topVer Topology version.
* @param primary Primary.
+ * @param topVer Topology version.
*/
public void skipUpdateEvent(Map<UUID, CacheContinuousQueryListener> lsnrs,
KeyCacheObject key,
@@ -241,6 +228,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
* @param primary {@code True} if called on primary node.
* @param preload Whether update happened during preloading.
* @param updateCntr Update counter.
+ * @param fut Dht atomic future.
* @param topVer Topology version.
* @throws IgniteCheckedException In case of error.
*/
@@ -253,7 +241,9 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
boolean primary,
boolean preload,
long updateCntr,
- AffinityTopologyVersion topVer) throws IgniteCheckedException {
+ @Nullable GridDhtAtomicUpdateFuture fut,
+ AffinityTopologyVersion topVer
+ ) throws IgniteCheckedException {
Map<UUID, CacheContinuousQueryListener> lsnrCol = updateListeners(internal, preload);
if (lsnrCol != null) {
@@ -267,6 +257,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
primary,
preload,
updateCntr,
+ fut,
topVer);
}
}
@@ -282,6 +273,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
* @param preload Whether update happened during preloading.
* @param updateCntr Update counter.
* @param topVer Topology version.
+ * @param fut Dht atomic future.
* @throws IgniteCheckedException In case of error.
*/
public void onEntryUpdated(
@@ -294,6 +286,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
boolean primary,
boolean preload,
long updateCntr,
+ @Nullable GridDhtAtomicUpdateFuture fut,
AffinityTopologyVersion topVer)
throws IgniteCheckedException
{
@@ -347,7 +340,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
CacheContinuousQueryEvent evt = new CacheContinuousQueryEvent<>(
cctx.kernalContext().cache().jcache(cctx.name()), cctx, e0);
- lsnr.onEntryUpdated(evt, primary, recordIgniteEvt);
+ lsnr.onEntryUpdated(evt, primary, recordIgniteEvt, fut);
}
}
@@ -401,7 +394,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
CacheContinuousQueryEvent evt = new CacheContinuousQueryEvent(
cctx.kernalContext().cache().jcache(cctx.name()), cctx, e0);
- lsnr.onEntryUpdated(evt, primary, recordIgniteEvt);
+ lsnr.onEntryUpdated(evt, primary, recordIgniteEvt, null);
}
}
}
@@ -511,7 +504,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
return executeQuery0(
locLsnr,
new IgniteClosure<Boolean, CacheContinuousQueryHandler>() {
- @Override public CacheContinuousQueryHandler apply(Boolean aBoolean) {
+ @Override public CacheContinuousQueryHandler apply(Boolean v2) {
return new CacheContinuousQueryHandler(
cctx.name(),
TOPIC_CACHE.topic(topicPrefix, cctx.localNodeId(), seq.getAndIncrement()),
@@ -800,6 +793,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
}
/**
+ *
*/
private class JCacheQuery {
/** */
@@ -931,9 +925,9 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
/**
*
*/
- private static class JCacheQueryLocalListener<K, V> implements CacheEntryUpdatedListener<K, V> {
+ static class JCacheQueryLocalListener<K, V> implements CacheEntryUpdatedListener<K, V> {
/** */
- private final CacheEntryListener<K, V> impl;
+ final CacheEntryListener<K, V> impl;
/** */
private final IgniteLogger log;
@@ -957,28 +951,28 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
try {
switch (evt.getEventType()) {
case CREATED:
- assert impl instanceof CacheEntryCreatedListener;
+ assert impl instanceof CacheEntryCreatedListener : evt;
((CacheEntryCreatedListener<K, V>)impl).onCreated(singleton(evt));
break;
case UPDATED:
- assert impl instanceof CacheEntryUpdatedListener;
+ assert impl instanceof CacheEntryUpdatedListener : evt;
((CacheEntryUpdatedListener<K, V>)impl).onUpdated(singleton(evt));
break;
case REMOVED:
- assert impl instanceof CacheEntryRemovedListener;
+ assert impl instanceof CacheEntryRemovedListener : evt;
((CacheEntryRemovedListener<K, V>)impl).onRemoved(singleton(evt));
break;
case EXPIRED:
- assert impl instanceof CacheEntryExpiredListener;
+ assert impl instanceof CacheEntryExpiredListener : evt;
((CacheEntryExpiredListener<K, V>)impl).onExpired(singleton(evt));
@@ -1009,6 +1003,13 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
return evts;
}
+
+ /**
+ * @return {@code True} if listener should be executed in non-system thread.
+ */
+ protected boolean async() {
+ return U.hasAnnotation(impl, IgniteAsyncCallback.class);
+ }
}
/**
@@ -1019,7 +1020,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
private static final long serialVersionUID = 0L;
/** */
- private CacheEntryEventFilter impl;
+ protected CacheEntryEventFilter impl;
/** */
private byte types;
@@ -1072,6 +1073,13 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter {
}
/**
+ * @return {@code True} if filter should be executed in non-system thread.
+ */
+ protected boolean async() {
+ return U.hasAnnotation(impl, IgniteAsyncCallback.class);
+ }
+
+ /**
* @param evtType Type.
* @return Flag value.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/lang/IgniteAsyncCallback.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/IgniteAsyncCallback.java b/modules/core/src/main/java/org/apache/ignite/lang/IgniteAsyncCallback.java
new file mode 100644
index 0000000..1e04ce6
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/lang/IgniteAsyncCallback.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.lang;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.CacheEntryListener;
+import javax.cache.event.CacheEntryUpdatedListener;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.configuration.IgniteConfiguration;
+
+/**
+ * If callback has this annotation then it will be executing in another thread.
+ * <p>
+ * Currently this annotation is supported for:
+ * <ol>
+ * <li>{@link ContinuousQuery} - {@link CacheEntryUpdatedListener} and {@link CacheEntryEventFilter}.</li>
+ * </ol>
+ * <p>
+ * For example, if {@link CacheEntryEventFilter filter} or {@link CacheEntryListener}
+ * has the annotation then callbacks will be executing to asyncCallback thread pool. It allows to use cache API
+ * in a callbacks. This thread pool can be configured by {@link IgniteConfiguration#setAsyncCallbackPoolSize(int)}.
+ * <h1 class="header">Example</h1>
+ * As an example, suppose we have cache with {@code 'Person'} objects and we need
+ * to query all persons with salary above then 1000. Also remote filter will update some entries.
+ * <p>
+ * Here is the {@code Person} class:
+ * <pre name="code" class="java">
+ * public class Person {
+ * // Name.
+ * private String name;
+ *
+ * // Salary.
+ * private double salary;
+ *
+ * ...
+ * }
+ * </pre>
+ * <p>
+ * Here is the {@code ExampleCacheEntryFilter} class:
+ * <pre name="code" class="java">
+ * @IgniteAsyncCallback
+ * public class ExampleCacheEntryFilter implements CacheEntryEventFilter<Integer, Person> {
+ * @IgniteInstanceResource
+ * private Ignite ignite;
+ *
+ * // Continuous listener will be notified for persons with salary above 1000.
+ * // Filter increases salary for some person on 100. Without @IgniteAsyncCallback annotation
+ * // this operation is not safe.
+ * public boolean evaluate(CacheEntryEvent<? extends K, ? extends V> evt) throws CacheEntryListenerException {
+ * Person p = evt.getValue();
+ *
+ * if (p.getSalary() > 1000)
+ * return true;
+ *
+ * ignite.cache("Person").put(evt.getKey(), new Person(p.getName(), p.getSalary() + 100));
+ *
+ * return false;
+ * }
+ * }
+ * </pre>
+ * <p>
+ * Query with asynchronous callback execute as usually:
+ * <pre name="code" class="java">
+ * // Create new continuous query.
+ * ContinuousQuery<Long, Person> qry = new ContinuousQuery<>();
+ *
+ * // Callback that is called locally when update notifications are received.
+ * // It simply prints out information about all created persons.
+ * qry.setLocalListener((evts) -> {
+ * for (CacheEntryEvent<? extends Long, ? extends Person> e : evts) {
+ * Person p = e.getValue();
+ *
+ * System.out.println(p.getFirstName() + " " + p.getLastName() + "'s salary is " + p.getSalary());
+ * }
+ * });
+ *
+ * // Sets remote filter.
+ * qry.setRemoteFilterFactory(() -> new ExampleCacheEntryFilter());
+ *
+ * // Execute query.
+ * QueryCursor<Cache.Entry<Long, Person>> cur = cache.query(qry);
+ * </pre>
+ *
+ * @see IgniteConfiguration#getAsyncCallbackPoolSize
+ * @see ContinuousQuery#getRemoteFilterFactory()
+ * @see ContinuousQuery#getLocalListener()
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface IgniteAsyncCallback {
+ // No-op.
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/main/java/org/apache/ignite/thread/IgniteStripedThreadPoolExecutor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/thread/IgniteStripedThreadPoolExecutor.java b/modules/core/src/main/java/org/apache/ignite/thread/IgniteStripedThreadPoolExecutor.java
index 35882b9..9f7c381 100644
--- a/modules/core/src/main/java/org/apache/ignite/thread/IgniteStripedThreadPoolExecutor.java
+++ b/modules/core/src/main/java/org/apache/ignite/thread/IgniteStripedThreadPoolExecutor.java
@@ -17,62 +17,63 @@
package org.apache.ignite.thread;
+import java.util.ArrayList;
import java.util.Collection;
-import java.util.LinkedList;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.NotNull;
/**
* An {@link ExecutorService} that executes submitted tasks using pooled grid threads.
*/
public class IgniteStripedThreadPoolExecutor implements ExecutorService {
/** */
- public static final int DFLT_SEG_POOL_SIZE = 8;
-
- /** */
- public static final int DFLT_CONCUR_LVL = 16;
-
- /** */
private final ExecutorService[] execs;
- /** */
- private final int segShift;
-
- /** */
- private final int segMask;
-
/**
+ * Create striped thread pool.
*
+ * @param concurrentLvl Concurrency level.
+ * @param gridName Node name.
+ * @param threadNamePrefix Thread name prefix.
*/
- public IgniteStripedThreadPoolExecutor() {
- execs = new ExecutorService[DFLT_CONCUR_LVL];
-
- ThreadFactory factory = new IgniteThreadFactory(null);
-
- for (int i = 0; i < DFLT_CONCUR_LVL; i++)
- execs[i] = Executors.newFixedThreadPool(DFLT_SEG_POOL_SIZE, factory);
+ public IgniteStripedThreadPoolExecutor(int concurrentLvl, String gridName, String threadNamePrefix) {
+ execs = new ExecutorService[concurrentLvl];
- // Find power-of-two sizes best matching arguments
- int sshift = 0;
- int ssize = 1;
+ ThreadFactory factory = new IgniteThreadFactory(gridName, threadNamePrefix);
- while (ssize < DFLT_CONCUR_LVL) {
- ++sshift;
-
- ssize <<= 1;
- }
+ for (int i = 0; i < concurrentLvl; i++)
+ execs[i] = Executors.newSingleThreadExecutor(factory);
+ }
- segShift = 32 - sshift;
- segMask = ssize - 1;
+ /**
+ * Executes the given command at some time in the future. The command with the same {@code index}
+ * will be executed in the same thread.
+ *
+ * @param task the runnable task
+ * @param idx Striped index.
+ * @throws RejectedExecutionException if this task cannot be
+ * accepted for execution.
+ * @throws NullPointerException If command is null
+ */
+ public void execute(Runnable task, int idx) {
+ execs[threadId(idx)].execute(task);
+ }
+ /**
+ * @param idx Index.
+ * @return Stripped thread ID.
+ */
+ public int threadId(int idx) {
+ return idx < execs.length ? idx : idx % execs.length;
}
/** {@inheritDoc} */
@@ -83,7 +84,10 @@ public class IgniteStripedThreadPoolExecutor implements ExecutorService {
/** {@inheritDoc} */
@Override public List<Runnable> shutdownNow() {
- List<Runnable> res = new LinkedList<>();
+ if (execs.length == 0)
+ return Collections.emptyList();
+
+ List<Runnable> res = new ArrayList<>(execs.length);
for (ExecutorService exec : execs) {
for (Runnable r : exec.shutdownNow())
@@ -124,105 +128,45 @@ public class IgniteStripedThreadPoolExecutor implements ExecutorService {
}
/** {@inheritDoc} */
- @Override public <T> Future<T> submit(Callable<T> task) {
- return execForTask(task).submit(task);
+ @NotNull @Override public <T> Future<T> submit(Callable<T> task) {
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
- @Override public <T> Future<T> submit(Runnable task, T result) {
- return execForTask(task).submit(task, result);
+ @NotNull @Override public <T> Future<T> submit(Runnable task, T res) {
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
- @Override public Future<?> submit(Runnable task) {
- return execForTask(task).submit(task);
+ @NotNull @Override public Future<?> submit(Runnable task) {
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
- @Override public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
- throws InterruptedException {
- List<Future<T>> futs = new LinkedList<>();
-
- for (Callable<T> task : tasks)
- futs.add(execForTask(task).submit(task));
-
- boolean done = false;
-
- try {
- for (Future<T> fut : futs) {
- try {
- fut.get();
- }
- catch (ExecutionException | InterruptedException ignored) {
- // No-op.
- }
- }
-
- done = true;
-
- return futs;
- }
- finally {
- if (!done) {
- for (Future<T> fut : futs)
- fut.cancel(true);
- }
- }
+ @NotNull @Override public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
- @Override public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout,
- TimeUnit unit) throws InterruptedException {
- throw new RuntimeException("Not implemented.");
+ @NotNull @Override public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
+ long timeout,
+ TimeUnit unit) {
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
- @Override public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException,
- ExecutionException {
- throw new RuntimeException("Not implemented.");
+ @NotNull @Override public <T> T invokeAny(Collection<? extends Callable<T>> tasks) {
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
- @Override public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
- throws InterruptedException, ExecutionException, TimeoutException {
- throw new RuntimeException("Not implemented.");
+ @Override public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) {
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
@Override public void execute(Runnable cmd) {
- execForTask(cmd).execute(cmd);
- }
-
- /**
- * Applies a supplemental hash function to a given hashCode, which
- * defends against poor quality hash functions. This is critical
- * because ConcurrentHashMap uses power-of-two length hash tables,
- * that otherwise encounter collisions for hashCodes that do not
- * differ in lower or upper bits.
- *
- * @param h Hash code.
- * @return Enhanced hash code.
- */
- private int hash(int h) {
- // Spread bits to regularize both segment and index locations,
- // using variant of single-word Wang/Jenkins hash.
- h += (h << 15) ^ 0xffffcd7d;
- h ^= (h >>> 10);
- h += (h << 3);
- h ^= (h >>> 6);
- h += (h << 2) + (h << 14);
- return h ^ (h >>> 16);
- }
-
- /**
- * @param cmd Command.
- * @return Service.
- */
- private <T> ExecutorService execForTask(T cmd) {
- assert cmd != null;
-
- //return execs[ThreadLocalRandom8.current().nextInt(DFLT_CONCUR_LVL)];
- return execs[(hash(cmd.hashCode()) >>> segShift) & segMask];
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
index 6e404b4..a1153cd 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
@@ -26,6 +26,7 @@ import javax.cache.processor.EntryProcessorResult;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.eviction.EvictableEntry;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateFuture;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -540,7 +541,8 @@ public class GridCacheTestEntryEx extends GridMetadataAwareAdapter implements Gr
UUID subjId,
String taskName,
@Nullable CacheObject prevVal,
- @Nullable Long updateCntr) throws IgniteCheckedException,
+ @Nullable Long updateCntr,
+ @Nullable GridDhtAtomicUpdateFuture fut) throws IgniteCheckedException,
GridCacheEntryRemovedException {
assert false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicPrimaryWriteOrderSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicPrimaryWriteOrderSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicPrimaryWriteOrderSelfTest.java
new file mode 100644
index 0000000..62fd984
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverAtomicPrimaryWriteOrderSelfTest.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import org.apache.ignite.cache.CacheAtomicWriteOrderMode;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+
+import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.PRIMARY;
+
+/**
+ *
+ */
+public class CacheContinuousQueryAsyncFailoverAtomicPrimaryWriteOrderSelfTest
+ extends CacheContinuousQueryFailoverAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicWriteOrderMode writeOrderMode() {
+ return PRIMARY;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return CacheMode.PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.ATOMIC;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected boolean asyncCallback() {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxReplicatedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxReplicatedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxReplicatedSelfTest.java
new file mode 100644
index 0000000..4460498
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxReplicatedSelfTest.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import org.apache.ignite.cache.CacheMode;
+
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+
+/**
+ *
+ */
+public class CacheContinuousQueryAsyncFailoverTxReplicatedSelfTest extends CacheContinuousQueryFailoverTxSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return REPLICATED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected boolean asyncCallback() {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java
new file mode 100644
index 0000000..8f0bd0e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFailoverTxSelfTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+
+/**
+ *
+ */
+public class CacheContinuousQueryAsyncFailoverTxSelfTest extends CacheContinuousQueryFailoverAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return TRANSACTIONAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected boolean asyncCallback() {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFilterListenerTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFilterListenerTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFilterListenerTest.java
new file mode 100644
index 0000000..0605bc8
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryAsyncFilterListenerTest.java
@@ -0,0 +1,986 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import java.io.Serializable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import javax.cache.configuration.FactoryBuilder;
+import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
+import javax.cache.event.CacheEntryCreatedListener;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryUpdatedListener;
+import javax.cache.processor.EntryProcessorException;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.CacheEntryProcessor;
+import org.apache.ignite.cache.CacheMemoryMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+import org.apache.ignite.lang.IgniteBiInClosure;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.spi.eventstorage.memory.MemoryEventStorageSpi;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.PRIMARY;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_TIERED;
+import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_VALUES;
+import static org.apache.ignite.cache.CacheMemoryMode.ONHEAP_TIERED;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ *
+ */
+public class CacheContinuousQueryAsyncFilterListenerTest extends GridCommonAbstractTest {
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int NODES = 5;
+
+ /** */
+ public static final int ITERATION_CNT = 100;
+
+ /** */
+ private boolean client;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setClientMode(client);
+
+ MemoryEventStorageSpi storeSpi = new MemoryEventStorageSpi();
+ storeSpi.setExpireCount(1000);
+
+ cfg.setEventStorageSpi(storeSpi);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGridsMultiThreaded(NODES - 1);
+
+ client = true;
+
+ startGrid(NODES - 1);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ super.afterTestsStopped();
+ }
+
+ ///
+ /// ASYNC FILTER AND LISTENER. TEST LISTENER.
+ ///
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerTx() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerTxJCacheApi() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerTxOffHeap() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, OFFHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerTxOffHeapJCacheApi() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, OFFHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerTxOffHeapValues() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, OFFHEAP_VALUES), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerAtomic() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, ATOMIC, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerAtomicJCacheApi() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, ATOMIC, ONHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerReplicatedAtomic() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(REPLICATED, 2, ATOMIC, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerReplicatedAtomicJCacheApi() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(REPLICATED, 2, ATOMIC, ONHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerReplicatedAtomicOffHeapValues() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(REPLICATED, 2, ATOMIC, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerAtomicOffHeap() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, ATOMIC, OFFHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerAtomicOffHeapValues() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, ATOMIC, OFFHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerAtomicWithoutBackup() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 0, ATOMIC, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerAtomicWithoutBackupJCacheApi() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 0, ATOMIC, ONHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListener() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerReplicated() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(REPLICATED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInListenerReplicatedJCacheApi() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(REPLICATED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, true);
+ }
+
+ ///
+ /// ASYNC FILTER AND LISTENER. TEST FILTER.
+ ///
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterTx() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterTxJCacheApi() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterTxOffHeap() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, OFFHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterTxOffHeapJCacheApi() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, OFFHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterTxOffHeapValues() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, OFFHEAP_VALUES), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomic() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, ATOMIC, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicJCacheApi() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, ATOMIC, ONHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterReplicatedAtomic() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(REPLICATED, 2, ATOMIC, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicOffHeap() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, ATOMIC, OFFHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicOffHeapJCacheApi() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, ATOMIC, OFFHEAP_TIERED), true, true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicOffHeapValues() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, ATOMIC, OFFHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicWithoutBackup() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 0, ATOMIC, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilter() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterReplicated() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(REPLICATED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterReplicatedJCacheApi() throws Exception {
+ testNonDeadLockInFilter(cacheConfiguration(REPLICATED, 2, TRANSACTIONAL, ONHEAP_TIERED), true, true, false);
+ }
+
+ ///
+ /// ASYNC LISTENER. TEST LISTENER.
+ ///
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterTxSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, ONHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterTxOffHeapSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, OFFHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterTxOffHeapValuesSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, OFFHEAP_VALUES), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, ATOMIC, ONHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterReplicatedAtomicSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(REPLICATED, 2, ATOMIC, ONHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicOffHeapSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, ATOMIC, OFFHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicOffHeapValuesSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, ATOMIC, OFFHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterAtomicWithoutBackupSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 0, ATOMIC, ONHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL, ONHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNonDeadLockInFilterReplicatedSyncFilter() throws Exception {
+ testNonDeadLockInListener(cacheConfiguration(REPLICATED, 2, TRANSACTIONAL, ONHEAP_TIERED), false, true, false);
+ }
+
+ /**
+ * @param ccfg Cache configuration.
+ * @param asyncFltr Async filter.
+ * @param asyncLsnr Async listener.
+ * @param jcacheApi Use JCache api for registration entry update listener.
+ * @throws Exception If failed.
+ */
+ private void testNonDeadLockInListener(CacheConfiguration ccfg,
+ final boolean asyncFltr,
+ boolean asyncLsnr,
+ boolean jcacheApi) throws Exception {
+ ignite(0).createCache(ccfg);
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ try {
+ for (int i = 0; i < ITERATION_CNT; i++) {
+ log.info("Start iteration: " + i);
+
+ int nodeIdx = i % NODES;
+
+ final IgniteCache cache = grid(nodeIdx).cache(ccfg.getName());
+
+ final QueryTestKey key = NODES - 1 != nodeIdx ? affinityKey(cache) : new QueryTestKey(1);
+
+ final QueryTestValue val0 = new QueryTestValue(1);
+ final QueryTestValue newVal = new QueryTestValue(2);
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final CountDownLatch evtFromLsnrLatch = new CountDownLatch(1);
+
+ IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> fltrClsr =
+ new IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>>() {
+ @Override public void apply(Ignite ignite, CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue> e) {
+ if (asyncFltr) {
+ assertFalse("Failed: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("sys-"));
+
+ assertTrue("Failed: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("callback-"));
+ }
+ }
+ };
+
+ IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> lsnrClsr =
+ new IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>>() {
+ @Override public void apply(Ignite ignite, CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue> e) {
+ IgniteCache<Object, Object> cache0 = ignite.cache(cache.getName());
+
+ QueryTestValue val = e.getValue();
+
+ if (val == null)
+ return;
+ else if (val.equals(newVal)) {
+ evtFromLsnrLatch.countDown();
+
+ return;
+ }
+ else if (!val.equals(val0))
+ return;
+
+ Transaction tx = null;
+
+ try {
+ if (cache0.getConfiguration(CacheConfiguration.class).getAtomicityMode() == TRANSACTIONAL)
+ tx = ignite.transactions().txStart(PESSIMISTIC, REPEATABLE_READ);
+
+ assertEquals(val, val0);
+
+ cache0.put(key, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ latch.countDown();
+ }
+ catch (Exception exp) {
+ log.error("Failed: ", exp);
+
+ throw new IgniteException(exp);
+ }
+ finally {
+ if (tx != null)
+ tx.close();
+ }
+ }
+ };
+
+ QueryCursor qry = null;
+ MutableCacheEntryListenerConfiguration<QueryTestKey, QueryTestValue> lsnrCfg = null;
+
+ CacheInvokeListener locLsnr = asyncLsnr ? new CacheInvokeListenerAsync(lsnrClsr)
+ : new CacheInvokeListener(lsnrClsr);
+
+ CacheEntryEventSerializableFilter<QueryTestKey, QueryTestValue> rmtFltr = asyncFltr ?
+ new CacheTestRemoteFilterAsync(fltrClsr) : new CacheTestRemoteFilter(fltrClsr);
+
+ if (jcacheApi) {
+ lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
+ FactoryBuilder.factoryOf(locLsnr),
+ FactoryBuilder.factoryOf(rmtFltr),
+ true,
+ false
+ );
+
+ cache.registerCacheEntryListener(lsnrCfg);
+ }
+ else {
+ ContinuousQuery<QueryTestKey, QueryTestValue> conQry = new ContinuousQuery<>();
+
+ conQry.setLocalListener(locLsnr);
+
+ conQry.setRemoteFilterFactory(FactoryBuilder.factoryOf(rmtFltr));
+
+ qry = cache.query(conQry);
+ }
+
+ try {
+ if (rnd.nextBoolean())
+ cache.put(key, val0);
+ else {
+ cache.invoke(key, new CacheEntryProcessor() {
+ @Override public Object process(MutableEntry entry, Object... arguments)
+ throws EntryProcessorException {
+ entry.setValue(val0);
+
+ return null;
+ }
+ });
+ }
+
+ assertTrue("Failed to waiting event.", U.await(latch, 3, SECONDS));
+
+ assertEquals(cache.get(key), new QueryTestValue(2));
+
+ assertTrue("Failed to waiting event from listener.", U.await(latch, 3, SECONDS));
+ }
+ finally {
+ if (qry != null)
+ qry.close();
+
+ if (lsnrCfg != null)
+ cache.deregisterCacheEntryListener(lsnrCfg);
+ }
+
+ log.info("Iteration finished: " + i);
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @param ccfg Cache configuration.
+ * @param asyncFilter Async filter.
+ * @param asyncLsnr Async listener.
+ * @param jcacheApi Use JCache api for start update listener.
+ * @throws Exception If failed.
+ */
+ private void testNonDeadLockInFilter(CacheConfiguration ccfg,
+ final boolean asyncFilter,
+ final boolean asyncLsnr,
+ boolean jcacheApi) throws Exception {
+ ignite(0).createCache(ccfg);
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ try {
+ for (int i = 0; i < ITERATION_CNT; i++) {
+ log.info("Start iteration: " + i);
+
+ int nodeIdx = i % NODES;
+
+ final IgniteCache cache = grid(nodeIdx).cache(ccfg.getName());
+
+ final QueryTestKey key = NODES - 1 != nodeIdx ? affinityKey(cache) : new QueryTestKey(1);
+
+ final QueryTestValue val0 = new QueryTestValue(1);
+ final QueryTestValue newVal = new QueryTestValue(2);
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final CountDownLatch evtFromLsnrLatch = new CountDownLatch(1);
+
+ IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> fltrClsr =
+ new IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>>() {
+ @Override public void apply(Ignite ignite, CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue> e) {
+ if (asyncFilter) {
+ assertFalse("Failed: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("sys-"));
+
+ assertTrue("Failed: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("callback-"));
+ }
+
+ IgniteCache<Object, Object> cache0 = ignite.cache(cache.getName());
+
+ QueryTestValue val = e.getValue();
+
+ if (val == null)
+ return;
+ else if (val.equals(newVal)) {
+ evtFromLsnrLatch.countDown();
+
+ return;
+ }
+ else if (!val.equals(val0))
+ return;
+
+ Transaction tx = null;
+
+ try {
+ if (cache0.getConfiguration(CacheConfiguration.class)
+ .getAtomicityMode() == TRANSACTIONAL)
+ tx = ignite.transactions().txStart(PESSIMISTIC, REPEATABLE_READ);
+
+ assertEquals(val, val0);
+
+ cache0.put(key, newVal);
+
+ if (tx != null)
+ tx.commit();
+
+ latch.countDown();
+ }
+ catch (Exception exp) {
+ log.error("Failed: ", exp);
+
+ throw new IgniteException(exp);
+ }
+ finally {
+ if (tx != null)
+ tx.close();
+ }
+ }
+ };
+
+ IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> lsnrClsr =
+ new IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>>() {
+ @Override public void apply(Ignite ignite, CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue> e) {
+ if (asyncLsnr) {
+ assertFalse("Failed: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("sys-"));
+
+ assertTrue("Failed: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("callback-"));
+ }
+
+ QueryTestValue val = e.getValue();
+
+ if (val == null || !val.equals(new QueryTestValue(1)))
+ return;
+
+ assertEquals(val, val0);
+
+ latch.countDown();
+ }
+ };
+
+
+ QueryCursor qry = null;
+ MutableCacheEntryListenerConfiguration<QueryTestKey, QueryTestValue> lsnrCfg = null;
+
+ CacheInvokeListener locLsnr = asyncLsnr ? new CacheInvokeListenerAsync(lsnrClsr)
+ : new CacheInvokeListener(lsnrClsr);
+
+ CacheEntryEventSerializableFilter<QueryTestKey, QueryTestValue> rmtFltr = asyncFilter ?
+ new CacheTestRemoteFilterAsync(fltrClsr) : new CacheTestRemoteFilter(fltrClsr);
+
+ if (jcacheApi) {
+ lsnrCfg = new MutableCacheEntryListenerConfiguration<>(
+ FactoryBuilder.factoryOf(locLsnr),
+ FactoryBuilder.factoryOf(rmtFltr),
+ true,
+ false
+ );
+
+ cache.registerCacheEntryListener(lsnrCfg);
+ }
+ else {
+ ContinuousQuery<QueryTestKey, QueryTestValue> conQry = new ContinuousQuery<>();
+
+ conQry.setLocalListener(locLsnr);
+
+ conQry.setRemoteFilterFactory(FactoryBuilder.factoryOf(rmtFltr));
+
+ qry = cache.query(conQry);
+ }
+
+ try {
+ if (rnd.nextBoolean())
+ cache.put(key, val0);
+ else
+ cache.invoke(key, new CacheEntryProcessor() {
+ @Override public Object process(MutableEntry entry, Object... arguments)
+ throws EntryProcessorException {
+ entry.setValue(val0);
+
+ return null;
+ }
+ });
+
+ assert U.await(latch, 3, SECONDS) : "Failed to waiting event.";
+
+ assertEquals(cache.get(key), new QueryTestValue(2));
+
+ assertTrue("Failed to waiting event from filter.", U.await(latch, 3, SECONDS));
+ }
+ finally {
+ if (qry != null)
+ qry.close();
+
+ if (lsnrCfg != null)
+ cache.deregisterCacheEntryListener(lsnrCfg);
+ }
+
+ log.info("Iteration finished: " + i);
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @param cache Ignite cache.
+ * @return Key.
+ */
+ private QueryTestKey affinityKey(IgniteCache cache) {
+ Affinity aff = affinity(cache);
+
+ for (int i = 0; i < 10_000; i++) {
+ QueryTestKey key = new QueryTestKey(i);
+
+ if (aff.isPrimary(localNode(cache), key))
+ return key;
+ }
+
+ throw new IgniteException("Failed to found primary key.");
+ }
+
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return TimeUnit.SECONDS.toMillis(15);
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ private static class CacheTestRemoteFilterAsync extends CacheTestRemoteFilter {
+ /**
+ * @param clsr Closure.
+ */
+ public CacheTestRemoteFilterAsync(
+ IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> clsr) {
+ super(clsr);
+ }
+ }
+
+ /**
+ *
+ */
+ private static class CacheTestRemoteFilter implements
+ CacheEntryEventSerializableFilter<QueryTestKey, QueryTestValue> {
+ /** */
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** */
+ private IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> clsr;
+
+ /**
+ * @param clsr Closure.
+ */
+ public CacheTestRemoteFilter(IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> clsr) {
+ this.clsr = clsr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e)
+ throws CacheEntryListenerException {
+ clsr.apply(ignite, e);
+
+ return true;
+ }
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ private static class CacheInvokeListenerAsync extends CacheInvokeListener {
+ /**
+ * @param clsr Closure.
+ */
+ public CacheInvokeListenerAsync(
+ IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> clsr) {
+ super(clsr);
+ }
+ }
+
+ /**
+ *
+ */
+ private static class CacheInvokeListener implements CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>,
+ CacheEntryCreatedListener<QueryTestKey, QueryTestValue>, Serializable {
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** */
+ private IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> clsr;
+
+ /**
+ * @param clsr Closure.
+ */
+ public CacheInvokeListener(IgniteBiInClosure<Ignite, CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> clsr) {
+ this.clsr = clsr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> events)
+ throws CacheEntryListenerException {
+ for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : events)
+ clsr.apply(ignite, e);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onCreated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> events) throws CacheEntryListenerException {
+ for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : events)
+ clsr.apply(ignite, e);
+ }
+ }
+
+ /**
+ * @param cacheMode Cache mode.
+ * @param backups Number of backups.
+ * @param atomicityMode Cache atomicity mode.
+ * @param memoryMode Cache memory mode.
+ * @return Cache configuration.
+ */
+ protected CacheConfiguration<Object, Object> cacheConfiguration(
+ CacheMode cacheMode,
+ int backups,
+ CacheAtomicityMode atomicityMode,
+ CacheMemoryMode memoryMode) {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
+
+ ccfg.setName("test-cache-" + atomicityMode + "-" + cacheMode + "-" + memoryMode + "-" + backups);
+ ccfg.setAtomicityMode(atomicityMode);
+ ccfg.setCacheMode(cacheMode);
+ ccfg.setMemoryMode(memoryMode);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setAtomicWriteOrderMode(PRIMARY);
+
+ if (cacheMode == PARTITIONED)
+ ccfg.setBackups(backups);
+
+ return ccfg;
+ }
+
+ /**
+ *
+ */
+ public static class QueryTestKey implements Serializable, Comparable {
+ /** */
+ private final Integer key;
+
+ /**
+ * @param key Key.
+ */
+ public QueryTestKey(Integer key) {
+ this.key = key;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ QueryTestKey that = (QueryTestKey)o;
+
+ return key.equals(that.key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return key.hashCode();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(QueryTestKey.class, this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int compareTo(Object o) {
+ return key - ((QueryTestKey)o).key;
+ }
+ }
+
+ /**
+ *
+ */
+ public static class QueryTestValue implements Serializable {
+ /** */
+ @GridToStringInclude
+ protected final Integer val1;
+
+ /** */
+ @GridToStringInclude
+ protected final String val2;
+
+ /**
+ * @param val Value.
+ */
+ public QueryTestValue(Integer val) {
+ this.val1 = val;
+ this.val2 = String.valueOf(val);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ QueryTestValue that = (QueryTestValue)o;
+
+ return val1.equals(that.val1) && val2.equals(that.val2);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = val1.hashCode();
+
+ res = 31 * res + val2.hashCode();
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(QueryTestValue.class, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2ff64c2a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryAsyncFilterRandomOperationTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryAsyncFilterRandomOperationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryAsyncFilterRandomOperationTest.java
new file mode 100644
index 0000000..928cfda
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFactoryAsyncFilterRandomOperationTest.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.query.continuous;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import javax.cache.configuration.Factory;
+import javax.cache.configuration.FactoryBuilder;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.lang.IgniteAsyncCallback;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ *
+ */
+public class CacheContinuousQueryFactoryAsyncFilterRandomOperationTest
+ extends CacheContinuousQueryFactoryFilterRandomOperationTest {
+ /** {@inheritDoc} */
+ @NotNull @Override protected Factory<? extends CacheEntryEventFilter<QueryTestKey, QueryTestValue>>
+ createFilterFactory() {
+ return new AsyncFilterFactory();
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ protected static class NonSerializableAsyncFilter implements
+ CacheEntryEventSerializableFilter<QueryTestKey, QueryTestValue>, Externalizable {
+ /** */
+ public NonSerializableAsyncFilter() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> evt) {
+ assertTrue("Failed. Current thread name: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("callback-"));
+
+ assertFalse("Failed. Current thread name: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("sys-"));
+
+ return isAccepted(evt.getValue());
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ fail("Entry filter should not be marshaled.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ fail("Entry filter should not be marshaled.");
+ }
+
+ /**
+ * @param val Value.
+ * @return {@code True} if value is even.
+ */
+ public static boolean isAccepted(QueryTestValue val) {
+ return val == null || val.val1 % 2 == 0;
+ }
+ }
+
+ /**
+ *
+ */
+ protected static class AsyncFilterFactory implements Factory<NonSerializableAsyncFilter> {
+ /** {@inheritDoc} */
+ @Override public NonSerializableAsyncFilter create() {
+ return new NonSerializableAsyncFilter();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Factory<? extends CacheEntryEventFilter<QueryTestKey, QueryTestValue>> noOpFilterFactory() {
+ return FactoryBuilder.factoryOf(NoopAsyncFilter.class);
+ }
+
+ /**
+ *
+ */
+ @IgniteAsyncCallback
+ protected static class NoopAsyncFilter implements
+ CacheEntryEventSerializableFilter<QueryTestKey, QueryTestValue>, Externalizable {
+ /** */
+ public NoopAsyncFilter() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean evaluate(CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> evt) {
+ assertTrue("Failed. Current thread name: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("callback-"));
+
+ assertFalse("Failed. Current thread name: " + Thread.currentThread().getName(),
+ Thread.currentThread().getName().contains("sys-"));
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ // No-op.
+ }
+ }
+}