You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by av...@apache.org on 2016/03/25 09:45:29 UTC
[1/2] ignite git commit: IGNITE-GG-11048 Fix missed event
notification in remove-remove case.
Repository: ignite
Updated Branches:
refs/heads/ignite-1071 872e00230 -> aa669cd30
IGNITE-GG-11048 Fix missed event notification in remove-remove case.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/4cb994a7
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/4cb994a7
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/4cb994a7
Branch: refs/heads/ignite-1071
Commit: 4cb994a7253c6b493b4a376ed1489513ecefd8f9
Parents: aa8dcd1
Author: nikolay_tikhonov <nt...@gridgain.com>
Authored: Fri Mar 25 11:20:39 2016 +0300
Committer: nikolay_tikhonov <nt...@gridgain.com>
Committed: Fri Mar 25 11:26:49 2016 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheMapEntry.java | 2 +-
.../dht/atomic/GridDhtAtomicCache.java | 4 +-
...acheContinuousQueryRandomOperationsTest.java | 197 ++++++++++++++++++-
...inuousQueryRandomOperationsTwoNodesTest.java | 28 +++
.../IgniteCacheQuerySelfTestSuite.java | 2 +
5 files changed, 224 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/4cb994a7/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 b58996a..9678fe0 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
@@ -2086,7 +2086,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
null,
null,
false,
- updateCntr0 == null ? 0 : updateCntr0);
+ 0);
}
}
else
http://git-wip-us.apache.org/repos/asf/ignite/blob/4cb994a7/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 25090bd..7a8b5cb 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
@@ -2118,7 +2118,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
"[entry=" + entry + ", filter=" + Arrays.toString(req.filter()) + ']');
}
}
- else if (lsnrs != null && updRes.success()) {
+ else if (lsnrs != null && updRes.updateCounter() != 0) {
ctx.continuousQueries().onEntryUpdated(
lsnrs,
entry.key(),
@@ -2871,7 +2871,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (updRes.removeVersion() != null)
ctx.onDeferredDelete(entry, updRes.removeVersion());
- if (lsnrs != null && updRes.success()) {
+ if (lsnrs != null && updRes.updateCounter() != 0) {
ctx.continuousQueries().onEntryUpdated(
lsnrs,
entry.key(),
http://git-wip-us.apache.org/repos/asf/ignite/blob/4cb994a7/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 cdf4ffd..f6ad411 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
@@ -31,12 +31,14 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
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 javax.cache.event.EventType;
import javax.cache.integration.CacheLoaderException;
import javax.cache.integration.CacheWriterException;
import javax.cache.processor.EntryProcessor;
@@ -56,6 +58,8 @@ import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cache.store.CacheStoreAdapter;
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.PA;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
@@ -121,11 +125,11 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
@Override protected void beforeTestsStarted() throws Exception {
super.beforeTestsStarted();
- startGridsMultiThreaded(NODES - 1);
+ startGridsMultiThreaded(getServerNodeCount());
client = true;
- startGrid(NODES - 1);
+ startGrid(getServerNodeCount());
}
/** {@inheritDoc} */
@@ -417,6 +421,171 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
/**
* @throws Exception If failed.
*/
+ public void testDoubleRemoveAtomicWithoutBackup() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED,
+ 0,
+ ATOMIC,
+ ONHEAP_TIERED,
+ false);
+
+ testDoubleRemove(ccfg);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDoubleRemoveAtomic() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED,
+ 1,
+ ATOMIC,
+ ONHEAP_TIERED,
+ false);
+
+ testDoubleRemove(ccfg);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDoubleRemoveAtomicOffheap() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED,
+ 0,
+ ATOMIC,
+ OFFHEAP_TIERED,
+ false);
+
+ testDoubleRemove(ccfg);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDoubleRemoveTx() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED,
+ 1,
+ TRANSACTIONAL,
+ ONHEAP_TIERED,
+ false);
+
+ testDoubleRemove(ccfg);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDoubleRemoveReplicatedTx() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED,
+ 0,
+ TRANSACTIONAL,
+ ONHEAP_TIERED,
+ false);
+
+ testDoubleRemove(ccfg);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDoubleRemoveReplicatedAtomic() throws Exception {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED,
+ 0,
+ ATOMIC,
+ ONHEAP_TIERED,
+ false);
+
+ testDoubleRemove(ccfg);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDoubleRemove(CacheConfiguration ccfg) throws Exception {
+ IgniteCache<QueryTestKey, QueryTestValue> cache = grid(getClientIndex()).createCache(ccfg);
+
+ try {
+ ContinuousQuery<QueryTestKey, QueryTestValue> qry = new ContinuousQuery<>();
+
+ final List<CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue>> evts =
+ new CopyOnWriteArrayList<>();
+
+ qry.setLocalListener(new CacheEntryUpdatedListener<QueryTestKey, QueryTestValue>() {
+ @Override public void onUpdated(Iterable<CacheEntryEvent<? extends QueryTestKey,
+ ? extends QueryTestValue>> events) throws CacheEntryListenerException {
+ for (CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> e : events)
+ evts.add(e);
+ }
+ });
+
+ QueryTestKey key = new QueryTestKey(1);
+
+ try (QueryCursor qryCur = cache.query(qry)) {
+ for (int i = 0; i < ITERATION_CNT; i++) {
+ log.info("Start iteration: " + i);
+
+ cache.put(key, new QueryTestValue(1));
+
+ cache.remove(key);
+ cache.invoke(key, (EntryProcessor<QueryTestKey, QueryTestValue, ? extends Object>)
+ (Object)new EntrySetValueProcessor(null, false));
+ cache.invoke(key, (EntryProcessor<QueryTestKey, QueryTestValue, ? extends Object>)
+ (Object)new EntrySetValueProcessor(null, false));
+ cache.remove(key);
+
+ cache.put(key, new QueryTestValue(2));
+
+ cache.invoke(key, (EntryProcessor<QueryTestKey, QueryTestValue, ? extends Object>)
+ (Object)new EntrySetValueProcessor(null, false));
+ cache.remove(key);
+
+ cache.put(key, new QueryTestValue(3));
+ cache.put(key, new QueryTestValue(4));
+
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return evts.size() == 6;
+ }
+ }, 5_000);
+
+ checkSingleEvent(evts.get(0), EventType.CREATED, new QueryTestValue(1), null);
+ checkSingleEvent(evts.get(1), EventType.REMOVED, null, new QueryTestValue(1));
+ checkSingleEvent(evts.get(2), EventType.CREATED, new QueryTestValue(2), null);
+ checkSingleEvent(evts.get(3), EventType.REMOVED, null, new QueryTestValue(2));
+ checkSingleEvent(evts.get(4), EventType.CREATED, new QueryTestValue(3), null);
+ checkSingleEvent(evts.get(5), EventType.UPDATED, new QueryTestValue(4), new QueryTestValue(3));
+
+ cache.remove(key);
+ cache.remove(key);
+
+ evts.clear();
+
+ log.info("Finish iteration: " + i);
+ }
+ }
+ }
+ finally {
+ grid(getClientIndex()).destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @param event Event.
+ * @param type Event type.
+ * @param val Value.
+ * @param oldVal Old value.
+ */
+ private void checkSingleEvent(
+ CacheEntryEvent<? extends QueryTestKey, ? extends QueryTestValue> event,
+ EventType type,
+ QueryTestValue val,
+ QueryTestValue oldVal) {
+ assertEquals(event.getEventType(), type);
+ assertEquals(event.getValue(), val);
+ assertEquals(event.getOldValue(), oldVal);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testTxClient() throws Exception {
CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED,
1,
@@ -656,7 +825,7 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
evtsQueues.add(evtsQueue);
- QueryCursor<?> cur = grid(NODES - 1).cache(ccfg.getName()).query(qry);
+ QueryCursor<?> cur = grid(getClientIndex()).cache(ccfg.getName()).query(qry);
curs.add(cur);
}
@@ -674,12 +843,12 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
evtsQueues.add(evtsQueue);
- QueryCursor<?> cur = grid(rnd.nextInt(NODES - 1)).cache(ccfg.getName()).query(qry);
+ QueryCursor<?> cur = grid(rnd.nextInt(getServerNodeCount())).cache(ccfg.getName()).query(qry);
curs.add(cur);
}
else {
- for (int i = 0; i < NODES - 1; i++) {
+ for (int i = 0; i <= getServerNodeCount(); i++) {
ContinuousQuery<Object, Object> qry = new ContinuousQuery<>();
final BlockingQueue<CacheEntryEvent<?, ?>> evtsQueue = new ArrayBlockingQueue<>(50_000);
@@ -708,7 +877,7 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
if (i % 20 == 0)
log.info("Iteration: " + i);
- for (int idx = 0; idx < NODES; idx++)
+ for (int idx = 0; idx < getServerNodeCount(); idx++)
randomUpdate(rnd, evtsQueues, expData, partCntr, grid(idx).cache(ccfg.getName()));
}
}
@@ -723,6 +892,20 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
}
/**
+ * @return Client node index.
+ */
+ private int getClientIndex() {
+ return getServerNodeCount() - 1;
+ }
+
+ /**
+ * @return Count nodes.
+ */
+ protected int getServerNodeCount() {
+ return NODES;
+ }
+
+ /**
* @param rnd Random generator.
* @param evtsQueues Events queue.
* @param expData Expected cache data.
@@ -1272,9 +1455,11 @@ public class CacheContinuousQueryRandomOperationsTest extends GridCommonAbstract
*/
public static class QueryTestValue implements Serializable {
/** */
+ @GridToStringInclude
protected final Integer val1;
/** */
+ @GridToStringInclude
protected final String val2;
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/4cb994a7/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTwoNodesTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTwoNodesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTwoNodesTest.java
new file mode 100644
index 0000000..ebb7861
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryRandomOperationsTwoNodesTest.java
@@ -0,0 +1,28 @@
+/*
+ * 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;
+
+/**
+ *
+ */
+public class CacheContinuousQueryRandomOperationsTwoNodesTest extends CacheContinuousQueryRandomOperationsTest {
+ /** {@inheritDoc} */
+ @Override protected int getServerNodeCount() {
+ return 1;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4cb994a7/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java
index 467639f..38d7456 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite.java
@@ -85,6 +85,7 @@ import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinu
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFailoverTxSelfTest;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryOperationP2PTest;
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;
import org.apache.ignite.internal.processors.cache.query.continuous.GridCacheContinuousQueryAtomicOffheapTieredTest;
import org.apache.ignite.internal.processors.cache.query.continuous.GridCacheContinuousQueryAtomicOffheapValuesTest;
@@ -218,6 +219,7 @@ public class IgniteCacheQuerySelfTestSuite extends TestSuite {
suite.addTestSuite(CacheContinuousQueryFailoverAtomicPrimaryWriteOrderOffheapTieredTest.class);
suite.addTestSuite(CacheContinuousQueryFailoverTxOffheapTieredTest.class);
suite.addTestSuite(CacheContinuousQueryRandomOperationsTest.class);
+ suite.addTestSuite(CacheContinuousQueryRandomOperationsTwoNodesTest.class);
suite.addTestSuite(CacheContinuousQueryFactoryFilterTest.class);
suite.addTestSuite(GridCacheContinuousQueryConcurrentTest.class);
suite.addTestSuite(CacheContinuousQueryOperationP2PTest.class);
[2/2] ignite git commit: Merge remote-tracking branch
'remotes/upstream/gridgain-7.5.10' into ignite-1071
Posted by av...@apache.org.
Merge remote-tracking branch 'remotes/upstream/gridgain-7.5.10' into ignite-1071
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/aa669cd3
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/aa669cd3
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/aa669cd3
Branch: refs/heads/ignite-1071
Commit: aa669cd300cec2fc7d598192fd56702e0778d8d5
Parents: 872e002 4cb994a
Author: Anton Vinogradov <av...@apache.org>
Authored: Fri Mar 25 11:44:33 2016 +0300
Committer: Anton Vinogradov <av...@apache.org>
Committed: Fri Mar 25 11:44:33 2016 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheMapEntry.java | 2 +-
.../dht/atomic/GridDhtAtomicCache.java | 4 +-
...acheContinuousQueryRandomOperationsTest.java | 197 ++++++++++++++++++-
...inuousQueryRandomOperationsTwoNodesTest.java | 28 +++
.../IgniteCacheQuerySelfTestSuite.java | 2 +
5 files changed, 224 insertions(+), 9 deletions(-)
----------------------------------------------------------------------