You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ra...@apache.org on 2015/09/21 18:25:11 UTC
[01/32] ignite git commit: Fixed test to really check query result.
Repository: ignite
Updated Branches:
refs/heads/feature/ignite-535-mqtt 53683e20d -> 296dd6e7d
Fixed test to really check query result.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/dcd8b42a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/dcd8b42a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/dcd8b42a
Branch: refs/heads/feature/ignite-535-mqtt
Commit: dcd8b42aa696bafbb4d7c11a9cf1520d54539001
Parents: ae1e1dc
Author: sboikov <sb...@gridgain.com>
Authored: Tue Sep 15 16:44:05 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue Sep 15 16:44:05 2015 +0300
----------------------------------------------------------------------
...CacheScanPartitionQueryFallbackSelfTest.java | 54 ++++++++------------
1 file changed, 21 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/dcd8b42a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
index df310b4..1ef470a 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
@@ -146,7 +146,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(null, part, false);
- doTestScanQuery(qry);
+ doTestScanQuery(qry, part);
}
finally {
stopAllGrids();
@@ -176,7 +176,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(null, part, false);
- doTestScanQuery(qry);
+ doTestScanQuery(qry, part);
}
finally {
stopAllGrids();
@@ -198,7 +198,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
try {
Ignite ignite = startGrids(GRID_CNT);
- final IgniteCacheProxy<Integer, Integer> cache = fillCache(ignite);
+ fillCache(ignite);
final AtomicBoolean done = new AtomicBoolean(false);
@@ -211,6 +211,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
while (!done.get()) {
startGrid(id);
+
Thread.sleep(3000);
stopGrid(id);
@@ -240,15 +241,10 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
int part = tup.get1();
- try {
- CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(
- null, part, false);
+ CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(
+ null, part, false);
- doTestScanQuery(qry);
- }
- catch (ClusterGroupEmptyCheckedException e) {
- log.warning("Invalid partition: " + part, e);
- }
+ doTestScanQuery(qry, part);
}
return null;
@@ -315,17 +311,12 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
while (!done.get()) {
int part = ThreadLocalRandom.current().nextInt(ignite(nodeId).affinity(null).partitions());
- try {
- QueryCursor<Cache.Entry<Integer, Integer>> cur =
- cache.query(new ScanQuery<Integer, Integer>(part));
+ QueryCursor<Cache.Entry<Integer, Integer>> cur =
+ cache.query(new ScanQuery<Integer, Integer>(part));
- U.debug(log, "Running query [node=" + nodeId + ", part=" + part + ']');
+ U.debug(log, "Running query [node=" + nodeId + ", part=" + part + ']');
- doTestScanQueryCursor(cur, part);
- }
- catch (ClusterGroupEmptyCheckedException e) {
- log.warning("Invalid partition: " + part, e);
- }
+ doTestScanQueryCursor(cur, part);
}
return null;
@@ -404,7 +395,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
}
};
- int part;
+ Integer part = null;
CacheQuery<Map.Entry<Integer, Integer>> qry = null;
if (test.get()) {
@@ -416,7 +407,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
new Thread(run).start();
if (test.get())
- doTestScanQuery(qry);
+ doTestScanQuery(qry, part);
else
latch.await();
}
@@ -454,20 +445,18 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
/**
* @param qry Query.
*/
- protected void doTestScanQuery(
- CacheQuery<Map.Entry<Integer, Integer>> qry) throws IgniteCheckedException {
+ protected void doTestScanQuery(CacheQuery<Map.Entry<Integer, Integer>> qry, int part)
+ throws IgniteCheckedException {
CacheQueryFuture<Map.Entry<Integer, Integer>> fut = qry.execute();
- Collection<Map.Entry<Integer, Integer>> expEntries = fut.get();
+ Collection<Map.Entry<Integer, Integer>> qryEntries = fut.get();
- for (Map.Entry<Integer, Integer> e : expEntries) {
- Map<Integer, Integer> map = entries.get(((GridCacheQueryAdapter)qry).partition());
+ Map<Integer, Integer> map = entries.get(part);
- if (map == null)
- assertTrue(expEntries.isEmpty());
- else
- assertEquals(map.get(e.getKey()), e.getValue());
- }
+ for (Map.Entry<Integer, Integer> e : qryEntries)
+ assertEquals(map.get(e.getKey()), e.getValue());
+
+ assertEquals("Invalid number of entries for partition: " + part, map.size(), qryEntries.size());
}
/**
@@ -484,7 +473,6 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
int cnt = 0;
for (Cache.Entry<Integer, Integer> e : cur) {
-
assertEquals(map.get(e.getKey()), e.getValue());
cnt++;
[29/32] ignite git commit: Merge branch 'ignite-1.4'
Posted by ra...@apache.org.
Merge branch 'ignite-1.4'
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/eae4df1e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/eae4df1e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/eae4df1e
Branch: refs/heads/feature/ignite-535-mqtt
Commit: eae4df1e5207cfa124e25dff5079c1405ae01803
Parents: 1021d4e 8b94494
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Mon Sep 21 17:15:38 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Mon Sep 21 17:15:38 2015 +0300
----------------------------------------------------------------------
.../org/apache/ignite/spi/discovery/tcp/ServerImpl.java | 12 ++++++------
.../dotnet/Apache.Ignite.Core/Impl/IgniteManager.cs | 2 --
modules/yardstick/config/ignite-base-config.xml | 2 +-
3 files changed, 7 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
[23/32] ignite git commit: Yardstick properties file fixed
Posted by ra...@apache.org.
Yardstick properties file fixed
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c28cec1c
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c28cec1c
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c28cec1c
Branch: refs/heads/feature/ignite-535-mqtt
Commit: c28cec1cd3b893adb4b5168af86d67bf5a998854
Parents: b6139f8
Author: agura <ag...@gridgain.com>
Authored: Fri Sep 18 14:50:19 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Fri Sep 18 14:50:19 2015 +0300
----------------------------------------------------------------------
modules/yardstick/config/benchmark-put-indexed-val.properties | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c28cec1c/modules/yardstick/config/benchmark-put-indexed-val.properties
----------------------------------------------------------------------
diff --git a/modules/yardstick/config/benchmark-put-indexed-val.properties b/modules/yardstick/config/benchmark-put-indexed-val.properties
index 8f3a505..e81ae6f 100644
--- a/modules/yardstick/config/benchmark-put-indexed-val.properties
+++ b/modules/yardstick/config/benchmark-put-indexed-val.properties
@@ -21,7 +21,7 @@
# JVM_OPTS=${JVM_OPTS}" -DIGNITE_QUIET=false"
# Uncomment to enable concurrent garbage collection (GC) if you encounter long GC pauses.
-JVM_OPTS=${JVM_OPTS}" -DIGNITE_QUIET=false" \
+JVM_OPTS=${JVM_OPTS}" -DIGNITE_QUIET=false \
-XX:+UseParNewGC \
-XX:+UseConcMarkSweepGC \
-XX:+UseTLAB \
[17/32] ignite git commit: ignite-1452 Cancel cache operations on
node stop
Posted by ra...@apache.org.
ignite-1452 Cancel cache operations on node stop
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/585761f2
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/585761f2
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/585761f2
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 585761f28e8b70487eaf2198d6ea39f7232b088d
Parents: b8c0b30
Author: sboikov <sb...@gridgain.com>
Authored: Thu Sep 17 16:26:02 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Sep 17 16:26:02 2015 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/IgniteKernal.java | 7 ---
.../processors/cache/GridCacheContext.java | 6 +--
.../cache/GridCacheEvictionManager.java | 6 +--
.../cache/GridCacheEvictionResponse.java | 2 +-
.../processors/cache/GridCacheIoManager.java | 47 +++++++++++++-------
.../processors/cache/GridCacheMessage.java | 7 +++
.../processors/cache/GridCacheMvccManager.java | 34 +++++++++++---
.../GridCachePartitionExchangeManager.java | 41 +++++++++++++----
.../processors/cache/GridCacheProcessor.java | 28 ++++++++----
.../GridDistributedLockResponse.java | 6 +--
.../GridDistributedTxPrepareResponse.java | 6 +--
.../distributed/dht/GridDhtTopologyFuture.java | 6 ++-
.../dht/GridDhtTransactionalCacheAdapter.java | 2 +-
.../dht/atomic/GridDhtAtomicUpdateResponse.java | 12 +++--
.../dht/atomic/GridNearAtomicUpdateFuture.java | 16 ++++---
.../dht/atomic/GridNearAtomicUpdateRequest.java | 2 +
.../atomic/GridNearAtomicUpdateResponse.java | 11 ++---
.../colocated/GridDhtColocatedLockFuture.java | 44 ++++++++++++++----
.../dht/preloader/GridDhtForceKeysFuture.java | 2 +-
.../dht/preloader/GridDhtForceKeysResponse.java | 6 +--
.../GridDhtPartitionsExchangeFuture.java | 19 ++++++--
.../distributed/near/GridNearGetResponse.java | 6 +--
.../distributed/near/GridNearLockFuture.java | 26 ++++++++---
.../near/GridNearOptimisticTxPrepareFuture.java | 20 +++++++--
.../near/GridNearTxFinishResponse.java | 6 +--
.../cache/query/GridCacheQueryResponse.java | 6 +--
.../continuous/CacheContinuousQueryHandler.java | 12 +++--
.../transactions/IgniteTxLocalAdapter.java | 4 +-
.../ignite/internal/util/GridSpinBusyLock.java | 10 +++++
.../IgniteCacheEntryProcessorNodeJoinTest.java | 24 +++++++---
.../loadtests/hashmap/GridCacheTestContext.java | 4 +-
.../IgniteCacheQueryNodeRestartSelfTest2.java | 2 -
32 files changed, 292 insertions(+), 138 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/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 daf7d23..82db059 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
@@ -1806,8 +1806,6 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
notifyLifecycleBeansEx(LifecycleEventType.BEFORE_NODE_STOP);
}
- GridCacheProcessor cacheProcessor = ctx.cache();
-
List<GridComponent> comps = ctx.components();
ctx.marshallerContext().onKernalStop();
@@ -1856,11 +1854,6 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
// Note that interrupted flag is cleared.
interrupted = true;
}
- finally {
- // Cleanup even on successful acquire.
- if (cacheProcessor != null)
- cacheProcessor.cancelUserOperations();
- }
}
if (interrupted)
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index 86ba3e6..5385dec 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -283,12 +283,12 @@ public class GridCacheContext<K, V> implements Externalizable {
GridCacheEvictionManager evictMgr,
GridCacheQueryManager<K, V> qryMgr,
CacheContinuousQueryManager contQryMgr,
- GridCacheAffinityManager affMgr,
CacheDataStructuresManager dataStructuresMgr,
GridCacheTtlManager ttlMgr,
GridCacheDrManager drMgr,
CacheConflictResolutionManager<K, V> rslvrMgr,
- CachePluginManager pluginMgr
+ CachePluginManager pluginMgr,
+ GridCacheAffinityManager affMgr
) {
assert ctx != null;
assert sharedCtx != null;
@@ -323,12 +323,12 @@ public class GridCacheContext<K, V> implements Externalizable {
this.evictMgr = add(evictMgr);
this.qryMgr = add(qryMgr);
this.contQryMgr = add(contQryMgr);
- this.affMgr = add(affMgr);
this.dataStructuresMgr = add(dataStructuresMgr);
this.ttlMgr = add(ttlMgr);
this.drMgr = add(drMgr);
this.rslvrMgr = add(rslvrMgr);
this.pluginMgr = add(pluginMgr);
+ this.affMgr = add(affMgr);
log = ctx.log(getClass());
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
index 3e0e2f9..1c34c76 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
@@ -1943,7 +1943,7 @@ public class GridCacheEvictionManager extends GridCacheManagerAdapter {
lock.readLock().unlock();
}
- if (res.error())
+ if (res.evictError())
// Complete future, since there was a class loading error on at least one node.
complete(false);
else
@@ -1985,14 +1985,14 @@ public class GridCacheEvictionManager extends GridCacheManagerAdapter {
boolean err = F.forAny(resMap.values(), new P1<GridCacheEvictionResponse>() {
@Override public boolean apply(GridCacheEvictionResponse res) {
- return res.error();
+ return res.evictError();
}
});
if (err) {
Collection<UUID> ids = F.view(resMap.keySet(), new P1<UUID>() {
@Override public boolean apply(UUID e) {
- return resMap.get(e).error();
+ return resMap.get(e).evictError();
}
});
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionResponse.java
index 4d40c8d..aa3911b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionResponse.java
@@ -116,7 +116,7 @@ public class GridCacheEvictionResponse extends GridCacheMessage {
/**
* @return {@code True} if request processing has finished with error.
*/
- boolean error() {
+ boolean evictError() {
return err;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index b55c84d..421ec82 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -182,8 +182,15 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
c = clsHandlers.get(new ListenerKey(cacheMsg.cacheId(), cacheMsg.getClass()));
if (c == null) {
- U.warn(log, "Received message without registered handler (will ignore) [msg=" + cacheMsg +
- ", nodeId=" + nodeId + ']');
+ if (cctx.kernalContext().isStopping()) {
+ if (log.isDebugEnabled())
+ log.debug("Received message without registered handler (will ignore) [msg=" + cacheMsg +
+ ", nodeId=" + nodeId + ']');
+ }
+ else {
+ U.warn(log, "Received message without registered handler (will ignore) [msg=" + cacheMsg +
+ ", nodeId=" + nodeId + ']');
+ }
return;
}
@@ -596,9 +603,13 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
*
* @param msg Message to send.
* @param destNodeId Destination node ID.
+ * @return {@code True} if should send message.
* @throws IgniteCheckedException If failed.
*/
- private void onSend(GridCacheMessage msg, @Nullable UUID destNodeId) throws IgniteCheckedException {
+ private boolean onSend(GridCacheMessage msg, @Nullable UUID destNodeId) throws IgniteCheckedException {
+ if (msg.error() != null && cctx.kernalContext().isStopping())
+ return false;
+
if (msg.messageId() < 0)
// Generate and set message ID.
msg.messageId(idGen.incrementAndGet());
@@ -609,6 +620,8 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
if (depEnabled && msg instanceof GridCacheDeployable)
cctx.deploy().prepare((GridCacheDeployable)msg);
}
+
+ return true;
}
/**
@@ -624,7 +637,8 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
public void send(ClusterNode node, GridCacheMessage msg, byte plc) throws IgniteCheckedException {
assert !node.isLocal();
- onSend(msg, node.id());
+ if (!onSend(msg, node.id()))
+ return;
if (log.isDebugEnabled())
log.debug("Sending cache message [msg=" + msg + ", node=" + U.toShortString(node) + ']');
@@ -663,12 +677,10 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
* @param msg Message to send.
* @param plc IO policy.
* @param fallback Callback for failed nodes.
- * @return {@code True} if nodes are empty or message was sent, {@code false} if
- * all nodes have left topology while sending this message.
* @throws IgniteCheckedException If send failed.
*/
@SuppressWarnings({"BusyWait", "unchecked"})
- public boolean safeSend(Collection<? extends ClusterNode> nodes, GridCacheMessage msg, byte plc,
+ public void safeSend(Collection<? extends ClusterNode> nodes, GridCacheMessage msg, byte plc,
@Nullable IgnitePredicate<ClusterNode> fallback) throws IgniteCheckedException {
assert nodes != null;
assert msg != null;
@@ -677,10 +689,11 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
if (log.isDebugEnabled())
log.debug("Message will not be sent as collection of nodes is empty: " + msg);
- return true;
+ return;
}
- onSend(msg, null);
+ if (!onSend(msg, null))
+ return;
if (log.isDebugEnabled())
log.debug("Sending cache message [msg=" + msg + ", nodes=" + U.toShortString(nodes) + ']');
@@ -709,7 +722,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
if (fallback != null && !fallback.apply(n))
// If fallback signalled to stop.
- return false;
+ return;
added = true;
}
@@ -721,7 +734,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
log.debug("Message will not be sent because all nodes left topology [msg=" + msg +
", nodes=" + U.toShortString(nodes) + ']');
- return false;
+ return;
}
}
@@ -737,7 +750,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
if (fallback != null && !fallback.apply(n))
// If fallback signalled to stop.
- return false;
+ return;
added = true;
}
@@ -757,7 +770,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
log.debug("Message will not be sent because all nodes left topology [msg=" + msg + ", nodes=" +
U.toShortString(nodes) + ']');
- return false;
+ return;
}
if (log.isDebugEnabled())
@@ -768,8 +781,6 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
if (log.isDebugEnabled())
log.debug("Sent cache message [msg=" + msg + ", nodes=" + U.toShortString(nodes) + ']');
-
- return true;
}
/**
@@ -800,7 +811,8 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
*/
public void sendOrderedMessage(ClusterNode node, Object topic, GridCacheMessage msg, byte plc,
long timeout) throws IgniteCheckedException {
- onSend(msg, node.id());
+ if (!onSend(msg, node.id()))
+ return;
int cnt = 0;
@@ -854,7 +866,8 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
assert node != null;
assert msg != null;
- onSend(msg, null);
+ if (!onSend(msg, null))
+ return;
try {
cctx.gridIO().send(node, TOPIC_CACHE, msg, plc);
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java
index 4e737a0..55688e4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMessage.java
@@ -77,6 +77,13 @@ public abstract class GridCacheMessage implements Message {
protected int cacheId;
/**
+ * @return Error, if any.
+ */
+ @Nullable public Throwable error() {
+ return null;
+ }
+
+ /**
* Gets next ID for indexed message ID.
*
* @return Message ID.
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
index 555bbda..e2d0302 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
@@ -120,6 +120,9 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
@SuppressWarnings( {"FieldAccessedSynchronizedAndUnsynchronized"})
private IgniteLogger exchLog;
+ /** */
+ private volatile boolean stopping;
+
/** Lock callback. */
@GridToStringExclude
private final GridCacheMvccCallback cb = new GridCacheMvccCallback() {
@@ -325,8 +328,10 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
/**
* Cancels all client futures.
*/
- public void cancelClientFutures() {
- cancelClientFutures(new IgniteCheckedException("Operation has been cancelled (node is stopping)."));
+ public void onStop() {
+ stopping = true;
+
+ cancelClientFutures(stopError());
}
/** {@inheritDoc} */
@@ -362,6 +367,13 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
}
/**
+ * @return Node stop exception.
+ */
+ private IgniteCheckedException stopError() {
+ return new IgniteCheckedException("Operation has been cancelled (node is stopping).");
+ }
+
+ /**
* @param from From version.
* @return To version.
*/
@@ -385,8 +397,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
assert old == null : "Old future is not null [futVer=" + futVer + ", fut=" + fut + ", old=" + old + ']';
- if (cctx.kernalContext().clientDisconnected())
- ((GridFutureAdapter)fut).onDone(disconnectedError(null));
+ onFutureAdded(fut);
}
/**
@@ -507,17 +518,26 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
fut.onNodeLeft(n.id());
}
- if (cctx.kernalContext().clientDisconnected())
- ((GridFutureAdapter)fut).onDone(disconnectedError(null));
-
// Just in case if future was completed before it was added.
if (fut.isDone())
removeFuture(fut);
+ else
+ onFutureAdded(fut);
return true;
}
/**
+ * @param fut Future.
+ */
+ private void onFutureAdded(IgniteInternalFuture<?> fut) {
+ if (stopping)
+ ((GridFutureAdapter)fut).onDone(stopError());
+ else if (cctx.kernalContext().clientDisconnected())
+ ((GridFutureAdapter)fut).onDone(disconnectedError(null));
+ }
+
+ /**
* @param fut Future to remove.
* @return {@code True} if removed.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 20340d1..34c571c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -147,6 +147,9 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
*/
private ExchangeFutureSet exchFuts = new ExchangeFutureSet();
+ /** */
+ private volatile IgniteCheckedException stopErr;
+
/** Discovery listener. */
private final GridLocalEventListener discoLsnr = new GridLocalEventListener() {
@Override public void onEvent(Event evt) {
@@ -381,7 +384,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
cctx.io().removeHandler(0, GridDhtPartitionsFullMessage.class);
cctx.io().removeHandler(0, GridDhtPartitionsSingleRequest.class);
- IgniteCheckedException err = cctx.kernalContext().clientDisconnected() ?
+ stopErr = cctx.kernalContext().clientDisconnected() ?
new IgniteClientDisconnectedCheckedException(cctx.kernalContext().cluster().clientReconnectFuture(),
"Client node disconnected: " + cctx.gridName()) :
new IgniteInterruptedCheckedException("Node is stopping: " + cctx.gridName());
@@ -391,11 +394,17 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
if (exchFuts0 != null) {
for (GridDhtPartitionsExchangeFuture f : exchFuts.values())
- f.onDone(err);
+ f.onDone(stopErr);
}
for (AffinityReadyFuture f : readyFuts.values())
- f.onDone(err);
+ f.onDone(stopErr);
+
+ for (GridDhtPartitionsExchangeFuture f : pendingExchangeFuts)
+ f.onDone(stopErr);
+
+ if (locExchFut != null)
+ locExchFut.onDone(stopErr);
U.cancel(exchWorker);
@@ -519,6 +528,8 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
fut.onDone(topVer);
}
+ else if (stopErr != null)
+ fut.onDone(stopErr);
return fut;
}
@@ -791,6 +802,9 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
if (discoEvt != null)
fut.onEvent(exchId, discoEvt);
+ if (stopErr != null)
+ fut.onDone(stopErr);
+
return fut;
}
@@ -799,12 +813,12 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
* @param err Error.
*/
public void onExchangeDone(GridDhtPartitionsExchangeFuture exchFut, @Nullable Throwable err) {
- if (err == null) {
- AffinityTopologyVersion topVer = exchFut.topologyVersion();
+ AffinityTopologyVersion topVer = exchFut.topologyVersion();
- if (log.isDebugEnabled())
- log.debug("Exchange done [topVer=" + topVer + ", fut=" + exchFut + ']');
+ if (log.isDebugEnabled())
+ log.debug("Exchange done [topVer=" + topVer + ", fut=" + exchFut + ", err=" + err + ']');
+ if (err == null) {
while (true) {
AffinityTopologyVersion readyVer = readyTopVer.get();
@@ -825,8 +839,17 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
}
}
}
- else if (log.isDebugEnabled())
- log.debug("Exchange done with error [fut=" + exchFut + ", err=" + err + ']');
+ else {
+ for (Map.Entry<AffinityTopologyVersion, AffinityReadyFuture> entry : readyFuts.entrySet()) {
+ if (entry.getKey().compareTo(topVer) <= 0) {
+ if (log.isDebugEnabled())
+ log.debug("Completing created topology ready future with error " +
+ "[ver=" + topVer + ", fut=" + entry.getValue() + ']');
+
+ entry.getValue().onDone(err);
+ }
+ }
+ }
ExchangeFutureSet exchFuts0 = exchFuts;
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 4ae0baa..c92de7d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -960,6 +960,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
}
+ cancelFutures();
+
List<? extends GridCacheSharedManager<?, ?>> sharedMgrs = sharedCtx.managers();
for (ListIterator<? extends GridCacheSharedManager<?, ?>> it = sharedMgrs.listIterator(sharedMgrs.size());
@@ -1323,12 +1325,12 @@ public class GridCacheProcessor extends GridProcessorAdapter {
evictMgr,
qryMgr,
contQryMgr,
- affMgr,
dataStructuresMgr,
ttlMgr,
drMgr,
rslvrMgr,
- pluginMgr
+ pluginMgr,
+ affMgr
);
cacheCtx.cacheObjectContext(cacheObjCtx);
@@ -1452,12 +1454,12 @@ public class GridCacheProcessor extends GridProcessorAdapter {
evictMgr,
qryMgr,
contQryMgr,
- affMgr,
dataStructuresMgr,
ttlMgr,
drMgr,
rslvrMgr,
- pluginMgr
+ pluginMgr,
+ affMgr
);
cacheCtx.cacheObjectContext(cacheObjCtx);
@@ -2325,9 +2327,14 @@ public class GridCacheProcessor extends GridProcessorAdapter {
try {
ctx.discovery().sendCustomEvent(new DynamicCacheChangeBatch(sndReqs));
- if (ctx.clientDisconnected())
+ if (ctx.isStopping()) {
+ err = new IgniteCheckedException("Failed to execute dynamic cache change request, " +
+ "node is stopping.");
+ }
+ else if (ctx.clientDisconnected()) {
err = new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
"Failed to execute dynamic cache change request, client node disconnected.");
+ }
}
catch (IgniteCheckedException e) {
err = e;
@@ -3036,9 +3043,14 @@ public class GridCacheProcessor extends GridProcessorAdapter {
try {
ctx.discovery().sendCustomEvent(new DynamicCacheChangeBatch(Collections.singleton(req)));
- if (ctx.clientDisconnected())
+ if (ctx.isStopping()) {
+ err = new IgniteCheckedException("Failed to execute dynamic cache change request, " +
+ "node is stopping.");
+ }
+ else if (ctx.clientDisconnected()) {
err = new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
"Failed to execute dynamic cache change request, client node disconnected.");
+ }
}
catch (IgniteCheckedException e) {
err = e;
@@ -3104,8 +3116,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
/**
* Cancel all user operations.
*/
- public void cancelUserOperations() {
- sharedCtx.mvcc().cancelClientFutures();
+ private void cancelFutures() {
+ sharedCtx.mvcc().onStop();
Exception err = new IgniteCheckedException("Operation has been cancelled (node is stopping).");
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
index cdb878d..8a95b14 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
@@ -137,10 +137,8 @@ public class GridDistributedLockResponse extends GridDistributedBaseMessage {
return futId;
}
- /**
- * @return Error.
- */
- public Throwable error() {
+ /** {@inheritDoc} */
+ @Override public Throwable error() {
return err;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareResponse.java
index 4264830..e798458 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareResponse.java
@@ -67,10 +67,8 @@ public class GridDistributedTxPrepareResponse extends GridDistributedBaseMessage
this.err = err;
}
- /**
- * @return Error.
- */
- public Throwable error() {
+ /** {@inheritDoc} */
+ @Override public Throwable error() {
return err;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFuture.java
index c11a3d7..6ade26f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFuture.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.cache.distributed.dht;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.jetbrains.annotations.Nullable;
/**
* Future that implements a barrier after which dht topology is safe to use. Topology is considered to be
@@ -38,9 +39,10 @@ public interface GridDhtTopologyFuture extends IgniteInternalFuture<AffinityTopo
public AffinityTopologyVersion topologyVersion();
/**
- * Returns is cache topology valid.
+ * Returns error is cache topology is not valid.
+ *
* @param cctx Cache context.
* @return valid ot not.
*/
- public boolean isCacheTopologyValid(GridCacheContext cctx);
+ @Nullable public Throwable validateCache(GridCacheContext cctx);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index b9514a9..1a869e7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -1217,7 +1217,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
Throwable err = res.error();
// Log error before sending reply.
- if (err != null && !(err instanceof GridCacheLockTimeoutException))
+ if (err != null && !(err instanceof GridCacheLockTimeoutException) && !ctx.kernalContext().isStopping())
U.error(log, "Failed to acquire lock for request: " + req, err);
try {
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
index 33651bc..04d36e5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
@@ -97,16 +97,15 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
/**
* Sets update error.
- * @param err
+ *
+ * @param err Error.
*/
public void onError(IgniteCheckedException err){
this.err = err;
}
- /**
- * @return Gets update error.
- */
- public IgniteCheckedException error() {
+ /** {@inheritDoc} */
+ @Override public IgniteCheckedException error() {
return err;
}
@@ -154,8 +153,7 @@ public class GridDhtAtomicUpdateResponse extends GridCacheMessage implements Gri
nearEvicted.add(key);
}
- /** {@inheritDoc}
- * @param ctx*/
+ /** {@inheritDoc} */
@Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
super.prepareMarshal(ctx);
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
index d93f68f..fb2c5ad 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
@@ -385,9 +385,10 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
GridDhtTopologyFuture fut = cache.topology().topologyVersionFuture();
if (fut.isDone()) {
- if (!fut.isCacheTopologyValid(cctx)) {
- onDone(new IgniteCheckedException("Failed to perform cache operation (cache topology is not valid): " +
- cctx.name()));
+ Throwable err = fut.validateCache(cctx);
+
+ if (err != null) {
+ onDone(err);
return;
}
@@ -811,6 +812,7 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
}
Exception err = null;
+ GridNearAtomicUpdateRequest singleReq0 = null;
Map<UUID, GridNearAtomicUpdateRequest> pendingMappings = null;
int size = keys.size();
@@ -837,13 +839,13 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
if (size == 1 && !fastMap) {
assert remapKeys == null || remapKeys.size() == 1;
- singleReq = mapSingleUpdate();
+ singleReq0 = singleReq = mapSingleUpdate();
}
else {
pendingMappings = mapUpdate(topNodes);
if (pendingMappings.size() == 1)
- singleReq = F.firstValue(pendingMappings);
+ singleReq0 = singleReq = F.firstValue(pendingMappings);
else {
if (syncMode == PRIMARY_SYNC) {
mappings = U.newHashMap(pendingMappings.size());
@@ -874,8 +876,8 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
}
// Optimize mapping for single key.
- if (singleReq != null)
- mapSingle(singleReq.nodeId(), singleReq);
+ if (singleReq0 != null)
+ mapSingle(singleReq0.nodeId(), singleReq0);
else {
assert pendingMappings != null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
index 5f5fbb5..ccb67d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
@@ -198,6 +198,8 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
boolean skipStore,
boolean clientReq
) {
+ assert futVer != null;
+
this.cacheId = cacheId;
this.nodeId = nodeId;
this.futVer = futVer;
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java
index 8bc145c..376f4ec 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java
@@ -116,6 +116,8 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
* @param futVer Future version.
*/
public GridNearAtomicUpdateResponse(int cacheId, UUID nodeId, GridCacheVersion futVer) {
+ assert futVer != null;
+
this.cacheId = cacheId;
this.nodeId = nodeId;
this.futVer = futVer;
@@ -149,16 +151,15 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
/**
* Sets update error.
- * @param err
+ *
+ * @param err Error.
*/
public void error(IgniteCheckedException err){
this.err = err;
}
- /**
- * @return Update error, if any.
- */
- public IgniteCheckedException error() {
+ /** {@inheritDoc} */
+ @Override public IgniteCheckedException error() {
return err;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
index 596ec77..1a08265 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
@@ -57,6 +57,7 @@ import org.apache.ignite.internal.util.future.GridEmbeddedFuture;
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.C1;
import org.apache.ignite.internal.util.typedef.C2;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
@@ -524,7 +525,22 @@ public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(GridDhtColocatedLockFuture.class, this, "inTx", inTx(), "super", super.toString());
+ Collection<String> futs = F.viewReadOnly(futures(), new C1<IgniteInternalFuture<?>, String>() {
+ @Override public String apply(IgniteInternalFuture<?> f) {
+ if (isMini(f)) {
+ MiniFuture m = (MiniFuture)f;
+
+ return "[node=" + m.node().id() + ", loc=" + m.node().isLocal() + ", done=" + f.isDone() + "]";
+ }
+ else
+ return "[loc=true, done=" + f.isDone() + "]";
+ }
+ });
+
+ return S.toString(GridDhtColocatedLockFuture.class, this,
+ "innerFuts", futs,
+ "inTx", inTx(),
+ "super", super.toString());
}
/**
@@ -565,9 +581,10 @@ public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture
if (topVer != null) {
for (GridDhtTopologyFuture fut : cctx.shared().exchange().exchangeFutures()){
if (fut.topologyVersion().equals(topVer)){
- if (!fut.isCacheTopologyValid(cctx)) {
- onDone(new IgniteCheckedException("Failed to perform cache operation (cache topology is not valid): " +
- cctx.name()));
+ Throwable err = fut.validateCache(cctx);
+
+ if (err != null) {
+ onDone(err);
return;
}
@@ -612,9 +629,10 @@ public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture
GridDhtTopologyFuture fut = cctx.topologyVersionFuture();
if (fut.isDone()) {
- if (!fut.isCacheTopologyValid(cctx)) {
- onDone(new IgniteCheckedException("Failed to perform cache operation (cache topology is not valid): " +
- cctx.name()));
+ Throwable err = fut.validateCache(cctx);
+
+ if (err != null) {
+ onDone(err);
return;
}
@@ -643,10 +661,15 @@ public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture
}
else {
fut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() {
- @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> t) {
+ @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> fut) {
try {
+ fut.get();
+
mapOnTopology(remap, c);
}
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
finally {
cctx.shared().txContextReset();
}
@@ -1327,8 +1350,13 @@ public final class GridDhtColocatedLockFuture extends GridCompoundIdentityFuture
affFut.listen(new CI1<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> fut) {
try {
+ fut.get();
+
remap();
}
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
finally {
cctx.shared().txContextReset();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
index 36a2da1..eaed424 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
@@ -283,7 +283,7 @@ public final class GridDhtForceKeysFuture<K, V> extends GridCompoundFuture<Objec
// Fail the whole thing.
if (e instanceof ClusterTopologyCheckedException)
fut.onResult((ClusterTopologyCheckedException)e);
- else
+ else if (!cctx.kernalContext().isStopping())
fut.onResult(e);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysResponse.java
index d31f096..93e39ce 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysResponse.java
@@ -98,10 +98,8 @@ public class GridDhtForceKeysResponse extends GridCacheMessage implements GridCa
this.err = err;
}
- /**
- * @return Error, if any.
- */
- public IgniteCheckedException error() {
+ /** {@inheritDoc} */
+ @Override public IgniteCheckedException error() {
return err;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/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 865bbdc..a1b03c1 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
@@ -1081,9 +1081,22 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
}
/** {@inheritDoc} */
- @Override public boolean isCacheTopologyValid(GridCacheContext cctx) {
- return cctx.config().getTopologyValidator() != null && cacheValidRes.containsKey(cctx.cacheId()) ?
- cacheValidRes.get(cctx.cacheId()) : true;
+ @Override public Throwable validateCache(GridCacheContext cctx) {
+ Throwable err = error();
+
+ if (err != null)
+ return err;
+
+ if (cctx.config().getTopologyValidator() != null) {
+ Boolean res = cacheValidRes.get(cctx.cacheId());
+
+ if (res != null && !res) {
+ return new IgniteCheckedException("Failed to perform cache operation " +
+ "(cache topology is not valid): " + cctx.name());
+ }
+ }
+
+ return null;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java
index 3276377..d4493a2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetResponse.java
@@ -163,10 +163,8 @@ public class GridNearGetResponse extends GridCacheMessage implements GridCacheDe
return topVer != null ? topVer : super.topologyVersion();
}
- /**
- * @return Error.
- */
- public IgniteCheckedException error() {
+ /** {@inheritDoc} */
+ @Override public IgniteCheckedException error() {
return err;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
index f3e5ca3..dcc8da6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
@@ -703,9 +703,10 @@ public final class GridNearLockFuture extends GridCompoundIdentityFuture<Boolean
if (topVer != null) {
for (GridDhtTopologyFuture fut : cctx.shared().exchange().exchangeFutures()){
if (fut.topologyVersion().equals(topVer)){
- if (!fut.isCacheTopologyValid(cctx)) {
- onDone(new IgniteCheckedException("Failed to perform cache operation (cache topology is not valid): " +
- cctx.name()));
+ Throwable err = fut.validateCache(cctx);
+
+ if (err != null) {
+ onDone(err);
return;
}
@@ -749,9 +750,10 @@ public final class GridNearLockFuture extends GridCompoundIdentityFuture<Boolean
GridDhtTopologyFuture fut = cctx.topologyVersionFuture();
if (fut.isDone()) {
- if (!fut.isCacheTopologyValid(cctx)) {
- onDone(new IgniteCheckedException("Failed to perform cache operation (cache topology is not valid): " +
- cctx.name()));
+ Throwable err = fut.validateCache(cctx);
+
+ if (err != null) {
+ onDone(err);
return;
}
@@ -777,10 +779,15 @@ public final class GridNearLockFuture extends GridCompoundIdentityFuture<Boolean
}
else {
fut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() {
- @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> t) {
+ @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> fut) {
try {
+ fut.get();
+
mapOnTopology(remap);
}
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
finally {
cctx.shared().txContextReset();
}
@@ -1435,8 +1442,13 @@ public final class GridNearLockFuture extends GridCompoundIdentityFuture<Boolean
affFut.listen(new CI1<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> fut) {
try {
+ fut.get();
+
remap();
}
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
finally {
cctx.shared().txContextReset();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
index 2048fdf..25028c4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
@@ -319,7 +319,9 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearTxPrepareFutureAd
assert ctx != null : cacheId;
- if (!topFut.isCacheTopologyValid(ctx)) {
+ Throwable err = topFut.validateCache(ctx);
+
+ if (err != null) {
if (invalidCaches != null)
invalidCaches.append(", ");
else
@@ -343,12 +345,17 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearTxPrepareFutureAd
}
else {
topFut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() {
- @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> t) {
+ @Override public void apply(final IgniteInternalFuture<AffinityTopologyVersion> fut) {
cctx.kernalContext().closure().runLocalSafe(new GridPlainRunnable() {
@Override public void run() {
try {
+ fut.get();
+
prepareOnTopology(remap, c);
}
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
finally {
cctx.txContextReset();
}
@@ -841,7 +848,14 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearTxPrepareFutureAd
if (affFut != null && !affFut.isDone()) {
affFut.listen(new CI1<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> fut) {
- remap();
+ try {
+ fut.get();
+
+ remap();
+ }
+ catch (IgniteCheckedException e) {
+ onDone(e);
+ }
}
});
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishResponse.java
index cec7d73..c860baa 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishResponse.java
@@ -75,10 +75,8 @@ public class GridNearTxFinishResponse extends GridDistributedTxFinishResponse {
this.err = err;
}
- /**
- * @return Error.
- */
- @Nullable public Throwable error() {
+ /** {@inheritDoc} */
+ @Nullable @Override public Throwable error() {
return err;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryResponse.java
index 3e4cdeb..78e2ac7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryResponse.java
@@ -193,10 +193,8 @@ public class GridCacheQueryResponse extends GridCacheMessage implements GridCach
return reqId;
}
- /**
- * @return Error.
- */
- public Throwable error() {
+ /** {@inheritDoc} */
+ @Override public Throwable error() {
return err;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/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 df6b4b7..c99e07f 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
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.processors.continuous.GridContinuousHandler;
import org.apache.ignite.internal.processors.platform.cache.query.PlatformContinuousQueryFilter;
import org.apache.ignite.internal.util.typedef.C1;
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.jetbrains.annotations.Nullable;
@@ -97,6 +98,9 @@ class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
/** Whether to skip primary check for REPLICATED cache. */
private transient boolean skipPrimaryCheck;
+ /** */
+ private transient int cacheId;
+
/**
* Required by {@link Externalizable}.
*/
@@ -145,6 +149,8 @@ class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
this.ignoreExpired = ignoreExpired;
this.taskHash = taskHash;
this.skipPrimaryCheck = skipPrimaryCheck;
+
+ cacheId = CU.cacheId(cacheName);
}
/** {@inheritDoc} */
@@ -457,6 +463,8 @@ class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
sync = in.readBoolean();
ignoreExpired = in.readBoolean();
taskHash = in.readInt();
+
+ cacheId = CU.cacheId(cacheName);
}
/**
@@ -466,9 +474,7 @@ class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler {
private GridCacheContext<K, V> cacheContext(GridKernalContext ctx) {
assert ctx != null;
- GridCacheAdapter<K, V> cache = ctx.cache().internalCache(cacheName);
-
- return cache == null ? null : cache.context();
+ return ctx.cache().<K, V>context().cacheContext(cacheId);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index 00b91dd..6ca1f72 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1105,6 +1105,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
/**
* Commits transaction to transaction manager. Used for one-phase commit transactions only.
+ *
+ * @param commit If {@code true} commits transaction, otherwise rollbacks.
*/
public void tmFinish(boolean commit) {
assert onePhaseCommit();
@@ -1118,7 +1120,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
state(commit ? COMMITTED : ROLLED_BACK);
- boolean needsCompletedVersions = needsCompletedVersions();
+ boolean needsCompletedVersions = commit && needsCompletedVersions();
assert !needsCompletedVersions || completedBase != null;
assert !needsCompletedVersions || committedVers != null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinBusyLock.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinBusyLock.java b/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinBusyLock.java
index 2aae6ef..6bfd4fb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinBusyLock.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/GridSpinBusyLock.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.util;
+import java.util.concurrent.TimeUnit;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
/**
@@ -76,6 +77,15 @@ public class GridSpinBusyLock {
}
/**
+ * @param millis Timeout.
+ * @return {@code True} if lock was acquired.
+ * @throws InterruptedException If interrupted.
+ */
+ public boolean tryBlock(long millis) throws InterruptedException {
+ return lock.tryWriteLock(millis, TimeUnit.MILLISECONDS);
+ }
+
+ /**
* Makes possible for activities entering busy state again.
*/
public void unblock() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryProcessorNodeJoinTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryProcessorNodeJoinTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryProcessorNodeJoinTest.java
index 6b4d473..151167a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryProcessorNodeJoinTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryProcessorNodeJoinTest.java
@@ -25,6 +25,7 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.EntryProcessorResult;
import javax.cache.processor.MutableEntry;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheAtomicityMode;
@@ -184,20 +185,29 @@ public class IgniteCacheEntryProcessorNodeJoinTest extends GridCommonAbstractTes
String val = "value-" + k;
- cache.invoke(key, new Processor(val));
+ procs.put(key, new Processor(val));
}
- cache.invokeAll(procs);
+ Map<String, EntryProcessorResult<Integer>> resMap = cache.invokeAll(procs);
+
+ for (String key : procs.keySet()) {
+ EntryProcessorResult<Integer> res = resMap.get(key);
+
+ assertNotNull(res);
+ assertEquals(k + 1, (Object) res.get());
+ }
}
else {
+ IgniteCache<String, Set<String>> cache = ignite(0).cache(null);
+
for (int i = 0; i < NUM_SETS; i++) {
String key = "set-" + i;
String val = "value-" + k;
- IgniteCache<String, Set<String>> cache = ignite(0).cache(null);
+ Integer valsCnt = cache.invoke(key, new Processor(val));
- cache.invoke(key, new Processor(val));
+ assertEquals(k + 1, (Object)valsCnt);
}
}
}
@@ -275,7 +285,7 @@ public class IgniteCacheEntryProcessorNodeJoinTest extends GridCommonAbstractTes
}
/** */
- private static class Processor implements EntryProcessor<String, Set<String>, Void>, Serializable {
+ private static class Processor implements EntryProcessor<String, Set<String>, Integer>, Serializable {
/** */
private String val;
@@ -287,7 +297,7 @@ public class IgniteCacheEntryProcessorNodeJoinTest extends GridCommonAbstractTes
}
/** {@inheritDoc} */
- @Override public Void process(MutableEntry<String, Set<String>> e, Object... args) {
+ @Override public Integer process(MutableEntry<String, Set<String>> e, Object... args) {
Set<String> vals = e.getValue();
if (vals == null)
@@ -297,7 +307,7 @@ public class IgniteCacheEntryProcessorNodeJoinTest extends GridCommonAbstractTes
e.setValue(vals);
- return null;
+ return vals.size();
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java b/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
index 7aae48c..88605b4 100644
--- a/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
+++ b/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
@@ -79,12 +79,12 @@ public class GridCacheTestContext<K, V> extends GridCacheContext<K, V> {
new GridCacheEvictionManager(),
new GridCacheLocalQueryManager<K, V>(),
new CacheContinuousQueryManager(),
- new GridCacheAffinityManager(),
new CacheDataStructuresManager(),
new GridCacheTtlManager(),
new GridOsCacheDrManager(),
new CacheOsConflictResolutionManager<K, V>(),
- new CachePluginManager(ctx, new CacheConfiguration())
+ new CachePluginManager(ctx, new CacheConfiguration()),
+ new GridCacheAffinityManager()
);
store().initialize(null, new IdentityHashMap<CacheStore, ThreadLocal>());
http://git-wip-us.apache.org/repos/asf/ignite/blob/585761f2/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest2.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest2.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest2.java
index 1276405..e00611b 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest2.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCacheQueryNodeRestartSelfTest2.java
@@ -185,8 +185,6 @@ public class IgniteCacheQueryNodeRestartSelfTest2 extends GridCommonAbstractTest
* @throws Exception If failed.
*/
public void testRestarts() throws Exception {
- fail("https://issues.apache.org/jira/browse/IGNITE-1452");
-
int duration = 90 * 1000;
int qryThreadNum = 4;
int restartThreadsNum = 2; // 4 + 2 = 6 nodes
[28/32] ignite git commit: ignite-1500 Invalid docs in distributed
data structures classes
Posted by ra...@apache.org.
ignite-1500 Invalid docs in distributed data structures classes
Signed-off-by: Anton Vinogradov <av...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/1021d4ea
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/1021d4ea
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/1021d4ea
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 1021d4eaf4b8318786f3ea7e44e5acef2fbb9004
Parents: c28cec1
Author: agura <ag...@gridgain.com>
Authored: Fri Sep 18 15:24:46 2015 +0300
Committer: Anton Vinogradov <av...@apache.org>
Committed: Mon Sep 21 10:34:35 2015 +0300
----------------------------------------------------------------------
.../java/org/apache/ignite/IgniteAtomicLong.java | 15 ++++++---------
.../org/apache/ignite/IgniteAtomicReference.java | 9 +++------
.../java/org/apache/ignite/IgniteAtomicSequence.java | 9 +++------
.../java/org/apache/ignite/IgniteAtomicStamped.java | 13 +++++--------
4 files changed, 17 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/1021d4ea/modules/core/src/main/java/org/apache/ignite/IgniteAtomicLong.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteAtomicLong.java b/modules/core/src/main/java/org/apache/ignite/IgniteAtomicLong.java
index 83e2525..77a5830 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteAtomicLong.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteAtomicLong.java
@@ -26,29 +26,26 @@ import java.io.Closeable;
* Distributed atomic long includes the following main functionality:
* <ul>
* <li>
- * Method {@link #get()} synchronously gets current value of atomic long.
+ * Method {@link #get()} gets current value of atomic long.
* </li>
* <li>
- * Various {@code get..(..)} methods synchronously get current value of atomic long
+ * Various {@code get..(..)} methods get current value of atomic long
* and increase or decrease value of atomic long.
* </li>
* <li>
- * Method {@link #addAndGet(long l)} synchronously sums {@code l} with current value of atomic long
+ * Method {@link #addAndGet(long l)} sums {@code l} with current value of atomic long
* and returns result.
* </li>
* <li>
- * Method {@link #incrementAndGet()} synchronously increases value of atomic long and returns result.
+ * Method {@link #incrementAndGet()} increases value of atomic long and returns result.
* </li>
* <li>
- * Method {@link #decrementAndGet()} synchronously decreases value of atomic long and returns result.
+ * Method {@link #decrementAndGet()} decreases value of atomic long and returns result.
* </li>
* <li>
- * Method {@link #getAndSet(long l)} synchronously gets current value of atomic long and sets {@code l}
+ * Method {@link #getAndSet(long l)} gets current value of atomic long and sets {@code l}
* as value of atomic long.
* </li>
- * </ul>
- * All previously described methods have asynchronous analogs.
- * <ul>
* <li>
* Method {@link #name()} gets name of atomic long.
* </li>
http://git-wip-us.apache.org/repos/asf/ignite/blob/1021d4ea/modules/core/src/main/java/org/apache/ignite/IgniteAtomicReference.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteAtomicReference.java b/modules/core/src/main/java/org/apache/ignite/IgniteAtomicReference.java
index a33cf35..a1e6e8e 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteAtomicReference.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteAtomicReference.java
@@ -26,17 +26,14 @@ import java.io.Closeable;
* Distributed atomic reference includes the following main functionality:
* <ul>
* <li>
- * Method {@link #get()} synchronously gets current value of an atomic reference.
+ * Method {@link #get()} gets current value of an atomic reference.
* </li>
* <li>
- * Method {@link #set(Object)} synchronously and unconditionally sets the value in the an atomic reference.
+ * Method {@link #set(Object)} unconditionally sets the value in the an atomic reference.
* </li>
* <li>
- * Methods {@code compareAndSet(...)} synchronously and conditionally set the value in the an atomic reference.
+ * Methods {@code compareAndSet(...)} conditionally set the value in the an atomic reference.
* </li>
- * </ul>
- * All previously described methods have asynchronous analogs.
- * <ul>
* <li>
* Method {@link #name()} gets name of atomic reference.
* </li>
http://git-wip-us.apache.org/repos/asf/ignite/blob/1021d4ea/modules/core/src/main/java/org/apache/ignite/IgniteAtomicSequence.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteAtomicSequence.java b/modules/core/src/main/java/org/apache/ignite/IgniteAtomicSequence.java
index 313aba5..a1e1392 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteAtomicSequence.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteAtomicSequence.java
@@ -26,19 +26,16 @@ import java.io.Closeable;
* Distributed atomic sequence includes the following main functionality:
* <ul>
* <li>
- * Method {@link #get()} synchronously gets current value from atomic sequence.
+ * Method {@link #get()} gets current value from atomic sequence.
* </li>
* <li>
- * Various {@code get..(..)} methods synchronously get current value from atomic sequence
+ * Various {@code get..(..)} methods get current value from atomic sequence
* and increase atomic sequences value.
* </li>
* <li>
- * Various {@code add..(..)} {@code increment(..)} methods synchronously increase atomic sequences value
+ * Various {@code add..(..)} {@code increment(..)} methods increase atomic sequences value
* and return increased value.
* </li>
- * </ul>
- * All previously described methods have asynchronous analogs.
- * <ul>
* <li>
* Method {@link #batchSize(int size)} sets batch size of current atomic sequence.
* </li>
http://git-wip-us.apache.org/repos/asf/ignite/blob/1021d4ea/modules/core/src/main/java/org/apache/ignite/IgniteAtomicStamped.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteAtomicStamped.java b/modules/core/src/main/java/org/apache/ignite/IgniteAtomicStamped.java
index c05fdae..6c73248 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteAtomicStamped.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteAtomicStamped.java
@@ -27,25 +27,22 @@ import org.apache.ignite.lang.IgniteBiTuple;
* Distributed atomic stamped includes the following main functionality:
* <ul>
* <li>
- * Method {@link #get()} synchronously gets both value and stamp of atomic.
+ * Method {@link #get()} gets both value and stamp of atomic.
* </li>
* <li>
- * Method {@link #value()} synchronously gets current value of atomic.
+ * Method {@link #value()} gets current value of atomic.
* </li>
* <li>
- * Method {@link #stamp()} synchronously gets current stamp of atomic.
+ * Method {@link #stamp()} gets current stamp of atomic.
* </li>
* <li>
- * Method {@link #set(Object, Object)} synchronously and unconditionally sets the value
+ * Method {@link #set(Object, Object)} unconditionally sets the value
* and the stamp in the atomic.
* </li>
* <li>
- * Methods {@code compareAndSet(...)} synchronously and conditionally set the value
+ * Methods {@code compareAndSet(...)} conditionally set the value
* and the stamp in the atomic.
* </li>
- * </ul>
- * All previously described methods have asynchronous analogs.
- * <ul>
* <li>
* Method {@link #name()} gets name of atomic stamped.
* </li>
[21/32] ignite git commit: Fixed yardstick config.
Posted by ra...@apache.org.
Fixed yardstick config.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/4c101dc2
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/4c101dc2
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/4c101dc2
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 4c101dc23be2365c6983381cac9dbe68873040d8
Parents: 3676cbe
Author: sboikov <sb...@gridgain.com>
Authored: Fri Sep 18 14:11:07 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Sep 18 14:11:07 2015 +0300
----------------------------------------------------------------------
modules/yardstick/config/ignite-base-config.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/4c101dc2/modules/yardstick/config/ignite-base-config.xml
----------------------------------------------------------------------
diff --git a/modules/yardstick/config/ignite-base-config.xml b/modules/yardstick/config/ignite-base-config.xml
index 6e94b3c..c77cc9a 100644
--- a/modules/yardstick/config/ignite-base-config.xml
+++ b/modules/yardstick/config/ignite-base-config.xml
@@ -25,7 +25,7 @@
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<bean id="base-ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration" abstract="true">
- <property name="peerClassLoadingEnabled" value="true"/>
+ <property name="peerClassLoadingEnabled" value="false"/>
<property name="metricsLogFrequency" value="5000"/>
[16/32] ignite git commit: ignite-1.4 Fixed Visor cmd options.
Posted by ra...@apache.org.
ignite-1.4 Fixed Visor cmd options.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b8c0b308
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b8c0b308
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b8c0b308
Branch: refs/heads/feature/ignite-535-mqtt
Commit: b8c0b308a7f02f0495315e280936f6bacd170e44
Parents: 5cfb6e6
Author: Alexey Kuznetsov <ak...@apache.org>
Authored: Thu Sep 17 17:17:05 2015 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Thu Sep 17 17:17:05 2015 +0700
----------------------------------------------------------------------
.../scala/org/apache/ignite/visor/commands/VisorConsole.scala | 6 ++++++
1 file changed, 6 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/b8c0b308/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
----------------------------------------------------------------------
diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
index 2abe8a7..6d91b05 100644
--- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
+++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
@@ -105,6 +105,12 @@ class VisorConsole {
val batchCommand = argValue("e", argLst)
cfgFile.foreach(cfg => {
+ if (cfg.trim.isEmpty) {
+ visor.warn("Expected path to configuration after \"-cfg\" option.")
+
+ visor.quit()
+ }
+
if (batchFile.isDefined || batchCommand.isDefined) {
visor.warn("Options can't contains both -cfg and one of -b or -e options.")
[25/32] ignite git commit: Merge remote-tracking branch
'origin/ignite-1.4' into ignite-1.4
Posted by ra...@apache.org.
Merge remote-tracking branch 'origin/ignite-1.4' into ignite-1.4
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/7ccd0b31
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/7ccd0b31
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/7ccd0b31
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 7ccd0b315192657f5fb2cde565ae4a12e4764b40
Parents: 90cccc9 d08ecf1
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Fri Sep 18 16:56:38 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Fri Sep 18 16:56:38 2015 +0300
----------------------------------------------------------------------
modules/yardstick/config/ignite-base-config.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
[04/32] ignite git commit: IGFS: Reverted back some test changes.
Posted by ra...@apache.org.
IGFS: Reverted back some test changes.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/126e68be
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/126e68be
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/126e68be
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 126e68bee16d7f31a7c1ae6810fe40d5ea3f8a34
Parents: ae1e1dc
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Sep 15 17:03:00 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Sep 15 17:03:00 2015 +0300
----------------------------------------------------------------------
.../ignite/internal/processors/igfs/IgfsAbstractSelfTest.java | 7 ++++---
.../internal/processors/igfs/IgfsDualAbstractSelfTest.java | 6 +++---
2 files changed, 7 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/126e68be/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
index 076c269..3e63cdc 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
@@ -2295,13 +2295,13 @@ public abstract class IgfsAbstractSelfTest extends IgfsCommonAbstractTest {
/**
* Create the file in the given IGFS and write provided data chunks to it.
*
- * @param uni File system adapter.
* @param file File.
+ * @param overwrite Overwrite flag.
* @param chunks Data chunks.
* @throws IOException In case of IO exception.
*/
- protected static void createFile(UniversalFileSystemAdapter uni, IgfsPath file, @Nullable byte[]... chunks)
- throws IOException {
+ protected static void createFile(UniversalFileSystemAdapter uni, IgfsPath file, boolean overwrite,
+ @Nullable byte[]... chunks) throws IOException {
OutputStream os = null;
try {
@@ -2313,6 +2313,7 @@ public abstract class IgfsAbstractSelfTest extends IgfsCommonAbstractTest {
U.closeQuiet(os);
IgfsEx igfsEx = uni.getAdapter(IgfsEx.class);
+
if (igfsEx != null)
awaitFileClose(igfsEx.asSecondary(), file);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/126e68be/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java
index 683054b..9527459 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java
@@ -1129,7 +1129,7 @@ public abstract class IgfsDualAbstractSelfTest extends IgfsAbstractSelfTest {
create(igfsSecondary, paths(DIR, SUBDIR), null);
create(igfs, null, null);
- createFile(igfsSecondary, FILE, chunk);
+ createFile(igfsSecondary, FILE, true, chunk);
checkFileContent(igfs, FILE, chunk);
}
@@ -1348,7 +1348,7 @@ public abstract class IgfsDualAbstractSelfTest extends IgfsAbstractSelfTest {
igfsSecondaryFileSystem.update(SUBDIR, props);
- createFile(igfsSecondary, FILE, /*BLOCK_SIZE,*/ chunk);
+ createFile(igfsSecondary, FILE, true, /*BLOCK_SIZE,*/ chunk);
appendFile(igfs, FILE, chunk);
@@ -1375,7 +1375,7 @@ public abstract class IgfsDualAbstractSelfTest extends IgfsAbstractSelfTest {
igfsSecondaryFileSystem.update(DIR, propsDir);
igfsSecondaryFileSystem.update(SUBDIR, propsSubDir);
- createFile(igfsSecondary, FILE, /*BLOCK_SIZE,*/ chunk);
+ createFile(igfsSecondary, FILE, true, /*BLOCK_SIZE,*/ chunk);
appendFile(igfs, FILE, chunk);
[24/32] ignite git commit: Removed "incubator" paths from
IgniteManager.
Posted by ra...@apache.org.
Removed "incubator" paths from IgniteManager.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/90cccc92
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/90cccc92
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/90cccc92
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 90cccc92a078d035eeb5ada47be8287901eaff3d
Parents: d64fc9d
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Fri Sep 18 16:56:18 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Fri Sep 18 16:56:18 2015 +0300
----------------------------------------------------------------------
.../src/main/dotnet/Apache.Ignite.Core/Impl/IgniteManager.cs | 2 --
1 file changed, 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/90cccc92/modules/platform/src/main/dotnet/Apache.Ignite.Core/Impl/IgniteManager.cs
----------------------------------------------------------------------
diff --git a/modules/platform/src/main/dotnet/Apache.Ignite.Core/Impl/IgniteManager.cs b/modules/platform/src/main/dotnet/Apache.Ignite.Core/Impl/IgniteManager.cs
index 0edaeab..8fd8825 100644
--- a/modules/platform/src/main/dotnet/Apache.Ignite.Core/Impl/IgniteManager.cs
+++ b/modules/platform/src/main/dotnet/Apache.Ignite.Core/Impl/IgniteManager.cs
@@ -351,8 +351,6 @@ namespace Apache.Ignite.Core.Impl
{
AppendTestClasses(ggHome + "\\examples", cpStr);
AppendTestClasses(ggHome + "\\modules", cpStr);
- AppendTestClasses(ggHome + "\\..\\incubator-ignite\\examples", cpStr);
- AppendTestClasses(ggHome + "\\..\\incubator-ignite\\modules", cpStr);
}
string ggLibs = ggHome + "\\libs";
[07/32] ignite git commit: added near eviction policy for yardstick
Posted by ra...@apache.org.
added near eviction policy for yardstick
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/8e252047
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/8e252047
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/8e252047
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 8e2520474fdf6e073a54e5d8a23b32fb266d5b4c
Parents: 6bd0f8f
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Tue Sep 15 17:09:23 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Tue Sep 15 17:09:23 2015 +0300
----------------------------------------------------------------------
.../yardstick/IgniteBenchmarkArguments.java | 24 ++++++++++----------
.../org/apache/ignite/yardstick/IgniteNode.java | 12 +++++++---
2 files changed, 21 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/8e252047/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteBenchmarkArguments.java
----------------------------------------------------------------------
diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteBenchmarkArguments.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteBenchmarkArguments.java
index 9710443..6e9059d 100644
--- a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteBenchmarkArguments.java
+++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteBenchmarkArguments.java
@@ -53,6 +53,10 @@ public class IgniteBenchmarkArguments {
private boolean nearCacheFlag = false;
/** */
+ @Parameter(names = {"-ncs", "--nearCacheSize"}, description = "Near cache size")
+ private int nearCacheSize;
+
+ /** */
@Parameter(names = {"-wom", "--writeOrderMode"}, description = "Write ordering mode")
private CacheAtomicWriteOrderMode orderMode;
@@ -81,10 +85,6 @@ public class IgniteBenchmarkArguments {
private String restTcpHost;
/** */
- @Parameter(names = {"-ss", "--syncSend"}, description = "Synchronous send")
- private boolean syncSnd;
-
- /** */
@Parameter(names = {"-r", "--range"}, description = "Key range")
private int range = 1_000_000;
@@ -162,6 +162,13 @@ public class IgniteBenchmarkArguments {
}
/**
+ * @return Near cache size ({@code 0} for unlimited).
+ */
+ public int getNearCacheSize() {
+ return nearCacheSize;
+ }
+
+ /**
* @return Synchronization.
*/
public CacheWriteSynchronizationMode syncMode() {
@@ -211,13 +218,6 @@ public class IgniteBenchmarkArguments {
}
/**
- * @return {@code True} if sending is synchronous.
- */
- public boolean isSyncSend() {
- return syncSnd;
- }
-
- /**
* @return Key range, from {@code 0} to this number.
*/
public int range() {
@@ -278,4 +278,4 @@ public class IgniteBenchmarkArguments {
@Override public String toString() {
return GridToStringBuilder.toString(IgniteBenchmarkArguments.class, this);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8e252047/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteNode.java
----------------------------------------------------------------------
diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteNode.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteNode.java
index 8c3bc45..ea3bd07 100644
--- a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteNode.java
+++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/IgniteNode.java
@@ -86,8 +86,14 @@ public class IgniteNode implements BenchmarkServer {
if (cl)
c.setClientMode(true);
- if (args.isNearCache())
- cc.setNearConfiguration(new NearCacheConfiguration());
+ if (args.isNearCache()) {
+ NearCacheConfiguration nearCfg = new NearCacheConfiguration();
+
+ if (args.getNearCacheSize() != 0)
+ nearCfg.setNearEvictionPolicy(new LruEvictionPolicy(args.getNearCacheSize()));
+
+ cc.setNearConfiguration(nearCfg);
+ }
cc.setWriteSynchronizationMode(args.syncMode());
@@ -204,4 +210,4 @@ public class IgniteNode implements BenchmarkServer {
public Ignite ignite() {
return ignite;
}
-}
\ No newline at end of file
+}
[30/32] ignite git commit: IGNITE-1370 Refactor StreamTupleExtractor
API for 0..1 and 0..n extraction.
Posted by ra...@apache.org.
IGNITE-1370 Refactor StreamTupleExtractor API for 0..1 and 0..n extraction.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/421a5234
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/421a5234
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/421a5234
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 421a5234b5a7e56e36952a4c1976b3118310073e
Parents: eae4df1 b80b171
Author: Raul Kripalani <ra...@apache.org>
Authored: Mon Sep 21 16:28:12 2015 +0100
Committer: Raul Kripalani <ra...@apache.org>
Committed: Mon Sep 21 16:28:12 2015 +0100
----------------------------------------------------------------------
.../socket/WordsSocketStreamerServer.java | 5 +-
.../org/apache/ignite/stream/StreamAdapter.java | 104 +++++++++++++++--
.../stream/StreamMultipleTupleExtractor.java | 38 +++++++
.../stream/StreamSingleTupleExtractor.java | 40 +++++++
.../ignite/stream/StreamTupleExtractor.java | 20 ++--
.../ignite/stream/socket/SocketStreamer.java | 3 +-
.../stream/socket/SocketStreamerSelfTest.java | 112 ++++++++++++++-----
7 files changed, 270 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
[22/32] ignite git commit: Merge remote-tracking branch
'origin/ignite-1.4' into ignite-1.4
Posted by ra...@apache.org.
Merge remote-tracking branch 'origin/ignite-1.4' into ignite-1.4
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d08ecf11
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d08ecf11
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d08ecf11
Branch: refs/heads/feature/ignite-535-mqtt
Commit: d08ecf11704b9b4b56ea81f9f03d10311ff9d27c
Parents: 4c101dc d64fc9d
Author: sboikov <sb...@gridgain.com>
Authored: Fri Sep 18 14:11:41 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Sep 18 14:11:41 2015 +0300
----------------------------------------------------------------------
.../cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
[03/32] ignite git commit: Merge remote-tracking branch
'origin/ignite-1.4' into ignite-1.4
Posted by ra...@apache.org.
Merge remote-tracking branch 'origin/ignite-1.4' into ignite-1.4
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/6bd0f8f4
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/6bd0f8f4
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/6bd0f8f4
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 6bd0f8f405268214c62e2289a0081a238053fc11
Parents: 72e270c dcd8b42
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Tue Sep 15 16:55:19 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Tue Sep 15 16:55:19 2015 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheProcessor.java | 10 +-
.../dht/GridClientPartitionTopology.java | 104 +++++++++++--------
.../dht/GridDhtPartitionTopology.java | 4 +-
.../dht/GridDhtPartitionTopologyImpl.java | 7 +-
.../cache/transactions/IgniteTxHandler.java | 2 +-
.../ignite/internal/util/IgniteUtils.java | 16 +++
.../processors/igfs/IgfsAbstractSelfTest.java | 3 +-
.../testsuites/IgniteHadoopTestSuite.java | 6 +-
...CacheScanPartitionQueryFallbackSelfTest.java | 54 ++++------
9 files changed, 114 insertions(+), 92 deletions(-)
----------------------------------------------------------------------
[15/32] ignite git commit: ignite-1.4 Fixed Visor cmd options.
Posted by ra...@apache.org.
ignite-1.4 Fixed Visor cmd options.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/5cfb6e68
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/5cfb6e68
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/5cfb6e68
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 5cfb6e6878dea2fa78d7593766035a5b535763a1
Parents: cd6a1d5
Author: Alexey Kuznetsov <ak...@apache.org>
Authored: Thu Sep 17 16:24:25 2015 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Thu Sep 17 16:24:25 2015 +0700
----------------------------------------------------------------------
.../main/scala/org/apache/ignite/visor/commands/VisorConsole.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/5cfb6e68/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
----------------------------------------------------------------------
diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
index 627f795..2abe8a7 100644
--- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
+++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
@@ -91,7 +91,7 @@ class VisorConsole {
println(" -np - no pause on exit (pause by default).")
println(" -cfg=<path> - connect with specified configuration.")
println(" -b=<path> - batch mode with file.")
- println(" -e==cmd1;cmd2;... - batch mode with commands.")
+ println(" -e=cmd1;cmd2;... - batch mode with commands.")
visor.quit()
}
[14/32] ignite git commit: Merge branch 'ignite-1.4'
Posted by ra...@apache.org.
Merge branch 'ignite-1.4'
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c311c3c9
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c311c3c9
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c311c3c9
Branch: refs/heads/feature/ignite-535-mqtt
Commit: c311c3c9e53398dd510839ea02efc541054461fe
Parents: 6e48c9c cd6a1d5
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Thu Sep 17 10:44:39 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Thu Sep 17 10:44:39 2015 +0300
----------------------------------------------------------------------
.../configuration/NearCacheConfiguration.java | 18 +-
.../distributed/dht/GridDhtLocalPartition.java | 26 +-
.../query/GridCacheDistributedQueryFuture.java | 27 +-
.../cache/query/GridCacheLocalQueryFuture.java | 5 +
.../cache/query/GridCacheQueryAdapter.java | 170 ++++++++-----
.../query/GridCacheQueryFutureAdapter.java | 11 +-
.../cache/query/GridCacheQueryManager.java | 30 ++-
.../distributed/CacheAffEarlySelfTest.java | 245 +++++++++++++++++++
.../GridCacheSwapScanQueryAbstractSelfTest.java | 118 ++++-----
.../processors/igfs/IgfsAbstractSelfTest.java | 5 +-
...CacheScanPartitionQueryFallbackSelfTest.java | 244 +++++-------------
.../ignite/visor/commands/VisorConsole.scala | 31 ++-
.../yardstick/IgniteBenchmarkArguments.java | 24 +-
.../org/apache/ignite/yardstick/IgniteNode.java | 12 +-
14 files changed, 610 insertions(+), 356 deletions(-)
----------------------------------------------------------------------
[12/32] ignite git commit: Basic suite tests uncommented
Posted by ra...@apache.org.
Basic suite tests uncommented
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/6187241e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/6187241e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/6187241e
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 6187241ecf68a9541172c98e85e86cecdc0fe7e9
Parents: e75917d
Author: agura <ag...@gridgain.com>
Authored: Wed Sep 16 18:36:34 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Wed Sep 16 18:36:34 2015 +0300
----------------------------------------------------------------------
.../java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java | 5 -----
1 file changed, 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/6187241e/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
index 390807c..1a9913f 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
@@ -36,7 +36,6 @@ import org.apache.ignite.internal.processors.affinity.GridAffinityProcessorRende
import org.apache.ignite.internal.processors.cache.GridProjectionForCachesOnDaemonNodeSelfTest;
import org.apache.ignite.internal.processors.cache.IgniteDaemonNodeMarshallerCacheTest;
import org.apache.ignite.internal.processors.cache.OffHeapTieredTransactionSelfTest;
-import org.apache.ignite.internal.processors.cache.distributed.CacheAffEarlySelfTest;
import org.apache.ignite.internal.processors.closure.GridClosureProcessorSelfTest;
import org.apache.ignite.internal.processors.continuous.GridEventConsumeSelfTest;
import org.apache.ignite.internal.processors.continuous.GridMessageListenSelfTest;
@@ -69,9 +68,6 @@ public class IgniteBasicTestSuite extends TestSuite {
public static TestSuite suite(Set<Class> ignoredTests) throws Exception {
TestSuite suite = new TestSuite("Ignite Basic Test Suite");
- suite.addTest(new TestSuite(CacheAffEarlySelfTest.class));
-
-/*
suite.addTest(IgniteLangSelfTestSuite.suite());
suite.addTest(IgniteUtilSelfTestSuite.suite(ignoredTests));
suite.addTest(IgniteMarshallerSelfTestSuite.suite(ignoredTests));
@@ -114,7 +110,6 @@ public class IgniteBasicTestSuite extends TestSuite {
suite.addTestSuite(OffHeapTieredTransactionSelfTest.class);
suite.addTestSuite(IgniteSlowClientDetectionSelfTest.class);
suite.addTestSuite(IgniteDaemonNodeMarshallerCacheTest.class);
-*/
return suite;
}
[27/32] ignite git commit: Merge remote-tracking branch
'origin/ignite-1.4' into ignite-1.4
Posted by ra...@apache.org.
Merge remote-tracking branch 'origin/ignite-1.4' into ignite-1.4
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/8b94494a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/8b94494a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/8b94494a
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 8b94494a4817780961c203e0b2b52715baa3b6da
Parents: 84a2300 7ccd0b3
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Fri Sep 18 23:16:10 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Fri Sep 18 23:16:10 2015 +0300
----------------------------------------------------------------------
.../src/main/dotnet/Apache.Ignite.Core/Impl/IgniteManager.cs | 2 --
modules/yardstick/config/ignite-base-config.xml | 2 +-
2 files changed, 1 insertion(+), 3 deletions(-)
----------------------------------------------------------------------
[31/32] ignite git commit: Merge branch 'master' into
feature/ignite-535-mqtt
Posted by ra...@apache.org.
Merge branch 'master' into feature/ignite-535-mqtt
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f03f3a3b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f03f3a3b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f03f3a3b
Branch: refs/heads/feature/ignite-535-mqtt
Commit: f03f3a3b48fa105f318e9493440671188770f4ef
Parents: 53683e2 421a523
Author: Raul Kripalani <ra...@apache.org>
Authored: Mon Sep 21 16:36:53 2015 +0100
Committer: Raul Kripalani <ra...@apache.org>
Committed: Mon Sep 21 16:36:53 2015 +0100
----------------------------------------------------------------------
.../org/apache/ignite/IgniteAtomicLong.java | 15 +-
.../apache/ignite/IgniteAtomicReference.java | 9 +-
.../org/apache/ignite/IgniteAtomicSequence.java | 9 +-
.../org/apache/ignite/IgniteAtomicStamped.java | 13 +-
.../configuration/NearCacheConfiguration.java | 18 +-
.../apache/ignite/internal/IgniteKernal.java | 7 -
.../processors/cache/GridCacheContext.java | 6 +-
.../cache/GridCacheEvictionManager.java | 6 +-
.../cache/GridCacheEvictionResponse.java | 2 +-
.../processors/cache/GridCacheIoManager.java | 47 ++--
.../processors/cache/GridCacheMessage.java | 7 +
.../processors/cache/GridCacheMvccManager.java | 34 ++-
.../GridCachePartitionExchangeManager.java | 41 +++-
.../processors/cache/GridCacheProcessor.java | 28 ++-
.../GridDistributedLockResponse.java | 6 +-
.../GridDistributedTxPrepareResponse.java | 6 +-
.../distributed/dht/GridDhtLocalPartition.java | 26 +-
.../distributed/dht/GridDhtTopologyFuture.java | 6 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 2 +-
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 2 +-
.../dht/atomic/GridDhtAtomicUpdateResponse.java | 12 +-
.../dht/atomic/GridNearAtomicUpdateFuture.java | 16 +-
.../dht/atomic/GridNearAtomicUpdateRequest.java | 2 +
.../atomic/GridNearAtomicUpdateResponse.java | 11 +-
.../colocated/GridDhtColocatedLockFuture.java | 44 +++-
.../dht/preloader/GridDhtForceKeysFuture.java | 2 +-
.../dht/preloader/GridDhtForceKeysResponse.java | 6 +-
.../GridDhtPartitionsExchangeFuture.java | 19 +-
.../distributed/near/GridNearGetResponse.java | 6 +-
.../distributed/near/GridNearLockFuture.java | 26 +-
.../near/GridNearOptimisticTxPrepareFuture.java | 20 +-
.../near/GridNearTxFinishResponse.java | 6 +-
.../query/GridCacheDistributedQueryFuture.java | 27 +-
.../cache/query/GridCacheLocalQueryFuture.java | 5 +
.../cache/query/GridCacheQueryAdapter.java | 170 ++++++++-----
.../query/GridCacheQueryFutureAdapter.java | 11 +-
.../cache/query/GridCacheQueryManager.java | 30 ++-
.../cache/query/GridCacheQueryResponse.java | 6 +-
.../continuous/CacheContinuousQueryHandler.java | 12 +-
.../transactions/IgniteTxLocalAdapter.java | 4 +-
.../ignite/internal/util/GridSpinBusyLock.java | 10 +
.../ignite/spi/discovery/tcp/ServerImpl.java | 12 +-
.../IgniteCacheEntryProcessorNodeJoinTest.java | 24 +-
.../distributed/CacheAffEarlySelfTest.java | 245 +++++++++++++++++++
.../GridCacheSwapScanQueryAbstractSelfTest.java | 118 ++++-----
.../processors/igfs/IgfsAbstractSelfTest.java | 5 +-
.../loadtests/hashmap/GridCacheTestContext.java | 4 +-
.../ignite/testframework/GridTestUtils.java | 14 +-
.../cache/CacheIndexStreamerTest.java | 137 +++++++++++
...CacheScanPartitionQueryFallbackSelfTest.java | 244 +++++-------------
.../IgniteCacheQueryNodeRestartSelfTest2.java | 2 -
.../Apache.Ignite.Core/Impl/IgniteManager.cs | 2 -
.../ignite/visor/commands/VisorConsole.scala | 37 ++-
.../config/benchmark-put-indexed-val.properties | 2 +-
modules/yardstick/config/ignite-base-config.xml | 2 +-
.../yardstick/IgniteBenchmarkArguments.java | 24 +-
.../org/apache/ignite/yardstick/IgniteNode.java | 12 +-
57 files changed, 1084 insertions(+), 535 deletions(-)
----------------------------------------------------------------------
[06/32] ignite git commit: IGFS: Reverted back some test changes.
Posted by ra...@apache.org.
IGFS: Reverted back some test changes.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/7a71b206
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/7a71b206
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/7a71b206
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 7a71b206227452706bbee4e9b64b2294eb39750d
Parents: cb7d972
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Sep 15 17:08:16 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Sep 15 17:08:16 2015 +0300
----------------------------------------------------------------------
.../internal/processors/igfs/IgfsAbstractSelfTest.java | 8 +++++---
.../internal/processors/igfs/IgfsDualAbstractSelfTest.java | 6 +++---
2 files changed, 8 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/7a71b206/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
index 3e63cdc..0a1e626 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
@@ -2296,12 +2296,11 @@ public abstract class IgfsAbstractSelfTest extends IgfsCommonAbstractTest {
* Create the file in the given IGFS and write provided data chunks to it.
*
* @param file File.
- * @param overwrite Overwrite flag.
* @param chunks Data chunks.
* @throws IOException In case of IO exception.
*/
- protected static void createFile(UniversalFileSystemAdapter uni, IgfsPath file, boolean overwrite,
- @Nullable byte[]... chunks) throws IOException {
+ protected static void createFile(UniversalFileSystemAdapter uni, IgfsPath file, @Nullable byte[]... chunks)
+ throws IOException {
OutputStream os = null;
try {
@@ -2689,5 +2688,8 @@ public abstract class IgfsAbstractSelfTest extends IgfsCommonAbstractTest {
if (igfsEx != null)
clear(igfsEx);
+
+ // Clear the filesystem.
+ uni.format();
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/7a71b206/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java
index 9527459..683054b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsDualAbstractSelfTest.java
@@ -1129,7 +1129,7 @@ public abstract class IgfsDualAbstractSelfTest extends IgfsAbstractSelfTest {
create(igfsSecondary, paths(DIR, SUBDIR), null);
create(igfs, null, null);
- createFile(igfsSecondary, FILE, true, chunk);
+ createFile(igfsSecondary, FILE, chunk);
checkFileContent(igfs, FILE, chunk);
}
@@ -1348,7 +1348,7 @@ public abstract class IgfsDualAbstractSelfTest extends IgfsAbstractSelfTest {
igfsSecondaryFileSystem.update(SUBDIR, props);
- createFile(igfsSecondary, FILE, true, /*BLOCK_SIZE,*/ chunk);
+ createFile(igfsSecondary, FILE, /*BLOCK_SIZE,*/ chunk);
appendFile(igfs, FILE, chunk);
@@ -1375,7 +1375,7 @@ public abstract class IgfsDualAbstractSelfTest extends IgfsAbstractSelfTest {
igfsSecondaryFileSystem.update(DIR, propsDir);
igfsSecondaryFileSystem.update(SUBDIR, propsSubDir);
- createFile(igfsSecondary, FILE, true, /*BLOCK_SIZE,*/ chunk);
+ createFile(igfsSecondary, FILE, /*BLOCK_SIZE,*/ chunk);
appendFile(igfs, FILE, chunk);
[10/32] ignite git commit: IGNITE-1239 - Fixed scan query failover on
changing topology.
Posted by ra...@apache.org.
IGNITE-1239 - Fixed scan query failover on changing topology.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/5f5220af
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/5f5220af
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/5f5220af
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 5f5220af0449fd3957d011fcef954b19da852e18
Parents: d928ef4
Author: sboikov <sb...@gridgain.com>
Authored: Wed Sep 16 18:06:40 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Sep 16 18:06:40 2015 +0300
----------------------------------------------------------------------
.../query/GridCacheDistributedQueryFuture.java | 27 ++-
.../cache/query/GridCacheLocalQueryFuture.java | 5 +
.../cache/query/GridCacheQueryAdapter.java | 170 ++++++++------
.../query/GridCacheQueryFutureAdapter.java | 11 +-
.../cache/query/GridCacheQueryManager.java | 30 ++-
.../GridCacheSwapScanQueryAbstractSelfTest.java | 118 +++++-----
...CacheScanPartitionQueryFallbackSelfTest.java | 224 ++++++-------------
7 files changed, 287 insertions(+), 298 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/5f5220af/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java
index 1d547c5..e745e30 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java
@@ -27,6 +27,7 @@ import java.util.concurrent.CountDownLatch;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.P1;
@@ -142,9 +143,24 @@ public class GridCacheDistributedQueryFuture<K, V, R> extends GridCacheQueryFutu
}
if (callOnPage)
- // We consider node departure as a reception of last empty
- // page from this node.
- onPage(nodeId, Collections.emptyList(), null, true);
+ onPage(nodeId, Collections.emptyList(),
+ new ClusterTopologyCheckedException("Remote node has left topology: " + nodeId), true);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void awaitFirstPage() throws IgniteCheckedException {
+ try {
+ firstPageLatch.await();
+
+ if (isDone() && error() != null)
+ // Throw the exception if future failed.
+ get();
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+
+ throw new IgniteInterruptedCheckedException(e);
+ }
}
/** {@inheritDoc} */
@@ -229,9 +245,12 @@ public class GridCacheDistributedQueryFuture<K, V, R> extends GridCacheQueryFutu
/** {@inheritDoc} */
@Override public boolean onDone(Collection<R> res, Throwable err) {
+ boolean done = super.onDone(res, err);
+
+ // Must release the lath after onDone() in order for a waiting thread to see an exception, if any.
firstPageLatch.countDown();
- return super.onDone(res, err);
+ return done;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/5f5220af/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheLocalQueryFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheLocalQueryFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheLocalQueryFuture.java
index 46af18a..248dfa8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheLocalQueryFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheLocalQueryFuture.java
@@ -77,6 +77,11 @@ public class GridCacheLocalQueryFuture<K, V, R> extends GridCacheQueryFutureAdap
// No-op.
}
+ /** {@inheritDoc} */
+ @Override public void awaitFirstPage() throws IgniteCheckedException {
+ get();
+ }
+
/** */
private class LocalQueryRunnable implements GridPlainRunnable {
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/5f5220af/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java
index 3ac5746..855e239 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryAdapter.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.query;
import java.util.Collection;
import java.util.Collections;
+import java.util.Deque;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Queue;
@@ -34,6 +35,7 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterGroupEmptyCheckedException;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtUnreservedPartitionException;
@@ -41,14 +43,13 @@ import org.apache.ignite.internal.processors.query.GridQueryProcessor;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.P1;
+import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.A;
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.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteClosure;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteReducer;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.jetbrains.annotations.Nullable;
@@ -63,13 +64,6 @@ import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryTy
* Query adapter.
*/
public class GridCacheQueryAdapter<T> implements CacheQuery<T> {
- /** Is local node predicate. */
- private static final IgnitePredicate<ClusterNode> IS_LOC_NODE = new IgnitePredicate<ClusterNode>() {
- @Override public boolean apply(ClusterNode n) {
- return n.isLocal();
- }
- };
-
/** */
private final GridCacheContext<?, ?> cctx;
@@ -446,7 +440,7 @@ public class GridCacheQueryAdapter<T> implements CacheQuery<T> {
cctx.checkSecurity(SecurityPermission.CACHE_READ);
- if (nodes.isEmpty())
+ if (nodes.isEmpty() && (type != SCAN || part == null))
return new GridCacheQueryErrorFuture<>(cctx.kernalContext(), new ClusterGroupEmptyCheckedException());
if (log.isDebugEnabled())
@@ -477,8 +471,8 @@ public class GridCacheQueryAdapter<T> implements CacheQuery<T> {
if (type == SQL_FIELDS || type == SPI)
return (CacheQueryFuture<R>)(loc ? qryMgr.queryFieldsLocal(bean) :
qryMgr.queryFieldsDistributed(bean, nodes));
- else if (type == SCAN && part != null && nodes.size() > 1)
- return new CacheQueryFallbackFuture<>(nodes, part, bean, qryMgr, cctx);
+ else if (type == SCAN && part != null && !cctx.isLocal())
+ return new CacheQueryFallbackFuture<>(part, bean, qryMgr, cctx);
else
return (CacheQueryFuture<R>)(loc ? qryMgr.queryLocal(bean) : qryMgr.queryDistributed(bean, nodes));
}
@@ -581,37 +575,48 @@ public class GridCacheQueryAdapter<T> implements CacheQuery<T> {
/** Partition. */
private final int part;
+ /** Flag indicating that a first item has been returned to a user. */
+ private boolean firstItemReturned;
+
/**
- * @param nodes Backups.
* @param part Partition.
* @param bean Bean.
* @param qryMgr Query manager.
* @param cctx Cache context.
*/
- public CacheQueryFallbackFuture(Collection<ClusterNode> nodes, int part, GridCacheQueryBean bean,
+ private CacheQueryFallbackFuture(int part, GridCacheQueryBean bean,
GridCacheQueryManager qryMgr, GridCacheContext cctx) {
- this.nodes = fallbacks(nodes);
this.bean = bean;
this.qryMgr = qryMgr;
this.cctx = cctx;
this.part = part;
+ nodes = fallbacks(cctx.discovery().topologyVersionEx());
+
init();
}
/**
- * @param nodes Nodes.
+ * @param topVer Topology version.
* @return Nodes for query execution.
*/
- private Queue<ClusterNode> fallbacks(Collection<ClusterNode> nodes) {
- Queue<ClusterNode> fallbacks = new LinkedList<>();
+ private Queue<ClusterNode> fallbacks(AffinityTopologyVersion topVer) {
+ Deque<ClusterNode> fallbacks = new LinkedList<>();
+ Collection<ClusterNode> owners = new HashSet<>();
- ClusterNode node = F.first(F.view(nodes, IS_LOC_NODE));
+ for (ClusterNode node : cctx.topology().owners(part, topVer)) {
+ if (node.isLocal())
+ fallbacks.addFirst(node);
+ else
+ fallbacks.add(node);
- if (node != null)
- fallbacks.add(node);
+ owners.add(node);
+ }
- fallbacks.addAll(node != null ? F.view(nodes, F.not(IS_LOC_NODE)) : nodes);
+ for (ClusterNode node : cctx.topology().moving(part)) {
+ if (!owners.contains(node))
+ fallbacks.add(node);
+ }
return fallbacks;
}
@@ -623,53 +628,9 @@ public class GridCacheQueryAdapter<T> implements CacheQuery<T> {
private void init() {
final ClusterNode node = nodes.poll();
- GridCacheQueryFutureAdapter<?, ?, R> fut0 = (GridCacheQueryFutureAdapter<?, ?, R>)(node.isLocal() ?
+ fut = (GridCacheQueryFutureAdapter<?, ?, R>)(node.isLocal() ?
qryMgr.queryLocal(bean) :
qryMgr.queryDistributed(bean, Collections.singleton(node)));
-
- fut0.listen(new IgniteInClosure<IgniteInternalFuture<Collection<R>>>() {
- @Override public void apply(IgniteInternalFuture<Collection<R>> fut) {
- try {
- onDone(fut.get());
- }
- catch (IgniteClientDisconnectedCheckedException e) {
- onDone(e);
- }
- catch (IgniteCheckedException e) {
- if (e.hasCause(GridDhtUnreservedPartitionException.class)) {
- unreservedTopVer = ((GridDhtUnreservedPartitionException)e.getCause()).topologyVersion();
-
- assert unreservedTopVer != null;
- }
-
- if (F.isEmpty(nodes)) {
- final AffinityTopologyVersion topVer = unreservedTopVer;
-
- if (topVer != null && --unreservedNodesRetryCnt > 0) {
- cctx.affinity().affinityReadyFuture(topVer).listen(
- new IgniteInClosure<IgniteInternalFuture<AffinityTopologyVersion>>() {
- @Override public void apply(
- IgniteInternalFuture<AffinityTopologyVersion> future) {
-
- nodes = fallbacks(cctx.topology().owners(part, topVer));
-
- // Race is impossible here because query retries are executed one by one.
- unreservedTopVer = null;
-
- init();
- }
- });
- }
- else
- onDone(e);
- }
- else
- init();
- }
- }
- });
-
- fut = fut0;
}
/** {@inheritDoc} */
@@ -683,8 +644,81 @@ public class GridCacheQueryAdapter<T> implements CacheQuery<T> {
}
/** {@inheritDoc} */
+ @Override public Collection<R> get() throws IgniteCheckedException {
+ assert false;
+
+ return super.get();
+ }
+
+ /** {@inheritDoc} */
@Override public R next() {
- return fut.next();
+ if (firstItemReturned)
+ return fut.next();
+
+ while (true) {
+ try {
+ fut.awaitFirstPage();
+
+ firstItemReturned = true;
+
+ return fut.next();
+ }
+ catch (IgniteClientDisconnectedCheckedException e) {
+ throw CU.convertToCacheException(e);
+ }
+ catch (IgniteCheckedException e) {
+ retryIfPossible(e);
+ }
+ }
+ }
+
+ /**
+ * @param e Exception for query run.
+ */
+ private void retryIfPossible(IgniteCheckedException e) {
+ try {
+ IgniteInternalFuture<?> retryFut;
+
+ if (e.hasCause(GridDhtUnreservedPartitionException.class)) {
+ AffinityTopologyVersion waitVer = ((GridDhtUnreservedPartitionException)e.getCause()).topologyVersion();
+
+ assert waitVer != null;
+
+ retryFut = cctx.affinity().affinityReadyFuture(waitVer);
+ }
+ else if (e.hasCause(ClusterTopologyCheckedException.class)) {
+ ClusterTopologyCheckedException topEx = X.cause(e, ClusterTopologyCheckedException.class);
+
+ retryFut = topEx.retryReadyFuture();
+ }
+ else if (e.hasCause(ClusterGroupEmptyCheckedException.class)) {
+ ClusterGroupEmptyCheckedException ex = X.cause(e, ClusterGroupEmptyCheckedException.class);
+
+ retryFut = ex.retryReadyFuture();
+ }
+ else
+ throw CU.convertToCacheException(e);
+
+ if (F.isEmpty(nodes)) {
+ if (--unreservedNodesRetryCnt > 0) {
+ if (retryFut != null)
+ retryFut.get();
+
+ nodes = fallbacks(unreservedTopVer == null ? cctx.discovery().topologyVersionEx() : unreservedTopVer);
+
+ unreservedTopVer = null;
+
+ init();
+ }
+ else
+ throw CU.convertToCacheException(e);
+ }
+ else
+ init();
+ }
+ catch (IgniteCheckedException ex) {
+ throw CU.convertToCacheException(ex);
+ }
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/5f5220af/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java
index ad9ee39..2a4fbda 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java
@@ -183,6 +183,13 @@ public abstract class GridCacheQueryFutureAdapter<K, V, R> extends GridFutureAda
}
/**
+ * Waits for the first page to be received from remote node(s), if any.
+ *
+ * @throws IgniteCheckedException If query execution failed with an error.
+ */
+ public abstract void awaitFirstPage() throws IgniteCheckedException;
+
+ /**
* Returns next page for the query.
*
* @return Next page or {@code null} if no more pages available.
@@ -380,13 +387,13 @@ public abstract class GridCacheQueryFutureAdapter<K, V, R> extends GridFutureAda
synchronized (mux) {
enqueue(Collections.emptyList());
- onPage(nodeId, true);
-
onDone(nodeId != null ?
new IgniteCheckedException("Failed to execute query on node [query=" + qry +
", nodeId=" + nodeId + "]", err) :
new IgniteCheckedException("Failed to execute query locally: " + qry, err));
+ onPage(nodeId, true);
+
mux.notifyAll();
}
else {
http://git-wip-us.apache.org/repos/asf/ignite/blob/5f5220af/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
index 1d934d8..25ace1b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
@@ -33,8 +33,9 @@ import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheMetricsImpl;
import org.apache.ignite.internal.processors.cache.CacheObject;
-import org.apache.ignite.internal.processors.cache.CachePeekModes;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
+import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.GridCacheInternal;
import org.apache.ignite.internal.processors.cache.GridCacheManagerAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheOffheapSwapEntry;
@@ -819,8 +820,12 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
final GridDhtCacheAdapter dht = cctx.isLocal() ? null : (cctx.isNear() ? cctx.near().dht() : cctx.dht());
+ final GridCacheAdapter cache = dht != null ? dht : cctx.cache();
+
final ExpiryPolicy plc = cctx.expiry();
+ final AffinityTopologyVersion topVer = cctx.affinity().affinityTopologyVersion();
+
final boolean backups = qry.includeBackups() || cctx.isReplicated();
final GridCloseableIteratorAdapter<IgniteBiTuple<K, V>> heapIt =
@@ -841,8 +846,6 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
else if (part < 0 || part >= cctx.affinity().partitions())
iter = F.emptyIterator();
else {
- AffinityTopologyVersion topVer = cctx.affinity().affinityTopologyVersion();
-
locPart = dht.topology().localPartition(part, topVer, false);
// double check for owning state
@@ -899,7 +902,15 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
V val;
try {
- val = prj.localPeek(key, CachePeekModes.ONHEAP_ONLY, expiryPlc);
+ GridCacheEntryEx entry = cache.peekEx(key);
+
+ CacheObject cacheVal =
+ entry != null ? entry.peek(true, false, false, topVer, expiryPlc) : null;
+
+ val = cacheVal != null ? (V)cacheVal.value(cctx.cacheObjectContext(), false) : null;
+ }
+ catch (GridCacheEntryRemovedException e) {
+ val = null;
}
catch (IgniteCheckedException e) {
if (log.isDebugEnabled())
@@ -1557,7 +1568,8 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
}
}
catch (Throwable e) {
- U.error(log, "Failed to run query [qry=" + qryInfo + ", node=" + cctx.nodeId() + "]", e);
+ if (!X.hasCause(e, GridDhtUnreservedPartitionException.class))
+ U.error(log, "Failed to run query [qry=" + qryInfo + ", node=" + cctx.nodeId() + "]", e);
onPageReady(loc, qryInfo, null, true, e);
@@ -1572,8 +1584,9 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
res.closeIfNotShared(recipient(qryInfo.senderId(), qryInfo.requestId()));
}
catch (IgniteCheckedException e) {
- U.error(log, "Failed to close local iterator [qry=" + qryInfo + ", node=" +
- cctx.nodeId() + "]", e);
+ if (!X.hasCause(e, GridDhtUnreservedPartitionException.class))
+ U.error(log, "Failed to close local iterator [qry=" + qryInfo + ", node=" +
+ cctx.nodeId() + "]", e);
}
}
}
@@ -1694,7 +1707,8 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
fut.get().closeIfNotShared(recipient(sndId, reqId));
}
catch (IgniteCheckedException e) {
- U.error(log, "Failed to close iterator.", e);
+ if (!X.hasCause(e, GridDhtUnreservedPartitionException.class))
+ U.error(log, "Failed to close iterator.", e);
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/5f5220af/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java
index 6f4a4ab..94c8d00 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java
@@ -19,13 +19,16 @@ package org.apache.ignite.internal.processors.cache.query;
import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
+import javax.cache.Cache;
+import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.IgniteKernal;
-import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
@@ -115,13 +118,13 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
* @throws Exception If failed.
*/
public void testQuery() throws Exception {
- checkQuery(((IgniteKernal)grid(0)).internalCache(ATOMIC_CACHE_NAME), false);
+ checkQuery(grid(0).cache(ATOMIC_CACHE_NAME), false);
- checkQuery(((IgniteKernal)grid(0)).internalCache(TRANSACTIONAL_CACHE_NAME), false);
+ checkQuery(grid(0).cache(TRANSACTIONAL_CACHE_NAME), false);
- checkQuery(((IgniteKernal)grid(0)).internalCache(ATOMIC_CACHE_NAME), true);
+ checkQuery(grid(0).cache(ATOMIC_CACHE_NAME), true);
- checkQuery(((IgniteKernal)grid(0)).internalCache(TRANSACTIONAL_CACHE_NAME), true);
+ checkQuery(grid(0).cache(TRANSACTIONAL_CACHE_NAME), true);
}
/**
@@ -130,16 +133,18 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
* @throws Exception If failed.
*/
@SuppressWarnings("unchecked")
- private void checkQuery(GridCacheAdapter cache, boolean scanPartitions) throws Exception {
+ private void checkQuery(IgniteCache cache, boolean scanPartitions) throws Exception {
final int ENTRY_CNT = 500;
Map<Integer, Map<Key, Person>> entries = new HashMap<>();
+ Affinity<Object> aff = ignite(0).affinity(cache.getName());
+
for (int i = 0; i < ENTRY_CNT; i++) {
Key key = new Key(i);
Person val = new Person("p-" + i, i);
- int part = cache.context().affinity().partition(key);
+ int part = aff.partition(key);
cache.getAndPut(key, val);
@@ -152,24 +157,26 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
}
try {
- int partitions = scanPartitions ? cache.context().affinity().partitions() : 1;
+ int partitions = scanPartitions ? aff.partitions() : 1;
for (int i = 0; i < partitions; i++) {
- CacheQuery<Map.Entry<Key, Person>> qry = cache.context().queries().createScanQuery(
- new IgniteBiPredicate<Key, Person>() {
- @Override public boolean apply(Key key, Person p) {
- assertEquals(key.id, (Integer)p.salary);
+ ScanQuery<Key, Person> qry = new ScanQuery<>(new IgniteBiPredicate<Key, Person>() {
+ @Override public boolean apply(Key key, Person p) {
+ assertEquals(key.id, (Integer)p.salary);
+
+ return key.id % 2 == 0;
+ }
+ });
- return key.id % 2 == 0;
- }
- }, (scanPartitions ? i : null), false);
+ if (scanPartitions)
+ qry.setPartition(i);
- Collection<Map.Entry<Key, Person>> res = qry.execute().get();
+ List<Cache.Entry<Key, Person>> res = cache.query(qry).getAll();
if (!scanPartitions)
assertEquals(ENTRY_CNT / 2, res.size());
- for (Map.Entry<Key, Person> e : res) {
+ for (Cache.Entry<Key, Person> e : res) {
Key k = e.getKey();
Person p = e.getValue();
@@ -183,9 +190,12 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
}
}
- qry = cache.context().queries().createScanQuery(null, (scanPartitions ? i : null), false);
+ qry = new ScanQuery<>();
+
+ if (scanPartitions)
+ qry.setPartition(i);
- res = qry.execute().get();
+ res = cache.query(qry).getAll();
if (!scanPartitions)
assertEquals(ENTRY_CNT, res.size());
@@ -204,23 +214,22 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
* @param expCnt Expected entries in query result.
* @throws Exception If failed.
*/
- private void testMultithreaded(final GridCacheAdapter cache, final int expCnt) throws Exception {
+ private void testMultithreaded(final IgniteCache cache, final int expCnt) throws Exception {
log.info("Starting multithreaded queries.");
GridTestUtils.runMultiThreaded(new Callable<Void>() {
@SuppressWarnings("unchecked")
@Override public Void call() throws Exception {
- CacheQuery<Map.Entry<Key, Person>> qry = cache.context().queries().createScanQuery(
- new IgniteBiPredicate<Key, Person>() {
- @Override public boolean apply(Key key, Person p) {
- assertEquals(key.id, (Integer)p.salary);
+ ScanQuery<Key, Person> qry = new ScanQuery<>(new IgniteBiPredicate<Key, Person>() {
+ @Override public boolean apply(Key key, Person p) {
+ assertEquals(key.id, (Integer)p.salary);
- return key.id % 2 == 0;
- }
- }, null, false);
+ return key.id % 2 == 0;
+ }
+ });
for (int i = 0; i < 250; i++) {
- Collection<Map.Entry<Key, Person>> res = qry.execute().get();
+ List<Cache.Entry<Key, Person>> res = cache.query(qry).getAll();
assertEquals(expCnt, res.size());
@@ -237,9 +246,9 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
* @throws Exception If failed.
*/
public void testQueryPrimitives() throws Exception {
- checkQueryPrimitives(((IgniteKernal)grid(0)).internalCache(ATOMIC_CACHE_NAME));
+ checkQueryPrimitives(grid(0).cache(ATOMIC_CACHE_NAME));
- checkQueryPrimitives(((IgniteKernal)grid(0)).internalCache(TRANSACTIONAL_CACHE_NAME));
+ checkQueryPrimitives(grid(0).cache(TRANSACTIONAL_CACHE_NAME));
}
/**
@@ -247,27 +256,26 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
* @throws Exception If failed.
*/
@SuppressWarnings("unchecked")
- private void checkQueryPrimitives(GridCacheAdapter cache) throws Exception {
+ private void checkQueryPrimitives(IgniteCache cache) throws Exception {
final int ENTRY_CNT = 500;
for (int i = 0; i < ENTRY_CNT; i++)
- cache.getAndPut(String.valueOf(i), (long) i);
+ cache.getAndPut(String.valueOf(i), (long)i);
try {
- CacheQuery<Map.Entry<String, Long>> qry = cache.context().queries().createScanQuery(
- new IgniteBiPredicate<String, Long>() {
- @Override public boolean apply(String key, Long val) {
- assertEquals(key, String.valueOf(val));
+ ScanQuery<String, Long> qry = new ScanQuery<>(new IgniteBiPredicate<String, Long>() {
+ @Override public boolean apply(String key, Long val) {
+ assertEquals(key, String.valueOf(val));
- return val % 2 == 0;
- }
- }, null, false);
+ return val % 2 == 0;
+ }
+ });
- Collection<Map.Entry<String, Long>> res = qry.execute().get();
+ Collection<Cache.Entry<String, Long>> res = cache.query(qry).getAll();
assertEquals(ENTRY_CNT / 2, res.size());
- for (Map.Entry<String, Long> e : res) {
+ for (Cache.Entry<String, Long> e : res) {
String key = e.getKey();
Long val = e.getValue();
@@ -276,9 +284,9 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
assertEquals(0, val % 2);
}
- qry = cache.context().queries().createScanQuery(null, null, false);
+ qry = new ScanQuery<>();
- res = qry.execute().get();
+ res = cache.query(qry).getAll();
assertEquals(ENTRY_CNT, res.size());
}
@@ -292,9 +300,9 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
* @throws Exception If failed.
*/
public void testQueryValueByteArray() throws Exception {
- checkQueryValueByteArray(((IgniteKernal)grid(0)).internalCache(ATOMIC_CACHE_NAME));
+ checkQueryValueByteArray(grid(0).cache(ATOMIC_CACHE_NAME));
- checkQueryValueByteArray(((IgniteKernal)grid(0)).internalCache(TRANSACTIONAL_CACHE_NAME));
+ checkQueryValueByteArray(grid(0).cache(TRANSACTIONAL_CACHE_NAME));
}
/**
@@ -302,27 +310,27 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
* @throws Exception If failed.
*/
@SuppressWarnings("unchecked")
- private void checkQueryValueByteArray(GridCacheAdapter cache) throws Exception {
+ private void checkQueryValueByteArray(IgniteCache cache) throws Exception {
final int ENTRY_CNT = 100;
for (int i = 0; i < ENTRY_CNT; i++)
cache.getAndPut(i, new byte[i]);
try {
- CacheQuery<Map.Entry<Integer, byte[]>> qry = cache.context().queries().createScanQuery(
+ ScanQuery<Integer, byte[]> qry = new ScanQuery<>(
new IgniteBiPredicate<Integer, byte[]>() {
@Override public boolean apply(Integer key, byte[] val) {
assertEquals(key, (Integer)val.length);
return key % 2 == 0;
}
- }, null, false);
+ });
- Collection<Map.Entry<Integer, byte[]>> res = qry.execute().get();
+ Collection<Cache.Entry<Integer, byte[]>> res = cache.query(qry).getAll();
assertEquals(ENTRY_CNT / 2, res.size());
- for (Map.Entry<Integer, byte[]> e : res) {
+ for (Cache.Entry<Integer, byte[]> e : res) {
Integer key = e.getKey();
byte[] val = e.getValue();
@@ -331,9 +339,9 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
assertEquals(0, key % 2);
}
- qry = cache.context().queries().createScanQuery(null, null, false);
+ qry = new ScanQuery<>();
- res = qry.execute().get();
+ res = cache.query(qry).getAll();
assertEquals(ENTRY_CNT, res.size());
}
@@ -419,9 +427,9 @@ public abstract class GridCacheSwapScanQueryAbstractSelfTest extends GridCommonA
/** {@inheritDoc} */
@Override public int hashCode() {
- int result = name != null ? name.hashCode() : 0;
+ int res = name != null ? name.hashCode() : 0;
- return 31 * result + salary;
+ return 31 * res + salary;
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/5f5220af/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
index 1ef470a..02b213e 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
@@ -22,36 +22,29 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
-import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import javax.cache.Cache;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheRebalanceMode;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.cluster.ClusterGroupEmptyCheckedException;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
-import org.apache.ignite.internal.processors.cache.query.CacheQuery;
-import org.apache.ignite.internal.processors.cache.query.CacheQueryFuture;
-import org.apache.ignite.internal.processors.cache.query.GridCacheQueryAdapter;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -91,18 +84,15 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
/** Expected first node ID. */
private static UUID expNodeId;
- /** Expected fallback node ID. */
- private static UUID expFallbackNodeId;
-
/** Communication SPI factory. */
private CommunicationSpiFactory commSpiFactory;
- /** Latch. */
- private static CountDownLatch latch;
-
/** Test entries. */
private Map<Integer, Map<Integer, Integer>> entries = new HashMap<>();
+ /** */
+ private boolean syncRebalance;
+
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
@@ -120,6 +110,10 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
ccfg.setCacheMode(cacheMode);
ccfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
ccfg.setBackups(backups);
+
+ if (syncRebalance)
+ ccfg.setRebalanceMode(CacheRebalanceMode.SYNC);
+
ccfg.setNearConfiguration(null);
cfg.setCacheConfiguration(ccfg);
@@ -144,7 +138,8 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
int part = anyLocalPartition(cache.context());
- CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(null, part, false);
+ QueryCursor<Cache.Entry<Integer, Integer>> qry =
+ cache.query(new ScanQuery<Integer, Integer>().setPartition(part));
doTestScanQuery(qry, part);
}
@@ -174,7 +169,8 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
expNodeId = tup.get2();
- CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(null, part, false);
+ QueryCursor<Cache.Entry<Integer, Integer>> qry =
+ cache.query(new ScanQuery<Integer, Integer>().setPartition(part));
doTestScanQuery(qry, part);
}
@@ -184,16 +180,22 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
}
/**
- * Scan should activate fallback mechanism when new nodes join topology and rebalancing happens in parallel with
- * scan query.
- *
* @throws Exception In case of error.
*/
public void testScanFallbackOnRebalancing() throws Exception {
+ scanFallbackOnRebalancing(false);
+ }
+
+ /**
+ * @param cur If {@code true} tests query cursor.
+ * @throws Exception In case of error.
+ */
+ private void scanFallbackOnRebalancing(final boolean cur) throws Exception {
cacheMode = CacheMode.PARTITIONED;
clientMode = false;
- backups = 1;
+ backups = 2;
commSpiFactory = new TestFallbackOnRebalancingCommunicationSpiFactory();
+ syncRebalance = true;
try {
Ignite ignite = startGrids(GRID_CNT);
@@ -214,6 +216,8 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
Thread.sleep(3000);
+ info("Will stop grid: " + getTestGridName(id));
+
stopGrid(id);
if (done.get())
@@ -224,7 +228,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
return null;
}
- }, GRID_CNT);
+ }, 2);
final AtomicInteger nodeIdx = new AtomicInteger();
@@ -233,18 +237,24 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
@Override public Object call() throws Exception {
int nodeId = nodeIdx.getAndIncrement();
- IgniteCacheProxy<Integer, Integer> cache = (IgniteCacheProxy<Integer, Integer>)
- grid(nodeId).<Integer, Integer>cache(null);
+ IgniteCache<Integer, Integer> cache = grid(nodeId).cache(null);
+
+ int cntr = 0;
while (!done.get()) {
- IgniteBiTuple<Integer, UUID> tup = remotePartition(cache.context());
+ int part = ThreadLocalRandom.current().nextInt(ignite(nodeId).affinity(null).partitions());
- int part = tup.get1();
+ if (cntr++ % 100 == 0)
+ info("Running query [node=" + nodeId + ", part=" + part + ']');
- CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(
- null, part, false);
+ try (QueryCursor<Cache.Entry<Integer, Integer>> cur0 =
+ cache.query(new ScanQuery<Integer, Integer>(part).setPageSize(5))) {
- doTestScanQuery(qry, part);
+ if (cur)
+ doTestScanQueryCursor(cur0, part);
+ else
+ doTestScanQuery(cur0, part);
+ }
}
return null;
@@ -269,9 +279,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
*
* @throws Exception In case of error.
*/
- public void testScanFallbackOnRebalancingCursor() throws Exception {
- fail("https://issues.apache.org/jira/browse/IGNITE-1239");
-
+ public void testScanFallbackOnRebalancingCursor1() throws Exception {
cacheMode = CacheMode.PARTITIONED;
clientMode = false;
backups = 1;
@@ -308,15 +316,19 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
IgniteCache<Integer, Integer> cache = grid(nodeId).cache(null);
+ int cntr = 0;
+
while (!done.get()) {
int part = ThreadLocalRandom.current().nextInt(ignite(nodeId).affinity(null).partitions());
- QueryCursor<Cache.Entry<Integer, Integer>> cur =
- cache.query(new ScanQuery<Integer, Integer>(part));
+ if (cntr++ % 100 == 0)
+ info("Running query [node=" + nodeId + ", part=" + part + ']');
- U.debug(log, "Running query [node=" + nodeId + ", part=" + part + ']');
+ try (QueryCursor<Cache.Entry<Integer, Integer>> cur =
+ cache.query(new ScanQuery<Integer, Integer>(part).setPageSize(5))) {
- doTestScanQueryCursor(cur, part);
+ doTestScanQueryCursor(cur, part);
+ }
}
return null;
@@ -332,95 +344,15 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
}
/**
- * Scan should try first remote node and fallbacks to second remote node.
- *
* @throws Exception If failed.
*/
- public void testScanFallback() throws Exception {
- cacheMode = CacheMode.PARTITIONED;
- backups = 1;
- commSpiFactory = new TestFallbackCommunicationSpiFactory();
-
- final Set<Integer> candidates = new TreeSet<>();
-
- final AtomicBoolean test = new AtomicBoolean(false);
-
- for(int j = 0; j < 2; j++) {
- clientMode = true;
-
- latch = new CountDownLatch(1);
-
- try {
- final Ignite ignite0 = startGrid(0);
-
- clientMode = false;
-
- final IgniteEx ignite1 = startGrid(1);
- final IgniteEx ignite2 = startGrid(2);
- startGrid(3);
-
- if (test.get()) {
- expNodeId = ignite1.localNode().id();
- expFallbackNodeId = ignite2.localNode().id();
- }
-
- final IgniteCacheProxy<Integer, Integer> cache = fillCache(ignite0);
-
- if (!test.get()) {
- candidates.addAll(localPartitions(ignite1));
-
- candidates.retainAll(localPartitions(ignite2));
- }
-
- Runnable run = new Runnable() {
- @Override public void run() {
- try {
- startGrid(4);
- startGrid(5);
-
- awaitPartitionMapExchange();
-
- if (!test.get()) {
- candidates.removeAll(localPartitions(ignite1));
-
- F.retain(candidates, false, localPartitions(ignite2));
- }
-
- latch.countDown();
- }
- catch (Exception e) {
- e.printStackTrace();
- }
-
- }
- };
-
- Integer part = null;
- CacheQuery<Map.Entry<Integer, Integer>> qry = null;
-
- if (test.get()) {
- part = F.first(candidates);
-
- qry = cache.context().queries().createScanQuery(null, part, false);
- }
-
- new Thread(run).start();
-
- if (test.get())
- doTestScanQuery(qry, part);
- else
- latch.await();
- }
- finally {
- test.set(true);
-
- stopAllGrids();
- }
- }
+ public void testScanFallbackOnRebalancingCursor2() throws Exception {
+ scanFallbackOnRebalancing(true);
}
/**
* @param ignite Ignite.
+ * @return Cache.
*/
protected IgniteCacheProxy<Integer, Integer> fillCache(Ignite ignite) {
IgniteCacheProxy<Integer, Integer> cache =
@@ -444,16 +376,14 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
/**
* @param qry Query.
+ * @param part Partition.
*/
- protected void doTestScanQuery(CacheQuery<Map.Entry<Integer, Integer>> qry, int part)
- throws IgniteCheckedException {
- CacheQueryFuture<Map.Entry<Integer, Integer>> fut = qry.execute();
-
- Collection<Map.Entry<Integer, Integer>> qryEntries = fut.get();
+ protected void doTestScanQuery(QueryCursor<Cache.Entry<Integer, Integer>> qry, int part) {
+ Collection<Cache.Entry<Integer, Integer>> qryEntries = qry.getAll();
Map<Integer, Integer> map = entries.get(part);
- for (Map.Entry<Integer, Integer> e : qryEntries)
+ for (Cache.Entry<Integer, Integer> e : qryEntries)
assertEquals(map.get(e.getKey()), e.getValue());
assertEquals("Invalid number of entries for partition: " + part, map.size(), qryEntries.size());
@@ -464,7 +394,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
* @param part Partition number.
*/
protected void doTestScanQueryCursor(
- QueryCursor<Cache.Entry<Integer, Integer>> cur, int part) throws IgniteCheckedException {
+ QueryCursor<Cache.Entry<Integer, Integer>> cur, int part) {
Map<Integer, Integer> map = entries.get(part);
@@ -483,6 +413,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
/**
* @param cctx Cctx.
+ * @return Local partition.
*/
private static int anyLocalPartition(GridCacheContext<?, ?> cctx) {
return F.first(cctx.topology().localPartitions()).id();
@@ -490,6 +421,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
/**
* @param cctx Cctx.
+ * @return Remote partition.
*/
private IgniteBiTuple<Integer, UUID> remotePartition(final GridCacheContext cctx) {
ClusterNode node = F.first(cctx.kernalContext().grid().cluster().forRemotes().nodes());
@@ -505,6 +437,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
/**
* @param ignite Ignite.
+ * @return Local partitions.
*/
private Set<Integer> localPartitions(Ignite ignite) {
GridCacheContext cctx = ((IgniteCacheProxy)ignite.cache(null)).context();
@@ -528,7 +461,7 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
*/
private interface CommunicationSpiFactory {
/**
- * Creates communication SPI instance.
+ * @return Communication SPI instance.
*/
TcpCommunicationSpi create();
}
@@ -541,13 +474,13 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
@Override public TcpCommunicationSpi create() {
return new TcpCommunicationSpi() {
@Override public void sendMessage(ClusterNode node, Message msg,
- IgniteInClosure<IgniteException> ackClosure) throws IgniteSpiException {
+ IgniteInClosure<IgniteException> ackC) throws IgniteSpiException {
Object origMsg = ((GridIoMessage)msg).message();
if (origMsg instanceof GridCacheQueryRequest)
fail(); //should use local node
- super.sendMessage(node, msg, ackClosure);
+ super.sendMessage(node, msg, ackC);
}
};
}
@@ -561,44 +494,13 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
@Override public TcpCommunicationSpi create() {
return new TcpCommunicationSpi() {
@Override public void sendMessage(ClusterNode node, Message msg,
- IgniteInClosure<IgniteException> ackClosure) throws IgniteSpiException {
+ IgniteInClosure<IgniteException> ackC) throws IgniteSpiException {
Object origMsg = ((GridIoMessage)msg).message();
if (origMsg instanceof GridCacheQueryRequest)
assertEquals(expNodeId, node.id());
- super.sendMessage(node, msg, ackClosure);
- }
- };
- }
- }
-
- /**
- *
- */
- private static class TestFallbackCommunicationSpiFactory implements CommunicationSpiFactory {
- /** {@inheritDoc} */
- @Override public TcpCommunicationSpi create() {
- return new TcpCommunicationSpi() {
- @Override public void sendMessage(ClusterNode node, Message msg,
- IgniteInClosure<IgniteException> ackClosure) throws IgniteSpiException {
- Object origMsg = ((GridIoMessage)msg).message();
-
- if (origMsg instanceof GridCacheQueryRequest) {
- if (latch.getCount() > 0)
- assertEquals(expNodeId, node.id());
- else
- assertEquals(expFallbackNodeId, node.id());
-
- try {
- latch.await();
- }
- catch (InterruptedException e) {
- throw new IgniteSpiException(e);
- }
- }
-
- super.sendMessage(node, msg, ackClosure);
+ super.sendMessage(node, msg, ackC);
}
};
}
[19/32] ignite git commit: minor
Posted by ra...@apache.org.
minor
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d64fc9d1
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d64fc9d1
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d64fc9d1
Branch: refs/heads/feature/ignite-535-mqtt
Commit: d64fc9d105c66c08234d7bdf72046128456620a5
Parents: 3676cbe
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Thu Sep 17 18:03:57 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Thu Sep 17 18:03:57 2015 +0300
----------------------------------------------------------------------
.../cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/d64fc9d1/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 d9f6840..0cbad48 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
@@ -418,4 +418,4 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
@Override public String toString() {
return S.toString(GridDhtAtomicUpdateFuture.class, this);
}
-}
\ No newline at end of file
+}
[08/32] ignite git commit: Merge remote-tracking branch
'origin/ignite-1.4' into ignite-1.4
Posted by ra...@apache.org.
Merge remote-tracking branch 'origin/ignite-1.4' into ignite-1.4
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f8577cab
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f8577cab
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f8577cab
Branch: refs/heads/feature/ignite-535-mqtt
Commit: f8577cab29faa87dd337295e3dbac68f0b94ac63
Parents: 8e25204 7a71b20
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Tue Sep 15 17:09:37 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Tue Sep 15 17:09:37 2015 +0300
----------------------------------------------------------------------
.../ignite/internal/processors/igfs/IgfsAbstractSelfTest.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
[20/32] ignite git commit: Merge branch 'ignite-1.4'
Posted by ra...@apache.org.
Merge branch 'ignite-1.4'
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b6139f86
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b6139f86
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b6139f86
Branch: refs/heads/feature/ignite-535-mqtt
Commit: b6139f86f8994dbbdd8acabedcd65b89e2b5aa92
Parents: c311c3c d64fc9d
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Fri Sep 18 11:44:13 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Fri Sep 18 11:44:13 2015 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/IgniteKernal.java | 7 -
.../processors/cache/GridCacheContext.java | 6 +-
.../cache/GridCacheEvictionManager.java | 6 +-
.../cache/GridCacheEvictionResponse.java | 2 +-
.../processors/cache/GridCacheIoManager.java | 47 ++++---
.../processors/cache/GridCacheMessage.java | 7 +
.../processors/cache/GridCacheMvccManager.java | 34 ++++-
.../GridCachePartitionExchangeManager.java | 41 ++++--
.../processors/cache/GridCacheProcessor.java | 28 ++--
.../GridDistributedLockResponse.java | 6 +-
.../GridDistributedTxPrepareResponse.java | 6 +-
.../distributed/dht/GridDhtTopologyFuture.java | 6 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 2 +-
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 2 +-
.../dht/atomic/GridDhtAtomicUpdateResponse.java | 12 +-
.../dht/atomic/GridNearAtomicUpdateFuture.java | 16 ++-
.../dht/atomic/GridNearAtomicUpdateRequest.java | 2 +
.../atomic/GridNearAtomicUpdateResponse.java | 11 +-
.../colocated/GridDhtColocatedLockFuture.java | 44 ++++--
.../dht/preloader/GridDhtForceKeysFuture.java | 2 +-
.../dht/preloader/GridDhtForceKeysResponse.java | 6 +-
.../GridDhtPartitionsExchangeFuture.java | 19 ++-
.../distributed/near/GridNearGetResponse.java | 6 +-
.../distributed/near/GridNearLockFuture.java | 26 +++-
.../near/GridNearOptimisticTxPrepareFuture.java | 20 ++-
.../near/GridNearTxFinishResponse.java | 6 +-
.../cache/query/GridCacheQueryResponse.java | 6 +-
.../continuous/CacheContinuousQueryHandler.java | 12 +-
.../transactions/IgniteTxLocalAdapter.java | 4 +-
.../ignite/internal/util/GridSpinBusyLock.java | 10 ++
.../IgniteCacheEntryProcessorNodeJoinTest.java | 24 +++-
.../loadtests/hashmap/GridCacheTestContext.java | 4 +-
.../ignite/testframework/GridTestUtils.java | 14 +-
.../cache/CacheIndexStreamerTest.java | 137 +++++++++++++++++++
.../IgniteCacheQueryNodeRestartSelfTest2.java | 2 -
.../ignite/visor/commands/VisorConsole.scala | 8 +-
36 files changed, 450 insertions(+), 141 deletions(-)
----------------------------------------------------------------------
[26/32] ignite git commit: minor (fixed warning messages)
Posted by ra...@apache.org.
minor (fixed warning messages)
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/84a23008
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/84a23008
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/84a23008
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 84a230084e9488a4ddb9b52f86ecd6c623baf745
Parents: d64fc9d
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Fri Sep 18 23:15:59 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Fri Sep 18 23:15:59 2015 +0300
----------------------------------------------------------------------
.../org/apache/ignite/spi/discovery/tcp/ServerImpl.java | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/84a23008/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 3e50b94..4ce46e8 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -4437,9 +4437,8 @@ class ServerImpl extends TcpDiscoveryImpl {
LT.warn(log, null, "Unknown connection detected (is some other software connecting to " +
"this Ignite port?" +
- (!spi.isSslEnabled() ? " missed SSL configuration?" : "" ) +
- ") [rmtAddr=" + sock.getRemoteSocketAddress() +
- ", locAddr=" + sock.getLocalSocketAddress() + ']');
+ (!spi.isSslEnabled() ? " missing SSL configuration on remote node?" : "" ) +
+ ") [rmtAddr=" + sock.getInetAddress() + ']', true);
return;
}
@@ -4555,8 +4554,9 @@ class ServerImpl extends TcpDiscoveryImpl {
U.error(log, "Caught exception on handshake [err=" + e +", sock=" + sock + ']', e);
if (X.hasCause(e, SSLException.class) && spi.isSslEnabled() && !spi.isNodeStopping0())
- LT.warn(log, null, "Failed to initialize connection. Not encrypted data received. " +
- "Missed SSL configuration on node? [sock=" + sock + ']');
+ LT.warn(log, null, "Failed to initialize connection " +
+ "(missing SSL configuration on remote node?) " +
+ "[rmtAddr=" + sock.getInetAddress() + ']', true);
else if ((X.hasCause(e, ObjectStreamException.class) || !sock.isClosed())
&& !spi.isNodeStopping0()) {
if (U.isMacInvalidArgumentError(e))
@@ -5293,4 +5293,4 @@ class ServerImpl extends TcpDiscoveryImpl {
this.sock = sock;
}
}
-}
\ No newline at end of file
+}
[18/32] ignite git commit: Test for ignite-973.
Posted by ra...@apache.org.
Test for ignite-973.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3676cbe7
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3676cbe7
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3676cbe7
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 3676cbe7f5f5f73199487318d6841e50a1f73496
Parents: 585761f
Author: sboikov <sb...@gridgain.com>
Authored: Thu Sep 17 17:52:24 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Sep 17 17:52:24 2015 +0300
----------------------------------------------------------------------
.../ignite/testframework/GridTestUtils.java | 14 +-
.../cache/CacheIndexStreamerTest.java | 137 +++++++++++++++++++
2 files changed, 150 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/3676cbe7/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
index edf7c52..be3f0e4 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
@@ -633,11 +633,23 @@ public final class GridTestUtils {
*/
@SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor")
public static <T> IgniteInternalFuture<T> runAsync(final Callable<T> task) {
+ return runAsync(task, "async-runner");
+ }
+
+ /**
+ * Runs callable task asyncronously.
+ *
+ * @param task Callable.
+ * @param threadName Thread name.
+ * @return Future with task result.
+ */
+ @SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor")
+ public static <T> IgniteInternalFuture<T> runAsync(final Callable<T> task, String threadName) {
if (!busyLock.enterBusy())
throw new IllegalStateException("Failed to start new threads (test is being stopped).");
try {
- final GridTestSafeThreadFactory thrFactory = new GridTestSafeThreadFactory("async-runner");
+ final GridTestSafeThreadFactory thrFactory = new GridTestSafeThreadFactory(threadName);
final GridFutureAdapter<T> fut = new GridFutureAdapter<T>() {
@Override public boolean cancel() throws IgniteCheckedException {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3676cbe7/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheIndexStreamerTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheIndexStreamerTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheIndexStreamerTest.java
new file mode 100644
index 0000000..25c3b81
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheIndexStreamerTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.cache.CacheMemoryMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.typedef.internal.U;
+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.swapspace.file.FileSwapSpaceSpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ *
+ */
+public class CacheIndexStreamerTest extends GridCommonAbstractTest {
+ /** */
+ private final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setSwapSpaceSpi(new FileSwapSpaceSpi());
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testStreamer() throws Exception {
+ final Ignite ignite = startGrid(0);
+
+ final IgniteCache<Integer, String> cache = ignite.createCache(cacheConfiguration());
+
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ final int KEYS= 10_000;
+
+ try {
+ IgniteInternalFuture streamerFut = GridTestUtils.runAsync(new Callable() {
+ @Override public Void call() throws Exception {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ while (!stop.get()) {
+ try (IgniteDataStreamer<Integer, String> streamer = ignite.dataStreamer(null)) {
+ for (int i = 0; i < 1; i++)
+ streamer.addData(rnd.nextInt(KEYS), String.valueOf(i));
+ }
+ }
+
+ return null;
+ }
+ }, "streamer-thread");
+
+ IgniteInternalFuture updateFut = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ while (!stop.get()) {
+ for (int i = 0; i < 100; i++) {
+ Integer key = rnd.nextInt(KEYS);
+
+ cache.put(key, String.valueOf(key));
+
+ cache.remove(key);
+ }
+ }
+
+ return null;
+ }
+ }, 1, "update-thread");
+
+ U.sleep(30_000);
+
+ stop.set(true);
+
+ streamerFut.get();
+ updateFut.get();
+ }
+ finally {
+ stop.set(true);
+
+ stopAllGrids();
+ }
+ }
+
+ /**
+ * @return Cache configuration.
+ */
+ private CacheConfiguration cacheConfiguration() {
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setAtomicityMode(ATOMIC);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setMemoryMode(CacheMemoryMode.OFFHEAP_TIERED);
+ ccfg.setOffHeapMaxMemory(0);
+ ccfg.setBackups(1);
+ ccfg.setIndexedTypes(Integer.class, String.class);
+
+ return ccfg;
+ }
+
+
+}
[13/32] ignite git commit: IGNITE-1498 Added support for -cfg=path
command line option.
Posted by ra...@apache.org.
IGNITE-1498 Added support for -cfg=path command line option.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/cd6a1d51
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/cd6a1d51
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/cd6a1d51
Branch: refs/heads/feature/ignite-535-mqtt
Commit: cd6a1d513adab5a24b33a9315bfd083c2f4383ee
Parents: 6187241
Author: Alexey Kuznetsov <ak...@apache.org>
Authored: Thu Sep 17 14:33:00 2015 +0700
Committer: Alexey Kuznetsov <ak...@apache.org>
Committed: Thu Sep 17 14:33:00 2015 +0700
----------------------------------------------------------------------
.../ignite/visor/commands/VisorConsole.scala | 31 +++++++++++++-------
1 file changed, 20 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/cd6a1d51/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
----------------------------------------------------------------------
diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
index bcfc6e0..627f795 100644
--- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
+++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/VisorConsole.scala
@@ -84,13 +84,14 @@ class VisorConsole {
if (hasArgFlag("?", argLst) || hasArgFlag("help", argLst)) {
println("Usage:")
- println(s" $progName [?]|[{-v}{-np}]|[{-b=<batch commands file path>} {-e=command1;command2}]")
+ println(s" $progName [? | -help]|[{-v}{-np} {-cfg=<path>}]|[{-b=<path>} {-e=command1;command2;...}]")
println(" Where:")
- println(" ?, /help, -help - show this message.")
- println(" -v - verbose mode (quiet by default).")
- println(" -np - no pause on exit (pause by default)")
- println(" -b - batch mode with file)")
- println(" -e - batch mode with commands)")
+ println(" ?, /help, -help - show this message.")
+ println(" -v - verbose mode (quiet by default).")
+ println(" -np - no pause on exit (pause by default).")
+ println(" -cfg=<path> - connect with specified configuration.")
+ println(" -b=<path> - batch mode with file.")
+ println(" -e==cmd1;cmd2;... - batch mode with commands.")
visor.quit()
}
@@ -99,14 +100,22 @@ class VisorConsole {
}
protected def buildReader(argLst: ArgList) = {
+ val cfgFile = argValue("cfg", argLst)
val batchFile = argValue("b", argLst)
val batchCommand = argValue("e", argLst)
+ cfgFile.foreach(cfg => {
+ if (batchFile.isDefined || batchCommand.isDefined) {
+ visor.warn("Options can't contains both -cfg and one of -b or -e options.")
+
+ visor.quit()
+ }
+
+ visor.searchCmd("open").foreach(_.withArgs("-cpath=" + cfg))
+ })
+
if (batchFile.isDefined && batchCommand.isDefined) {
- visor.warn(
- "Illegal options can't contains both command file and commands",
- s"Usage: $progName {-b=<batch commands file path>} {-e=command1;command2}"
- )
+ visor.warn("Options can't contains both command file and commands.")
visor.quit()
}
@@ -303,7 +312,7 @@ object VisorConsole extends VisorConsole with App {
addCommands()
private val argLst = parse(args.mkString(" "))
-
+
private val reader = buildReader(argLst)
visor.reader(reader)
[02/32] ignite git commit: fixed javadoc ignite-1.4
Posted by ra...@apache.org.
fixed javadoc ignite-1.4
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/72e270c6
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/72e270c6
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/72e270c6
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 72e270c63e99ac17a077b26fb39302f603d15188
Parents: 3f75c1c
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Tue Sep 15 16:55:00 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Tue Sep 15 16:55:00 2015 +0300
----------------------------------------------------------------------
.../configuration/NearCacheConfiguration.java | 18 ++++++++++++++++--
1 file changed, 16 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/72e270c6/modules/core/src/main/java/org/apache/ignite/configuration/NearCacheConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/NearCacheConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/NearCacheConfiguration.java
index bbaa030..3c59bd6 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/NearCacheConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/NearCacheConfiguration.java
@@ -25,7 +25,12 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import static org.apache.ignite.configuration.CacheConfiguration.DFLT_NEAR_START_SIZE;
/**
- * Client cache configuration.
+ * Client (near) cache configuration.
+ * <p>
+ * Distributed cache can also be fronted by a Near cache,
+ * which is a smaller local cache that stores most recently
+ * or most frequently accessed data. Just like with a partitioned cache,
+ * the user can control the size of the near cache and its eviction policies.
*/
public class NearCacheConfiguration<K, V> extends MutableConfiguration<K, V> {
/** */
@@ -45,6 +50,8 @@ public class NearCacheConfiguration<K, V> extends MutableConfiguration<K, V> {
}
/**
+ * Creates near cache configuration copying properties from passed in configuration.
+ *
* @param ccfg Configuration to copy.
*/
public NearCacheConfiguration(NearCacheConfiguration<K, V> ccfg) {
@@ -55,13 +62,20 @@ public class NearCacheConfiguration<K, V> extends MutableConfiguration<K, V> {
}
/**
+ * Gets near eviction policy. By default, returns {@code null}
+ * which means that evictions are disabled for near cache.
+ *
* @return Near eviction policy.
+ * @see CacheConfiguration#getEvictionPolicy()
+ * @see CacheConfiguration#isEvictSynchronized()
*/
public EvictionPolicy<K, V> getNearEvictionPolicy() {
return nearEvictPlc;
}
/**
+ * Sets near eviction policy.
+ *
* @param nearEvictPlc Near eviction policy.
* @return {@code this} for chaining.
*/
@@ -97,4 +111,4 @@ public class NearCacheConfiguration<K, V> extends MutableConfiguration<K, V> {
@Override public String toString() {
return S.toString(NearCacheConfiguration.class, this, super.toString());
}
-}
\ No newline at end of file
+}
[32/32] ignite git commit: IGNITE-535 Finish MQTT Streamer docs and
tests. Upgrade latter to AMQ 5.12.0.
Posted by ra...@apache.org.
IGNITE-535 Finish MQTT Streamer docs and tests. Upgrade latter to AMQ 5.12.0.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/296dd6e7
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/296dd6e7
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/296dd6e7
Branch: refs/heads/feature/ignite-535-mqtt
Commit: 296dd6e7d86fe6d0914a9fbf8062632c04e4d22c
Parents: f03f3a3
Author: Raul Kripalani <ra...@apache.org>
Authored: Mon Sep 21 17:24:44 2015 +0100
Committer: Raul Kripalani <ra...@apache.org>
Committed: Mon Sep 21 17:24:44 2015 +0100
----------------------------------------------------------------------
modules/mqtt/pom.xml | 9 +-
.../apache/ignite/stream/mqtt/MqttStreamer.java | 156 ++++++++++++++++++-
.../stream/mqtt/IgniteMqttStreamerTest.java | 80 +++++++++-
3 files changed, 224 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/296dd6e7/modules/mqtt/pom.xml
----------------------------------------------------------------------
diff --git a/modules/mqtt/pom.xml b/modules/mqtt/pom.xml
index 4b0b46c..21511e8 100644
--- a/modules/mqtt/pom.xml
+++ b/modules/mqtt/pom.xml
@@ -37,7 +37,7 @@
<properties>
<paho.version>1.0.2</paho.version>
- <activemq.version>5.11.1</activemq.version>
+ <activemq.version>5.12.0</activemq.version>
<guava-retryier.version>2.0.0</guava-retryier.version>
</properties>
@@ -69,13 +69,6 @@
<dependency>
<groupId>org.apache.activemq</groupId>
- <artifactId>activemq-kahadb-store</artifactId>
- <version>${activemq.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.activemq</groupId>
<artifactId>activemq-mqtt</artifactId>
<version>${activemq.version}</version>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/ignite/blob/296dd6e7/modules/mqtt/src/main/java/org/apache/ignite/stream/mqtt/MqttStreamer.java
----------------------------------------------------------------------
diff --git a/modules/mqtt/src/main/java/org/apache/ignite/stream/mqtt/MqttStreamer.java b/modules/mqtt/src/main/java/org/apache/ignite/stream/mqtt/MqttStreamer.java
index b86d385..f18ae42 100644
--- a/modules/mqtt/src/main/java/org/apache/ignite/stream/mqtt/MqttStreamer.java
+++ b/modules/mqtt/src/main/java/org/apache/ignite/stream/mqtt/MqttStreamer.java
@@ -62,12 +62,17 @@ import org.eclipse.paho.client.mqttv3.MqttMessage;
* <li>Specifying the subscriber's QoS for a single topic or for multiple topics.</li>
* <li>Allows setting {@link MqttConnectOptions} to support features like last will testament, persistent
* sessions, etc.</li>
- * <li>Specifying the client ID.</li>
+ * <li>Specifying the client ID. A random one will be generated and maintained throughout reconnections if the user
+ * does not provide one.</li>
+ * <li>(Re-)Connection retries based on the <i>guava-retrying</i> library. Retry wait and retry stop policies
+ * can be configured.</li>
+ * <li>Blocking the start() method until connected for the first time.</li>
* </ul>
*
- * Note: features like durable subscriptions, last will testament, etc. must be configured via the
+ * Note: features like durable subscriptions, last will testament, etc. can be configured via the
* {@link #connectOptions} property.
*
+ * @see <a href="https://github.com/rholder/guava-retrying">guava-retrying library</a>
* @author Raul Kripalani
*/
public class MqttStreamer<K, V> extends StreamAdapter<MqttMessage, K, V> implements MqttCallback {
@@ -75,46 +80,65 @@ public class MqttStreamer<K, V> extends StreamAdapter<MqttMessage, K, V> impleme
/** Logger. */
private IgniteLogger log;
+ /** The MQTT client object for internal use. */
private MqttClient client;
+ /** The broker URL, set by the user. */
private String brokerUrl;
+ /** The topic to subscribe to, if a single topic. */
private String topic;
+ /** The quality of service to use for a single topic subscription (optional). */
private Integer qualityOfService;
+ /** The topics to subscribe to, if many. */
private List<String> topics;
+ /** The qualities of service to use for multiple topic subscriptions. If specified, it must contain the same
+ * number of elements as {@link #topics}. */
private List<Integer> qualitiesOfService;
- /** Client ID in case we're using durable subscribers. */
+ /** The MQTT client ID (optional). */
private String clientId;
+ /** A configurable persistence mechanism. If not set, Paho will use its default. */
private MqttClientPersistence persistence;
+ /** The MQTT client connect options, where users can configured the last will and testament, durability, etc. */
private MqttConnectOptions connectOptions;
- // disconnect parameters
+ /** Quiesce timeout on disconnection. */
private Integer disconnectQuiesceTimeout;
+ /** Whether to disconnect forcibly or not. */
private boolean disconnectForcibly;
+ /** If disconnecting forcibly, the timeout. */
private Integer disconnectForciblyTimeout;
+ /** The strategy to determine how long to wait between retry attempts. By default, this streamer uses a
+ * Fibonacci-based strategy. */
private WaitStrategy retryWaitStrategy = WaitStrategies.fibonacciWait();
+ /** The strategy to determine when to stop retrying to (re-)connect. By default, we never stop. */
private StopStrategy retryStopStrategy = StopStrategies.neverStop();
+ /** The internal connection retrier object with a thread pool of size 1. */
private MqttConnectionRetrier connectionRetrier;
+ /** Whether to block the start() method until connected for the first time. */
+ private boolean blockUntilConnected;
+
+ /** State keeping. */
private volatile boolean stopped = true;
+ /** State keeping. */
private volatile boolean connected;
+ /** Cached log prefix for cache messages. */
private String cachedLogPrefix;
- private boolean blockUntilConnected;
-
/**
* Starts streamer.
*
@@ -136,7 +160,11 @@ public class MqttStreamer<K, V> extends StreamAdapter<MqttMessage, K, V> impleme
A.ensure(getSingleTupleExtractor() == null || getMultipleTupleExtractor() == null, "cannot provide " +
"both single and multiple tuple extractor");
A.notNullOrEmpty(brokerUrl, "broker URL");
- A.notNullOrEmpty(clientId, "client ID");
+
+ // if the client ID is empty, generate one
+ if (clientId == null || clientId.length() == 0) {
+ clientId = MqttClient.generateClientId();
+ }
// if we have both a single topic and a list of topics (but the list of topic is not of
// size 1 and == topic, as this would be a case of re-initialization), fail
@@ -257,6 +285,9 @@ public class MqttStreamer<K, V> extends StreamAdapter<MqttMessage, K, V> impleme
// MQTT Client callback methods
// -------------------------------
+ /**
+ * {@inheritDoc}
+ */
@Override public void connectionLost(Throwable throwable) {
connected = false;
@@ -268,6 +299,9 @@ public class MqttStreamer<K, V> extends StreamAdapter<MqttMessage, K, V> impleme
connectionRetrier.connect();
}
+ /**
+ * {@inheritDoc}
+ */
@Override public void messageArrived(String topic, MqttMessage message) throws Exception {
if (getMultipleTupleExtractor() != null) {
Map<K, V> entries = getMultipleTupleExtractor().extract(message);
@@ -285,6 +319,9 @@ public class MqttStreamer<K, V> extends StreamAdapter<MqttMessage, K, V> impleme
}
}
+ /**
+ * {@inheritDoc}
+ */
@Override public void deliveryComplete(IMqttDeliveryToken token) {
// ignore, as we don't send messages
}
@@ -293,127 +330,229 @@ public class MqttStreamer<K, V> extends StreamAdapter<MqttMessage, K, V> impleme
// Getters and setters
// -------------------------------
+ /**
+ * @return
+ */
public String getBrokerUrl() {
return brokerUrl;
}
+ /**
+ * @param brokerUrl The Broker URL (compulsory).
+ */
public void setBrokerUrl(String brokerUrl) {
this.brokerUrl = brokerUrl;
}
+ /**
+ * @return
+ */
public String getTopic() {
return topic;
}
+ /**
+ * @param topic The topic to subscribe to, if a single topic.
+ */
public void setTopic(String topic) {
this.topic = topic;
}
+ /**
+ * @return
+ */
public Integer getQualityOfService() {
return qualityOfService;
}
+ /**
+ * @param qualityOfService The quality of service to use for a single topic subscription (optional).
+ */
public void setQualityOfService(Integer qualityOfService) {
this.qualityOfService = qualityOfService;
}
+ /**
+ * @return
+ */
public List<String> getTopics() {
return topics;
}
+ /**
+ * @param topics The topics to subscribe to, if many.
+ */
public void setTopics(List<String> topics) {
this.topics = topics;
}
+ /**
+ * @return
+ */
public List<Integer> getQualitiesOfService() {
return qualitiesOfService;
}
+ /**
+ * @param qualitiesOfService The qualities of service to use for multiple topic subscriptions.
+ * If specified, the list must contain the same number of elements as {@link #topics}.
+ */
public void setQualitiesOfService(List<Integer> qualitiesOfService) {
this.qualitiesOfService = qualitiesOfService;
}
+ /**
+ * @return
+ */
public String getClientId() {
return clientId;
}
+ /**
+ * @param clientId The MQTT client ID (optional). If one is not provided, we'll create one for you and maintain
+ * it througout any reconnection attempts.
+ */
public void setClientId(String clientId) {
this.clientId = clientId;
}
+ /**
+ * @return
+ */
public MqttClientPersistence getPersistence() {
return persistence;
}
+ /**
+ * @param persistence A configurable persistence mechanism. If not set, Paho will use its default.
+ */
public void setPersistence(MqttClientPersistence persistence) {
this.persistence = persistence;
}
+ /**
+ * @return
+ */
public MqttConnectOptions getConnectOptions() {
return connectOptions;
}
+ /**
+ * @param connectOptions The MQTT client connect options, where users can configured the last will and testament, durability, etc.
+ */
public void setConnectOptions(MqttConnectOptions connectOptions) {
this.connectOptions = connectOptions;
}
+ /**
+ * @return
+ */
public boolean isDisconnectForcibly() {
return disconnectForcibly;
}
+ /**
+ * @param disconnectForcibly Whether to disconnect forcibly or not. By default, it's false.
+ */
public void setDisconnectForcibly(boolean disconnectForcibly) {
this.disconnectForcibly = disconnectForcibly;
}
+ /**
+ * @return
+ */
public Integer getDisconnectQuiesceTimeout() {
return disconnectQuiesceTimeout;
}
+ /**
+ * @param disconnectQuiesceTimeout Quiesce timeout on disconnection. If not provided, this streamer won't use any.
+ */
public void setDisconnectQuiesceTimeout(Integer disconnectQuiesceTimeout) {
this.disconnectQuiesceTimeout = disconnectQuiesceTimeout;
}
+ /**
+ * @return
+ */
public Integer getDisconnectForciblyTimeout() {
return disconnectForciblyTimeout;
}
+ /**
+ * @param disconnectForciblyTimeout If disconnecting forcibly, the timeout. Compulsory in that case.
+ */
public void setDisconnectForciblyTimeout(Integer disconnectForciblyTimeout) {
this.disconnectForciblyTimeout = disconnectForciblyTimeout;
}
+ /**
+ * @return
+ */
public WaitStrategy getRetryWaitStrategy() {
return retryWaitStrategy;
}
+ /**
+ * @param retryWaitStrategy The strategy to determine how long to wait between retry attempts.
+ * By default, this streamer uses a Fibonacci-based strategy.
+ */
public void setRetryWaitStrategy(WaitStrategy retryWaitStrategy) {
this.retryWaitStrategy = retryWaitStrategy;
}
+ /**
+ * @return
+ */
public StopStrategy getRetryStopStrategy() {
return retryStopStrategy;
}
+ /**
+ * @param retryStopStrategy The strategy to determine when to stop retrying to (re-)connect. By default, we never stop.
+ */
public void setRetryStopStrategy(StopStrategy retryStopStrategy) {
this.retryStopStrategy = retryStopStrategy;
}
+ /**
+ * @return
+ */
public boolean isBlockUntilConnected() {
return blockUntilConnected;
}
+ /**
+ * @param blockUntilConnected Whether to block the start() method until connected for the first time. By default,
+ * false.
+ */
public void setBlockUntilConnected(boolean blockUntilConnected) {
this.blockUntilConnected = blockUntilConnected;
}
+ /**
+ * A utility class to help us with (re-)connecting to the MQTT broker. It uses a single-threaded executor to perform
+ * the (re-)connections.
+ */
private class MqttConnectionRetrier {
+ /** The guava-retrying retrier object. */
private final Retryer<Boolean> retrier;
+
+ /** Single-threaded pool. */
private ExecutorService executor = Executors.newSingleThreadExecutor();
+ /**
+ * Constructor.
+ * @param retrier The retryier object.
+ */
public MqttConnectionRetrier(Retryer<Boolean> retrier) {
this.retrier = retrier;
}
+ /**
+ * Method that is called by the streamer to ask us to (re-)connect.
+ */
public void connect() {
Callable<Boolean> callable = retrier.wrap(new Callable<Boolean>() {
@Override public Boolean call() throws Exception {
@@ -460,6 +599,9 @@ public class MqttStreamer<K, V> extends StreamAdapter<MqttMessage, K, V> impleme
}
}
+ /**
+ * Stops this connection utility class by shutting down the thread pool.
+ */
public void stop() {
executor.shutdownNow();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/296dd6e7/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
----------------------------------------------------------------------
diff --git a/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java b/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
index 012486a..5ac7339 100644
--- a/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
+++ b/modules/mqtt/src/test/java/org/apache/ignite/stream/mqtt/IgniteMqttStreamerTest.java
@@ -65,24 +65,41 @@ import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
*/
public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
+ /** The test data. */
private static final Map<Integer, String> TEST_DATA = new HashMap<>();
+
+ /** Topic name for single topic tests. */
private static final String SINGLE_TOPIC_NAME = "abc";
+
+ /** Topic names for multiple topic tests. */
private static final List<String> MULTIPLE_TOPIC_NAMES = Arrays.asList("def", "ghi", "jkl", "mno");
+ /** The AMQ broker with an MQTT interface. */
private BrokerService broker;
+
+ /** The MQTT client. */
private MqttClient client;
+
+ /** The broker URL. */
private String brokerUrl;
+
+ /** The broker port. **/
private int port;
+
+ /** The MQTT streamer currently under test. */
private MqttStreamer<Integer, String> streamer;
+
+ /** The UUID of the currently active remote listener. */
private UUID remoteListener;
+ /** The Ignite data streamer. */
+ private IgniteDataStreamer<Integer, String> dataStreamer;
+
static {
for (int i = 0; i < 100; i++)
TEST_DATA.put(i, "v" + i);
}
- private IgniteDataStreamer<Integer, String> dataStreamer;
-
/** Constructor. */
public IgniteMqttStreamerTest() {
super(true);
@@ -99,14 +116,17 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
// create the broker
broker = new BrokerService();
- broker.deleteAllMessages();
+ broker.setDeleteAllMessagesOnStartup(true);
broker.setPersistent(false);
+ broker.setPersistenceAdapter(null);
+ broker.setPersistenceFactory(null);
PolicyMap policyMap = new PolicyMap();
PolicyEntry policy = new PolicyEntry();
policy.setQueuePrefetch(1);
broker.setDestinationPolicy(policyMap);
broker.getDestinationPolicy().setDefaultEntry(policy);
+ broker.setSchedulerSupport(false);
// add the MQTT transport connector to the broker
broker.addConnector("mqtt://localhost:" + port);
@@ -143,6 +163,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
}
+ /**
+ * @throws Exception
+ */
public void testSingleTopic_NoQoS_OneEntryPerMessage() throws Exception {
// configure streamer
streamer.setSingleTupleExtractor(singleTupleExtractor());
@@ -162,6 +185,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
assertCacheEntriesLoaded(50);
}
+ /**
+ * @throws Exception
+ */
public void testMultipleTopics_NoQoS_OneEntryPerMessage() throws Exception {
// configure streamer
streamer.setSingleTupleExtractor(singleTupleExtractor());
@@ -185,6 +211,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
assertTrue(broker.getBroker().getDestinationMap().containsKey(new ActiveMQTopic("ghi")));
}
+ /**
+ * @throws Exception
+ */
public void testSingleTopic_NoQoS_MultipleEntriesOneMessage() throws Exception {
// configure streamer
streamer.setMultipleTupleExtractor(multipleTupleExtractor());
@@ -204,6 +233,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
assertCacheEntriesLoaded(50);
}
+ /**
+ * @throws Exception
+ */
public void testMultipleTopics_NoQoS_MultipleEntriesOneMessage() throws Exception {
// configure streamer
streamer.setMultipleTupleExtractor(multipleTupleExtractor());
@@ -227,6 +259,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
assertTrue(broker.getBroker().getDestinationMap().containsKey(new ActiveMQTopic("ghi")));
}
+ /**
+ * @throws Exception
+ */
public void testSingleTopic_NoQoS_ConnectOptions_Durable() throws Exception {
// configure streamer
streamer.setSingleTupleExtractor(singleTupleExtractor());
@@ -265,6 +300,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
assertCacheEntriesLoaded(100);
}
+ /**
+ * @throws Exception
+ */
public void testSingleTopic_NoQoS_Reconnect() throws Exception {
// configure streamer
streamer.setSingleTupleExtractor(singleTupleExtractor());
@@ -306,6 +344,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
assertCacheEntriesLoaded(100);
}
+ /**
+ * @throws Exception
+ */
public void testSingleTopic_NoQoS_RetryOnce() throws Exception {
// configure streamer
streamer.setSingleTupleExtractor(singleTupleExtractor());
@@ -339,6 +380,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
}
+ /**
+ * @throws Exception
+ */
public void testMultipleTopics_MultipleQoS_OneEntryPerMessage() throws Exception {
// configure streamer
streamer.setSingleTupleExtractor(singleTupleExtractor());
@@ -363,6 +407,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
assertTrue(broker.getBroker().getDestinationMap().containsKey(new ActiveMQTopic("ghi")));
}
+ /**
+ * @throws Exception
+ */
public void testMultipleTopics_MultipleQoS_Mismatch() throws Exception {
// configure streamer
streamer.setSingleTupleExtractor(singleTupleExtractor());
@@ -379,6 +426,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
}
+ /**
+ * @throws Exception
+ */
private MqttStreamer<Integer, String> createMqttStreamer(IgniteDataStreamer<Integer, String> dataStreamer) {
MqttStreamer<Integer, String> streamer = new MqttStreamer<>();
streamer.setIgnite(grid());
@@ -393,7 +443,10 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
return streamer;
}
- public void sendMessages(final List<String> topics, int fromIdx, int count, boolean singleMessage) throws MqttException {
+ /**
+ * @throws Exception
+ */
+ private void sendMessages(final List<String> topics, int fromIdx, int count, boolean singleMessage) throws MqttException {
if (singleMessage) {
final List<StringBuilder> sbs = new ArrayList<>(topics.size());
// initialize String Builders for each topic
@@ -423,6 +476,9 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
}
}
+ /**
+ * @throws Exception
+ */
private CountDownLatch subscribeToPutEvents(int expect) {
Ignite ignite = grid();
@@ -439,14 +495,16 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
return latch;
}
+ /**
+ * @throws Exception
+ */
private void assertCacheEntriesLoaded(int count) {
// get the cache and check that the entries are present
IgniteCache<Integer, String> cache = grid().cache(null);
// for each key from 0 to count from the TEST_DATA (ordered by key), check that the entry is present in cache
- for (Integer key : new ArrayList<>(new TreeSet<>(TEST_DATA.keySet())).subList(0, count)) {
+ for (Integer key : new ArrayList<>(new TreeSet<>(TEST_DATA.keySet())).subList(0, count))
assertEquals(TEST_DATA.get(key), cache.get(key));
- }
// assert that the cache exactly the specified amount of elements
assertEquals(count, cache.size(CachePeekMode.ALL));
@@ -455,6 +513,11 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
grid().events(grid().cluster().forCacheNodes(null)).stopRemoteListen(remoteListener);
}
+ /**
+ * Returns a {@link StreamSingleTupleExtractor} for testing.
+ *
+ * @throws Exception
+ */
public static StreamSingleTupleExtractor<MqttMessage, Integer, String> singleTupleExtractor() {
return new StreamSingleTupleExtractor<MqttMessage, Integer, String>() {
@Override public Map.Entry<Integer, String> extract(MqttMessage msg) {
@@ -464,6 +527,11 @@ public class IgniteMqttStreamerTest extends GridCommonAbstractTest {
};
}
+ /**
+ * Returns a {@link StreamMultipleTupleExtractor} for testing.
+ *
+ * @throws Exception
+ */
public static StreamMultipleTupleExtractor<MqttMessage, Integer, String> multipleTupleExtractor() {
return new StreamMultipleTupleExtractor<MqttMessage, Integer, String>() {
@Override public Map<Integer, String> extract(MqttMessage msg) {
[09/32] ignite git commit: import optimize
Posted by ra...@apache.org.
import optimize
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d928ef4a
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d928ef4a
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d928ef4a
Branch: refs/heads/feature/ignite-535-mqtt
Commit: d928ef4ae0ae7fadbb05dacfc97ca6402657894d
Parents: f8577ca
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Wed Sep 16 02:28:35 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Wed Sep 16 02:28:35 2015 +0300
----------------------------------------------------------------------
.../distributed/dht/GridDhtLocalPartition.java | 26 ++++++++++----------
1 file changed, 13 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/d928ef4a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
index 956f2bf..2deabfe 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
@@ -17,18 +17,6 @@
package org.apache.ignite.internal.processors.cache.distributed.dht;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.atomic.AtomicStampedReference;
-import java.util.concurrent.locks.ReentrantLock;
-import javax.cache.CacheException;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.IgniteInternalFuture;
@@ -58,6 +46,18 @@ import org.jetbrains.annotations.NotNull;
import org.jsr166.ConcurrentHashMap8;
import org.jsr166.LongAdder8;
+import javax.cache.CacheException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicStampedReference;
+import java.util.concurrent.locks.ReentrantLock;
+
import static org.apache.ignite.IgniteSystemProperties.IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE;
import static org.apache.ignite.events.EventType.EVT_CACHE_REBALANCE_OBJECT_UNLOADED;
import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.EVICTED;
@@ -787,4 +787,4 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
"createTime", U.format(createTime),
"mapPubSize", mapPubSize);
}
-}
\ No newline at end of file
+}
[05/32] ignite git commit: Merge remote-tracking branch
'origin/ignite-1.4' into ignite-1.4
Posted by ra...@apache.org.
Merge remote-tracking branch 'origin/ignite-1.4' into ignite-1.4
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/cb7d972e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/cb7d972e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/cb7d972e
Branch: refs/heads/feature/ignite-535-mqtt
Commit: cb7d972e6dfe7db70ac42a9a1889a661a2c26819
Parents: 126e68b 6bd0f8f
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Sep 15 17:03:19 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Sep 15 17:03:19 2015 +0300
----------------------------------------------------------------------
.../configuration/NearCacheConfiguration.java | 18 ++++++-
...CacheScanPartitionQueryFallbackSelfTest.java | 54 ++++++++------------
2 files changed, 37 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
[11/32] ignite git commit: ignite-1171 Test for reproducing added
Posted by ra...@apache.org.
ignite-1171 Test for reproducing added
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/e75917d4
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/e75917d4
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/e75917d4
Branch: refs/heads/feature/ignite-535-mqtt
Commit: e75917d47aaa34d8ff6219245668959fb68f0227
Parents: 5f5220a
Author: agura <ag...@gridgain.com>
Authored: Wed Sep 16 18:23:16 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Wed Sep 16 18:23:16 2015 +0300
----------------------------------------------------------------------
.../distributed/CacheAffEarlySelfTest.java | 245 +++++++++++++++++++
.../ignite/testsuites/IgniteBasicTestSuite.java | 5 +
2 files changed, 250 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/e75917d4/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheAffEarlySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheAffEarlySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheAffEarlySelfTest.java
new file mode 100644
index 0000000..7f0ca11
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheAffEarlySelfTest.java
@@ -0,0 +1,245 @@
+/*
+ * 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.distributed;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Random;
+import java.util.UUID;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgniteFutureTimeoutException;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.marshaller.optimized.OptimizedMarshaller;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class CacheAffEarlySelfTest extends GridCommonAbstractTest {
+ /** Grid count. */
+ private static int GRID_CNT = 8;
+
+ /** Operation timeout. */
+ private static long OP_TIMEOUT = 5000;
+
+ /** Always dump threads or only once per operation. */
+ private static boolean ALWAYS_DUMP_THREADS = false;
+
+ /** Stopped. */
+ private volatile boolean stopped;
+
+ /** Iteration. */
+ private int iters = 10;
+
+ /** Concurrent. */
+ private boolean concurrent = true;
+
+ /** Futs. */
+ private Collection<IgniteInternalFuture<?>> futs = new ArrayList<>(GRID_CNT);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TcpDiscoveryVmIpFinder finder = new TcpDiscoveryVmIpFinder(true);
+ finder.setAddresses(Collections.singletonList("127.0.0.1:47500..47510"));
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+ discoSpi.setIpFinder(finder);
+
+ cfg.setDiscoverySpi(discoSpi);
+
+ OptimizedMarshaller marsh = new OptimizedMarshaller();
+ marsh.setRequireSerializable(false);
+
+ cfg.setMarshaller(marsh);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return 6 * 60 * 1000L;
+ }
+
+ /**
+ *
+ */
+ public void testStartNodes() throws Exception {
+ for (int i = 0; i < iters; i++) {
+ try {
+ System.out.println("*** Iteration " + (i + 1) + '/' + iters);
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(new Runnable() {
+ @Override public void run() {
+ try {
+ doTest();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }, 1);
+
+ fut.get(30000);
+ }
+ catch (IgniteFutureTimeoutCheckedException e) {
+ // No-op.
+ }
+ finally {
+ stopAllGrids(true);
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ public void doTest() throws Exception {
+ for (int i = 0; i < GRID_CNT; i++) {
+ final int idx = i;
+
+ final Ignite grid = concurrent ? null : startGrid(idx);
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(new Runnable() {
+ @Override public void run() {
+ Random rnd = new Random();
+
+ try {
+ final Ignite ignite = grid == null ? startGrid(idx) : grid;
+
+ final IgniteCache<UUID, UUID> cache = getCache(ignite).withAsync();
+
+ CacheAffEarlySelfTest.this.execute(cache, new IgniteInClosure<IgniteCache<UUID,UUID>>() {
+ @Override public void apply(IgniteCache<UUID, UUID> entries) {
+ cache.put(ignite.cluster().localNode().id(), UUID.randomUUID());
+ }
+ });
+
+ while (!stopped) {
+ int val = Math.abs(rnd.nextInt(100));
+ if (val >= 0 && val < 40)
+ execute(cache, new IgniteInClosure<IgniteCache<UUID, UUID>>() {
+ @Override public void apply(IgniteCache<UUID, UUID> entries) {
+ cache.containsKey(ignite.cluster().localNode().id());
+ }
+ });
+ else if (val >= 40 && val < 80)
+ execute(cache, new IgniteInClosure<IgniteCache<UUID, UUID>>() {
+ @Override public void apply(IgniteCache<UUID, UUID> entries) {
+ cache.get(ignite.cluster().localNode().id());
+ }
+ });
+ else
+ execute(cache, new IgniteInClosure<IgniteCache<UUID, UUID>>() {
+ @Override public void apply(IgniteCache<UUID, UUID> entries) {
+ cache.put(ignite.cluster().localNode().id(), UUID.randomUUID());
+ }
+ });
+
+ Thread.sleep(50);
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }, 1);
+
+ futs.add(fut);
+ }
+
+ Thread.sleep(10000);
+
+ stopped = true;
+
+ for (IgniteInternalFuture<?> fut : futs)
+ fut.get();
+ }
+
+ /**
+ * @param cache Cache.
+ * @param c Closure.
+ */
+ private void execute(IgniteCache<UUID, UUID> cache, IgniteInClosure<IgniteCache<UUID, UUID>> c) {
+ c.apply(cache);
+
+ IgniteFuture<Object> fut = cache.future();
+
+ boolean success = false;
+
+ int iter = 0;
+
+ while (!success) {
+ try {
+ fut.get(OP_TIMEOUT);
+
+ success = true;
+ }
+ catch (IgniteFutureTimeoutException e) {
+ debug(iter == 0 || ALWAYS_DUMP_THREADS);
+ }
+
+ iter++;
+ }
+ }
+
+ /**
+ *
+ */
+ private void debug(boolean dumpThreads) {
+ log.info("DUMPING DEBUG INFO:");
+
+ for (Ignite ignite : G.allGrids())
+ ((IgniteKernal)ignite).dumpDebugInfo();
+
+ if (dumpThreads) {
+ U.dumpThreads(null);
+
+ U.dumpThreads(log);
+ }
+ }
+
+ /**
+ * @param grid Grid.
+ */
+ private IgniteCache<UUID, UUID> getCache(Ignite grid) {
+ CacheConfiguration<UUID, UUID> ccfg = defaultCacheConfiguration();
+
+ ccfg.setCacheMode(CacheMode.PARTITIONED);
+ ccfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
+ ccfg.setBackups(1);
+ ccfg.setNearConfiguration(null);
+
+ return grid.getOrCreateCache(ccfg);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/e75917d4/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
index 1a9913f..390807c 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
@@ -36,6 +36,7 @@ import org.apache.ignite.internal.processors.affinity.GridAffinityProcessorRende
import org.apache.ignite.internal.processors.cache.GridProjectionForCachesOnDaemonNodeSelfTest;
import org.apache.ignite.internal.processors.cache.IgniteDaemonNodeMarshallerCacheTest;
import org.apache.ignite.internal.processors.cache.OffHeapTieredTransactionSelfTest;
+import org.apache.ignite.internal.processors.cache.distributed.CacheAffEarlySelfTest;
import org.apache.ignite.internal.processors.closure.GridClosureProcessorSelfTest;
import org.apache.ignite.internal.processors.continuous.GridEventConsumeSelfTest;
import org.apache.ignite.internal.processors.continuous.GridMessageListenSelfTest;
@@ -68,6 +69,9 @@ public class IgniteBasicTestSuite extends TestSuite {
public static TestSuite suite(Set<Class> ignoredTests) throws Exception {
TestSuite suite = new TestSuite("Ignite Basic Test Suite");
+ suite.addTest(new TestSuite(CacheAffEarlySelfTest.class));
+
+/*
suite.addTest(IgniteLangSelfTestSuite.suite());
suite.addTest(IgniteUtilSelfTestSuite.suite(ignoredTests));
suite.addTest(IgniteMarshallerSelfTestSuite.suite(ignoredTests));
@@ -110,6 +114,7 @@ public class IgniteBasicTestSuite extends TestSuite {
suite.addTestSuite(OffHeapTieredTransactionSelfTest.class);
suite.addTestSuite(IgniteSlowClientDetectionSelfTest.class);
suite.addTestSuite(IgniteDaemonNodeMarshallerCacheTest.class);
+*/
return suite;
}