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/03/17 11:40:20 UTC
ignite git commit: IGNITE-2791 Fixed tests.
Repository: ignite
Updated Branches:
refs/heads/ignite-2791 b794817a7 -> c938bf131
IGNITE-2791 Fixed tests.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c938bf13
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c938bf13
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c938bf13
Branch: refs/heads/ignite-2791
Commit: c938bf131dcb214e58a3d2949d5ea06fad6d1832
Parents: b794817
Author: nikolay_tikhonov <nt...@gridgain.com>
Authored: Thu Mar 17 13:40:15 2016 +0300
Committer: nikolay_tikhonov <nt...@gridgain.com>
Committed: Thu Mar 17 13:40:15 2016 +0300
----------------------------------------------------------------------
.../dht/GridClientPartitionTopology.java | 2 +-
.../dht/GridDhtPartitionTopology.java | 2 +-
.../dht/GridDhtPartitionTopologyImpl.java | 5 +---
.../GridDhtPartitionsExchangeFuture.java | 8 +++---
.../continuous/CacheContinuousQueryHandler.java | 2 +-
.../continuous/GridContinuousProcessor.java | 26 ++++++++++++++++----
...ContinuousQueryFailoverAbstractSelfTest.java | 2 +-
.../CacheContinuousQueryLostPartitionTest.java | 2 --
8 files changed, 30 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c938bf13/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index ce9ff64..ad4943e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@ -874,7 +874,7 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
- @Override public Map<Integer, Long> updateCounters(AffinityTopologyVersion topVer) {
+ @Override public Map<Integer, Long> updateCounters() {
lock.readLock().lock();
try {
http://git-wip-us.apache.org/repos/asf/ignite/blob/c938bf13/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
index fe3318c..84889f8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
@@ -221,7 +221,7 @@ public interface GridDhtPartitionTopology {
/**
* @return Partition update counters.
*/
- public Map<Integer, Long> updateCounters(@Nullable AffinityTopologyVersion topVer);
+ public Map<Integer, Long> updateCounters();
/**
* @param part Partition to own.
http://git-wip-us.apache.org/repos/asf/ignite/blob/c938bf13/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
index 48b8d25..b3786cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
@@ -1316,16 +1316,13 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
- @Override public Map<Integer, Long> updateCounters(AffinityTopologyVersion topVer0) {
+ @Override public Map<Integer, Long> updateCounters() {
lock.readLock().lock();
try {
Map<Integer, Long> res = new HashMap<>(cntrMap);
for (GridDhtLocalPartition part : locParts.values()) {
- if (topVer0 != null && !part.primary(topVer0))
- continue;
-
Long cntr0 = res.get(part.id());
Long cntr1 = part.updateCounter();
http://git-wip-us.apache.org/repos/asf/ignite/blob/c938bf13/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index a0e03eb..68a05e4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -631,7 +631,7 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
if (top.cacheId() == cacheCtx.cacheId()) {
cacheCtx.topology().update(exchId,
top.partitionMap(true),
- top.updateCounters(null));
+ top.updateCounters());
break;
}
@@ -995,7 +995,7 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
m.addLocalPartitionMap(cacheCtx.cacheId(), locMap);
- m.partitionUpdateCounters(cacheCtx.cacheId(), cacheCtx.topology().updateCounters(null));
+ m.partitionUpdateCounters(cacheCtx.cacheId(), cacheCtx.topology().updateCounters());
}
}
@@ -1037,7 +1037,7 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
m.addFullPartitionsMap(cacheCtx.cacheId(), locMap);
- m.addPartitionUpdateCounters(cacheCtx.cacheId(), cacheCtx.topology().updateCounters(null));
+ m.addPartitionUpdateCounters(cacheCtx.cacheId(), cacheCtx.topology().updateCounters());
}
}
}
@@ -1046,7 +1046,7 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
for (GridClientPartitionTopology top : cctx.exchange().clientTopologies()) {
m.addFullPartitionsMap(top.cacheId(), top.partitionMap(true));
- m.addPartitionUpdateCounters(top.cacheId(), top.updateCounters(null));
+ m.addPartitionUpdateCounters(top.cacheId(), top.updateCounters());
}
if (log.isDebugEnabled())
http://git-wip-us.apache.org/repos/asf/ignite/blob/c938bf13/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 3498470..10fbd89 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
@@ -299,7 +299,7 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
final GridCacheContext<K, V> cctx = cacheContext(ctx);
if (!internal && cctx != null && initUpdCntrs != null) {
- Map<Integer, Long> map = cctx.topology().updateCounters(cctx.affinity().affinityTopologyVersion());
+ Map<Integer, Long> map = cctx.topology().updateCounters();
for (Map.Entry<Integer, Long> e : map.entrySet()) {
Long cntr0 = initUpdCntrs.get(e.getKey());
http://git-wip-us.apache.org/repos/asf/ignite/blob/c938bf13/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
index 0eb22d2..c90a12e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
@@ -227,10 +227,16 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
if (interCache != null && cntrs != null && interCache.context() != null
&& !interCache.isLocal() && !CU.clientNode(ctx.grid().localNode())) {
- Map<Integer, Long> map = interCache.context().topology()
- .updateCounters(interCache.context().affinity().affinityTopologyVersion());
+ GridCacheContext<Object, Object> cctx = interCache.context();
+
+ Map<Integer, Long> map = cctx.topology().updateCounters();
for (Map.Entry<Integer, Long> e : map.entrySet()) {
+ if (!ctx.cache().context().exchange().hasPendingExchange() &&
+ !cctx.affinity().primary(cctx.localNode(), e.getKey(),
+ cctx.affinity().affinityTopologyVersion()))
+ continue;
+
Long cntr0 = cntrs.get(e.getKey());
Long cntr1 = e.getValue();
@@ -239,7 +245,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
}
- routine.handler().updateCounters(topVer, msg.updateCounters());
+ routine.handler().updateCounters(topVer, cntrs);
}
fut.onRemoteRegistered();
@@ -925,8 +931,18 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
GridCacheAdapter cache = ctx.cache().internalCache(hnd0.cacheName());
if (cache != null && !cache.isLocal()) {
- Map<Integer, Long> cntrs = cache.context().topology().updateCounters(
- cache.context().affinity().affinityTopologyVersion());
+ GridCacheContext cctx = cache.context();
+
+ Map<Integer, Long> cntrs = cache.context().topology().updateCounters();
+
+ AffinityTopologyVersion topVer = cctx.affinity().affinityTopologyVersion();
+
+ if (!ctx.cache().context().exchange().hasPendingExchange()) {
+ for (Integer partId : new ArrayList<>(cntrs.keySet())) {
+ if (!cctx.affinity().primary(cctx.localNode(), partId, topVer))
+ cntrs.remove(partId);
+ }
+ }
req.addUpdateCounters(cntrs);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/c938bf13/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 d8f7257..f104f21 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
@@ -416,7 +416,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends GridC
Affinity<Object> aff = grid(i).affinity(null);
- Map<Integer, Long> act = grid(i).cachex(null).context().topology().updateCounters(null);
+ Map<Integer, Long> act = grid(i).cachex(null).context().topology().updateCounters();
for (Map.Entry<Integer, Long> e : updCntrs.entrySet()) {
if (aff.mapPartitionToPrimaryAndBackups(e.getKey()).contains(grid(i).localNode()))
http://git-wip-us.apache.org/repos/asf/ignite/blob/c938bf13/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 f4659dc..025dd80 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
@@ -140,8 +140,6 @@ public class CacheContinuousQueryLostPartitionTest extends GridCommonAbstractTes
// node2 now becomes the primary for the key.
stopGrid(0);
- awaitPartitionMapExchange();
-
cache2.put(key, "2");
// Sanity check.