You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2016/01/15 13:09:17 UTC
[01/16] ignite git commit: Changed copyright from 2015 to 2016
Repository: ignite
Updated Branches:
refs/heads/ignite-1811 969faefd3 -> caaab6e54
Changed copyright from 2015 to 2016
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2d106303
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2d106303
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2d106303
Branch: refs/heads/ignite-1811
Commit: 2d106303ab406aec98265b0ae2bd07c7862d0f23
Parents: 7175a42
Author: Denis Magda <dm...@gridgain.com>
Authored: Thu Jan 14 10:17:07 2016 +0300
Committer: Denis Magda <dm...@gridgain.com>
Committed: Thu Jan 14 10:17:07 2016 +0300
----------------------------------------------------------------------
.../main/java/org/apache/ignite/internal/IgniteVersionUtils.java | 4 ++--
.../org/apache/ignite/startup/GridRandomCommandLineLoader.java | 4 ++--
2 files changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d106303/modules/core/src/main/java/org/apache/ignite/internal/IgniteVersionUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteVersionUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteVersionUtils.java
index 02cbc81..bd8726f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteVersionUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteVersionUtils.java
@@ -47,7 +47,7 @@ public class IgniteVersionUtils {
public static final String ACK_VER_STR;
/** Copyright blurb. */
- public static final String COPYRIGHT = "2015 Copyright(C) Apache Software Foundation";
+ public static final String COPYRIGHT = "2016 Copyright(C) Apache Software Foundation";
/**
* Static initializer.
@@ -77,4 +77,4 @@ public class IgniteVersionUtils {
private IgniteVersionUtils() {
// No-op.
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d106303/modules/core/src/test/java/org/apache/ignite/startup/GridRandomCommandLineLoader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/startup/GridRandomCommandLineLoader.java b/modules/core/src/test/java/org/apache/ignite/startup/GridRandomCommandLineLoader.java
index 20d1b32..13c38ef 100644
--- a/modules/core/src/test/java/org/apache/ignite/startup/GridRandomCommandLineLoader.java
+++ b/modules/core/src/test/java/org/apache/ignite/startup/GridRandomCommandLineLoader.java
@@ -61,7 +61,7 @@ public final class GridRandomCommandLineLoader {
private static final String IGNITE_PROG_NAME = "IGNITE_PROG_NAME";
/** Copyright text. Ant processed. */
- private static final String COPYRIGHT = "2015 Copyright(C) Apache Software Foundation.";
+ private static final String COPYRIGHT = "2016 Copyright(C) Apache Software Foundation.";
/** Version. Ant processed. */
private static final String VER = "x.x.x";
@@ -408,4 +408,4 @@ public final class GridRandomCommandLineLoader {
return options;
}
-}
\ No newline at end of file
+}
[05/16] ignite git commit: Added IgniteGetTxBenchmark (cache 'get'
using transactional cache).
Posted by sb...@apache.org.
Added IgniteGetTxBenchmark (cache 'get' using transactional cache).
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d608ebdf
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d608ebdf
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d608ebdf
Branch: refs/heads/ignite-1811
Commit: d608ebdfd3f9088255dd6e32de6e90e1564569b1
Parents: 8b0c59a
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jan 15 10:06:02 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jan 15 10:06:02 2016 +0300
----------------------------------------------------------------------
.../yardstick/cache/IgniteGetTxBenchmark.java | 30 ++++++++++++++++++++
1 file changed, 30 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/d608ebdf/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteGetTxBenchmark.java
----------------------------------------------------------------------
diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteGetTxBenchmark.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteGetTxBenchmark.java
new file mode 100644
index 0000000..fbb73e2
--- /dev/null
+++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteGetTxBenchmark.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.yardstick.cache;
+
+import org.apache.ignite.IgniteCache;
+
+/**
+ *
+ */
+public class IgniteGetTxBenchmark extends IgniteGetBenchmark {
+ /** {@inheritDoc} */
+ @Override protected IgniteCache<Integer, Object> cache() {
+ return ignite().cache("tx");
+ }
+}
[13/16] ignite git commit: ignite-1811
Posted by sb...@apache.org.
ignite-1811
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/56b9da06
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/56b9da06
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/56b9da06
Branch: refs/heads/ignite-1811
Commit: 56b9da06314598e6fe1bb41d8a86de241e3b3095
Parents: a728ecc
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jan 15 14:08:59 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jan 15 14:08:59 2016 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheContext.java | 19 +-
.../dht/GridPartitionedGetFuture.java | 2 +-
.../dht/GridPartitionedSingleGetFuture.java | 2 +-
.../distributed/near/GridNearGetFuture.java | 2 +-
.../internal/TestRecordingCommunicationSpi.java | 82 ++++-
...idCacheConfigurationConsistencySelfTest.java | 58 +---
...niteCacheClientNodeChangingTopologyTest.java | 4 +-
.../IgniteCacheReadFromBackupTest.java | 343 ++++++++++++++++++-
.../IgniteCrossCacheTxStoreSelfTest.java | 1 +
.../near/GridCacheGetStoreErrorSelfTest.java | 9 +-
10 files changed, 440 insertions(+), 82 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/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 df72774..fc48b9d 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
@@ -1435,6 +1435,13 @@ public class GridCacheContext<K, V> implements Externalizable {
}
/**
+ * @return {@code True} if store and read-through mode are enabled in configuration.
+ */
+ public boolean readThroughConfigured() {
+ return store().configured() && cacheCfg.isReadThrough();
+ }
+
+ /**
* @return {@code True} if {@link CacheConfiguration#isLoadPreviousValue()} flag is set.
*/
public boolean loadPreviousValue() {
@@ -1981,16 +1988,10 @@ public class GridCacheContext<K, V> implements Externalizable {
private boolean hasPartition(int part, List<ClusterNode> affNodes, AffinityTopologyVersion topVer) {
assert affinityNode();
- return (topology().rebalanceFinished(topVer) && (isReplicated() || affNodes.contains(locNode)))
- || partitionOwned(part);
- }
+ GridDhtPartitionTopology top = topology();
- /**
- * @param part Partition.
- * @return {@code True} if partition is in owned state.
- */
- private boolean partitionOwned(int part) {
- return topology().partitionState(localNodeId(), part) == OWNING;
+ return (top.rebalanceFinished(topVer) && (isReplicated() || affNodes.contains(locNode)))
+ || (top.partitionState(localNodeId(), part) == OWNING);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index be50384..2bc6869 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -507,7 +507,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
boolean topStable = cctx.isReplicated() || topVer.equals(cctx.topology().topologyVersion());
// Entry not found, do not continue search if topology did not change and there is no store.
- if (!cctx.store().configured() && (topStable || partitionOwned(part))) {
+ if (!cctx.readThroughConfigured() && (topStable || partitionOwned(part))) {
if (!skipVals && cctx.config().isStatisticsEnabled())
cache.metrics0().onRead(false);
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index ba14151..96c20e6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -427,7 +427,7 @@ public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> im
boolean topStable = cctx.isReplicated() || topVer.equals(cctx.topology().topologyVersion());
// Entry not found, complete future with null result if topology did not change and there is no store.
- if (!cctx.store().configured() && (topStable || partitionOwned(part))) {
+ if (!cctx.readThroughConfigured() && (topStable || partitionOwned(part))) {
if (!skipVals && cctx.config().isStatisticsEnabled())
colocated.metrics0().onRead(false);
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 4802539..12a1b74 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -620,7 +620,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
boolean topStable = cctx.isReplicated() || topVer.equals(cctx.topology().topologyVersion());
// Entry not found, do not continue search if topology did not change and there is no store.
- return !cctx.store().configured() && (topStable || partitionOwned(part));
+ return !cctx.readThroughConfigured() && (topStable || partitionOwned(part));
}
}
catch (GridCacheEntryRemovedException ignored) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
index bf84387..8a602ad 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
@@ -18,16 +18,24 @@
package org.apache.ignite.internal;
import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_GRID_NAME;
+
/**
*
*/
@@ -38,15 +46,46 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
/** */
private List<Object> recordedMsgs = new ArrayList<>();
+ /** */
+ private List<T2<ClusterNode, GridIoMessage>> blockedMsgs = new ArrayList<>();
+
+ /** */
+ private Map<Class<?>, Set<String>> blockCls = new HashMap<>();
+
+ /** */
+ private IgnitePredicate<GridIoMessage> blockP;
+
/** {@inheritDoc} */
@Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
throws IgniteSpiException {
if (msg instanceof GridIoMessage) {
- Object msg0 = ((GridIoMessage)msg).message();
+ GridIoMessage ioMsg = (GridIoMessage)msg;
+
+ Object msg0 = ioMsg.message();
synchronized (this) {
if (recordCls != null && msg0.getClass().equals(recordCls))
recordedMsgs.add(msg0);
+
+ boolean block = false;
+
+ if (blockP != null && blockP.apply(ioMsg))
+ block = true;
+ else {
+ Set<String> blockNodes = blockCls.get(msg0.getClass());
+
+ if (blockNodes != null) {
+ String nodeName = (String)node.attributes().get(ATTR_GRID_NAME);
+
+ block = blockNodes.contains(nodeName);
+ }
+ }
+
+ if (block) {
+ blockedMsgs.add(new T2<>(node, ioMsg));
+
+ return;
+ }
}
}
@@ -74,4 +113,45 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
return msgs;
}
}
+
+ /**
+ * @param blockP Message block predicate.
+ */
+ public void blockMessages(IgnitePredicate<GridIoMessage> blockP) {
+ synchronized (this) {
+ this.blockP = blockP;
+ }
+ }
+
+ /**
+ * @param cls Message class.
+ * @param nodeName Node name.
+ */
+ public void blockMessages(Class<?> cls, String nodeName) {
+ synchronized (this) {
+ Set<String> set = blockCls.get(cls);
+
+ if (set == null) {
+ set = new HashSet<>();
+
+ blockCls.put(cls, set);
+ }
+
+ set.add(nodeName);
+ }
+ }
+
+ /**
+ * Stops block messages and sends all already blocked messages.
+ */
+ public void stopBlock() {
+ synchronized (this) {
+ blockCls.clear();
+
+ for (T2<ClusterNode, GridIoMessage> msg : blockedMsgs)
+ super.sendMessage(msg.get1(), msg.get2());
+
+ blockedMsgs.clear();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
index e28e89f..a1f917f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
@@ -19,25 +19,19 @@ package org.apache.ignite.internal.processors.cache;
import java.io.Externalizable;
import java.io.Serializable;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Map;
import java.util.concurrent.Callable;
import javax.cache.Cache;
-import javax.cache.integration.CacheLoaderException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheInterceptorAdapter;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.affinity.AffinityFunction;
import org.apache.ignite.cache.affinity.AffinityNodeIdHashResolver;
-import org.apache.ignite.cache.affinity.fair.FairAffinityFunction;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cache.eviction.EvictionFilter;
import org.apache.ignite.cache.eviction.fifo.FifoEvictionPolicy;
import org.apache.ignite.cache.eviction.lru.LruEvictionPolicy;
import org.apache.ignite.cache.eviction.random.RandomEvictionPolicy;
-import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DeploymentMode;
@@ -46,7 +40,6 @@ import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
@@ -54,7 +47,6 @@ import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.GridStringLogger;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
@@ -862,49 +854,9 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
}, IgniteCheckedException.class, null);
}
- /** */
- private static class TestStore implements CacheStore<Object,Object> {
- /** {@inheritDoc} */
- @Nullable @Override public Object load(Object key) {
- return null;
- }
-
- /** {@inheritDoc} */
- @Override public void loadCache(IgniteBiInClosure<Object, Object> clo, @Nullable Object... args) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public Map<Object, Object> loadAll(Iterable<?> keys) throws CacheLoaderException {
- return Collections.emptyMap();
- }
-
- /** {@inheritDoc} */
- @Override public void write(Cache.Entry<?, ?> entry) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void writeAll(Collection<Cache.Entry<?, ?>> entries) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void delete(Object key) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void deleteAll(Collection<?> keys) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public void sessionEnd(boolean commit) {
- // No-op.
- }
- }
-
+ /**
+ *
+ */
private static class TestRendezvousAffinityFunction extends RendezvousAffinityFunction {
/**
* Empty constructor required by {@link Externalizable}.
@@ -941,6 +893,10 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
// No-op, just different class.
}
+ /**
+ *
+ */
private static class TestCacheDefaultAffinityKeyMapper extends GridCacheDefaultAffinityKeyMapper {
+ // No-op, just different class.
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
index e7657a6..13f2598 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
@@ -2010,7 +2010,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
private List<Object> recordedMsgs = new ArrayList<>();
/** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackClosure)
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
throws IgniteSpiException {
if (msg instanceof GridIoMessage) {
Object msg0 = ((GridIoMessage)msg).message();
@@ -2032,7 +2032,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
}
}
- super.sendMessage(node, msg, ackClosure);
+ super.sendMessage(node, msg, ackC);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
index ee72909..af018cc 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
@@ -17,17 +17,42 @@
package org.apache.ignite.internal.processors.cache.distributed;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.cache.Cache;
+import javax.cache.configuration.Factory;
+import javax.cache.integration.CacheLoaderException;
+import javax.cache.integration.CacheWriterException;
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.cache.store.CacheStore;
+import org.apache.ignite.cache.store.CacheStoreAdapter;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessageV2;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.lang.IgnitePredicate;
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.testframework.junits.common.GridCommonAbstractTest;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+
/**
*
*/
@@ -50,6 +75,7 @@ public class IgniteCacheReadFromBackupTest extends GridCommonAbstractTest {
return cfg;
}
+
/** {@inheritDoc} */
@Override protected void beforeTestsStarted() throws Exception {
super.beforeTestsStarted();
@@ -67,48 +93,335 @@ public class IgniteCacheReadFromBackupTest extends GridCommonAbstractTest {
/**
* @throws Exception If failed.
*/
+ public void testGetFromBackupStoreReadThroughEnabled() throws Exception {
+ for (CacheConfiguration<Object, Object> ccfg : cacheConfigurations()) {
+ ccfg.setCacheStoreFactory(new TestStoreFactory());
+ ccfg.setReadThrough(true);
+
+ boolean near = (ccfg.getNearConfiguration() != null);
+
+ log.info("Test cache [mode=" + ccfg.getCacheMode() +
+ ", atomicity=" + ccfg.getAtomicityMode() +
+ ", backups=" + ccfg.getBackups() +
+ ", near=" + near + "]");
+
+ ignite(0).createCache(ccfg);
+
+ awaitPartitionMapExchange();
+
+ try {
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ log.info("Check node: " + ignite.name());
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+
+ TestRecordingCommunicationSpi spi = recordGetRequests(ignite, near);
+
+ Integer key = backupKey(cache);
+
+ assertNull(cache.get(key));
+
+ List<Object> msgs = spi.recordedMessages();
+
+ assertEquals(1, msgs.size());
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetFromBackupStoreReadThroughDisabled() throws Exception {
+ for (CacheConfiguration<Object, Object> ccfg : cacheConfigurations()) {
+ ccfg.setCacheStoreFactory(new TestStoreFactory());
+ ccfg.setReadThrough(false);
+
+ boolean near = (ccfg.getNearConfiguration() != null);
+
+ log.info("Test cache [mode=" + ccfg.getCacheMode() +
+ ", atomicity=" + ccfg.getAtomicityMode() +
+ ", backups=" + ccfg.getBackups() +
+ ", near=" + near + "]");
+
+ ignite(0).createCache(ccfg);
+
+ awaitPartitionMapExchange();
+
+ try {
+ checkLocalRead(NODES, ccfg);
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetFromPrimaryPreloadInProgress() throws Exception {
+ for (final CacheConfiguration<Object, Object> ccfg : cacheConfigurations()) {
+ boolean near = (ccfg.getNearConfiguration() != null);
+
+ log.info("Test cache [mode=" + ccfg.getCacheMode() +
+ ", atomicity=" + ccfg.getAtomicityMode() +
+ ", backups=" + ccfg.getBackups() +
+ ", near=" + near + "]");
+
+ ignite(0).createCache(ccfg);
+
+ awaitPartitionMapExchange();
+
+ try {
+ Map<Ignite, Integer> backupKeys = new HashMap<>();
+ Map<Ignite, Integer> nearKeys = new HashMap<>();
+
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+
+ backupKeys.put(ignite, backupKey(cache));
+
+ if (ccfg.getCacheMode() == PARTITIONED)
+ nearKeys.put(ignite, nearKey(cache));
+
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ spi.blockMessages(new IgnitePredicate<GridIoMessage>() {
+ @Override public boolean apply(GridIoMessage ioMsg) {
+ if (!ioMsg.message().getClass().equals(GridDhtPartitionSupplyMessageV2.class))
+ return false;
+
+ GridDhtPartitionSupplyMessageV2 msg = (GridDhtPartitionSupplyMessageV2)ioMsg.message();
+
+ return msg.cacheId() == CU.cacheId(ccfg.getName());
+ }
+ });
+ }
+
+ try (Ignite newNode = startGrid(NODES)) {
+ IgniteCache<Integer, Integer> cache = newNode.cache(ccfg.getName());
+
+ TestRecordingCommunicationSpi newNodeSpi = recordGetRequests(newNode, near);
+
+ Integer key = backupKey(cache);
+
+ assertNull(cache.get(key));
+
+ List<Object> msgs = newNodeSpi.recordedMessages();
+
+ assertEquals(1, msgs.size());
+
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ log.info("Check node: " + ignite.name());
+
+ checkLocalRead(ignite, ccfg, backupKeys.get(ignite), nearKeys.get(ignite));
+ }
+
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ spi.stopBlock();
+ }
+
+ awaitPartitionMapExchange();
+
+ checkLocalRead(NODES + 1, ccfg);
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testNoPrimaryReadPreloadFinished() throws Exception {
+ for (CacheConfiguration<Object, Object> ccfg : cacheConfigurations()) {
+ boolean near = (ccfg.getNearConfiguration() != null);
+
+ log.info("Test cache [mode=" + ccfg.getCacheMode() +
+ ", atomicity=" + ccfg.getAtomicityMode() +
+ ", backups=" + ccfg.getBackups() +
+ ", near=" + near + "]");
+
+ ignite(0).createCache(ccfg);
+
+ awaitPartitionMapExchange();
+
+ try {
+ checkLocalRead(NODES, ccfg);
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
}
/**
+ * @param nodes Number of nodes.
* @param ccfg Cache configuration.
* @throws Exception If failed.
*/
- private void checkNoPrimaryRead(CacheConfiguration<Object, Object> ccfg) throws Exception {
- ignite(0).createCache(ccfg);
+ private void checkLocalRead(int nodes, CacheConfiguration<Object, Object> ccfg) throws Exception {
+ for (int i = 0; i < nodes; i++) {
+ Ignite ignite = ignite(i);
- try {
- for (int i = 0; i < NODES; i++) {
- Ignite ignite = ignite(i);
+ log.info("Check node: " + ignite.name());
- IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
- TestRecordingCommunicationSpi spi =
- (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+ List<Integer> backupKeys = backupKeys(cache, 2, 0);
- Integer key = nearKey(cache);
+ Integer backupKey = backupKeys.get(0);
- assertNull(cache.get(key));
- }
+ Integer nearKey = ccfg.getCacheMode() == PARTITIONED ? nearKey(cache) : null;
+
+ checkLocalRead(ignite, ccfg, backupKey, nearKey);
+
+ Set<Integer> keys = new HashSet<>(backupKeys);
+
+ Map<Integer, Integer> vals = cache.getAll(keys);
+
+ for (Integer key : keys)
+ assertNull(vals.get(key));
+
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ List<Object> msgs = spi.recordedMessages();
+
+ assertEquals(0, msgs.size());
}
- finally {
- ignite(0).destroyCache(ccfg.getName());
+ }
+
+ /**
+ * @param ignite Node.
+ * @param ccfg Cache configuration.
+ * @param backupKey Backup key.
+ * @param nearKey Near key.
+ * @throws Exception If failed.
+ */
+ private void checkLocalRead(Ignite ignite,
+ CacheConfiguration<Object, Object> ccfg,
+ Integer backupKey,
+ Integer nearKey) throws Exception {
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+
+ TestRecordingCommunicationSpi spi = recordGetRequests(ignite, ccfg.getNearConfiguration() != null);
+
+ List<Object> msgs;
+
+ if (nearKey != null) {
+ assertNull(cache.get(nearKey));
+
+ msgs = spi.recordedMessages();
+
+ assertEquals(1, msgs.size());
}
+
+ assertNull(cache.get(backupKey));
+
+ msgs = spi.recordedMessages();
+
+ assertTrue(msgs.isEmpty());
+ }
+
+ /**
+ * @param ignite Node.
+ * @param near Near cache flag.
+ * @return Communication SPI.
+ */
+ private TestRecordingCommunicationSpi recordGetRequests(Ignite ignite, boolean near) {
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ spi.record(near ? GridNearGetRequest.class : GridNearSingleGetRequest.class);
+
+ return spi;
+ }
+
+ /**
+ * @return Cache configurations to test.
+ */
+ private List<CacheConfiguration<Object, Object>> cacheConfigurations() {
+ List<CacheConfiguration<Object, Object>> ccfgs = new ArrayList<>();
+
+ ccfgs.add(cacheConfiguration(REPLICATED, ATOMIC, 0, false));
+ ccfgs.add(cacheConfiguration(REPLICATED, TRANSACTIONAL, 0, false));
+
+ ccfgs.add(cacheConfiguration(PARTITIONED, ATOMIC, 1, false));
+ ccfgs.add(cacheConfiguration(PARTITIONED, ATOMIC, 1, true));
+ ccfgs.add(cacheConfiguration(PARTITIONED, ATOMIC, 2, false));
+
+ ccfgs.add(cacheConfiguration(PARTITIONED, TRANSACTIONAL, 1, false));
+ ccfgs.add(cacheConfiguration(PARTITIONED, TRANSACTIONAL, 1, true));
+ ccfgs.add(cacheConfiguration(PARTITIONED, TRANSACTIONAL, 2, false));
+
+ return ccfgs;
}
/**
* @param cacheMode Cache mode.
+ * @param atomicityMode Cache atomicity mode.
* @param backups Number of backups.
+ * @param nearEnabled {@code True} if near cache should be enabled.
* @return Cache configuration.
*/
- private CacheConfiguration<Object, Object> cacheConfiguration(CacheMode cacheMode, int backups) {
+ private CacheConfiguration<Object, Object> cacheConfiguration(CacheMode cacheMode,
+ CacheAtomicityMode atomicityMode,
+ int backups,
+ boolean nearEnabled) {
CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
ccfg.setCacheMode(cacheMode);
+ ccfg.setAtomicityMode(atomicityMode);
- if (cacheMode != CacheMode.REPLICATED)
+ if (cacheMode != REPLICATED) {
ccfg.setBackups(backups);
+ if (nearEnabled)
+ ccfg.setNearConfiguration(new NearCacheConfiguration<>());
+ }
+
return ccfg;
}
+
+ /**
+ *
+ */
+ private static class TestStoreFactory implements Factory<CacheStore<Object, Object>> {
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public CacheStore<Object, Object> create() {
+ return new CacheStoreAdapter() {
+ @Override public Object load(Object key) throws CacheLoaderException {
+ return null;
+ }
+
+ @Override public void write(Cache.Entry entry) throws CacheWriterException {
+ // No-op.
+ }
+
+ @Override public void delete(Object key) throws CacheWriterException {
+ // No-op.
+ }
+ };
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
index 68cac17..94613db 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
@@ -300,6 +300,7 @@ public class IgniteCrossCacheTxStoreSelfTest extends GridCommonAbstractTest {
throws CacheLoaderException {
}
+ /** {@inheritDoc} */
@Override public void sessionEnd(boolean commit) throws CacheWriterException {
evts.offer("sessionEnd " + commit);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56b9da06/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java
index 7bd845a..3e6a245 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheGetStoreErrorSelfTest.java
@@ -107,7 +107,11 @@ public class GridCacheGetStoreErrorSelfTest extends GridCommonAbstractTest {
checkGetError(false, LOCAL);
}
- /** @throws Exception If failed. */
+ /**
+ * @param nearEnabled Near cache flag.
+ * @param cacheMode Cache mode.
+ * @throws Exception If failed.
+ */
private void checkGetError(boolean nearEnabled, CacheMode cacheMode) throws Exception {
this.nearEnabled = nearEnabled;
this.cacheMode = cacheMode;
@@ -147,14 +151,17 @@ public class GridCacheGetStoreErrorSelfTest extends GridCommonAbstractTest {
*/
@SuppressWarnings("PublicInnerClass")
public static class TestStore extends CacheStoreAdapter<Object, Object> {
+ /** {@inheritDoc} */
@Override public Object load(Object key) {
throw new IgniteException("Failed to get key from store: " + key);
}
+ /** {@inheritDoc} */
@Override public void write(Cache.Entry<?, ?> entry) {
// No-op.
}
+ /** {@inheritDoc} */
@Override public void delete(Object key) {
// No-op.
}
[06/16] ignite git commit: IGNITE-2032 Unwind undeploys in preloader
- Fixes #369.
Posted by sb...@apache.org.
IGNITE-2032 Unwind undeploys in preloader - Fixes #369.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2af1d9bc
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2af1d9bc
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2af1d9bc
Branch: refs/heads/ignite-1811
Commit: 2af1d9bcb6c89fc94c54c2b6c02a922241a6896e
Parents: d608ebd
Author: vershov <ve...@gridgain.com>
Authored: Thu Jan 14 17:21:56 2016 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Fri Jan 15 10:27:05 2016 +0300
----------------------------------------------------------------------
.../dht/preloader/GridDhtPartitionsExchangeFuture.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2af1d9bc/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 a10294f..22fb59e 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
@@ -569,6 +569,9 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
}
else
cacheCtx.affinity().clientEventTopologyChange(discoEvt, exchId.topologyVersion());
+
+ if (!exchId.isJoined())
+ cacheCtx.preloader().unwindUndeploys();
}
if (exchId.isLeft())
@@ -845,8 +848,9 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
// Partition release future is done so we can flush the write-behind store.
cacheCtx.store().forceFlush();
- // Process queued undeploys prior to sending/spreading map.
- cacheCtx.preloader().unwindUndeploys();
+ if (!exchId.isJoined())
+ // Process queued undeploys prior to sending/spreading map.
+ cacheCtx.preloader().unwindUndeploys();
GridDhtPartitionTopology top = cacheCtx.topology();
[11/16] ignite git commit: Renamed fields to change fields write
order (to preserve backward compatibility).
Posted by sb...@apache.org.
Renamed fields to change fields write order (to preserve backward compatibility).
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/fa427dce
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/fa427dce
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/fa427dce
Branch: refs/heads/ignite-1811
Commit: fa427dce25061b3c85be04f83debb452e60f08a2
Parents: c4ff711
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jan 15 11:17:30 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jan 15 11:17:30 2016 +0300
----------------------------------------------------------------------
.../continuous/CacheContinuousQueryEntry.java | 70 ++++++++++----------
1 file changed, 35 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/fa427dce/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
index 4d3786a..bcc2576 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java
@@ -84,17 +84,17 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
private GridDeploymentInfo depInfo;
/** Partition. */
- private int part;
+ private int _part;
/** Update counter. */
- private long updateCntr;
+ private long _updateCntr;
/** Flags. */
private byte flags;
/** */
@GridToStringInclude
- private AffinityTopologyVersion topVer;
+ private AffinityTopologyVersion _topVer;
/** Filtered events. */
private GridLongList filteredEvts;
@@ -134,9 +134,9 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
this.key = key;
this.newVal = newVal;
this.oldVal = oldVal;
- this.part = part;
- this.updateCntr = updateCntr;
- this.topVer = topVer;
+ this._part = part;
+ this._updateCntr = updateCntr;
+ this._topVer = topVer;
this.keepBinary = keepBinary;
}
@@ -144,7 +144,7 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
* @return Topology version if applicable.
*/
@Nullable AffinityTopologyVersion topologyVersion() {
- return topVer;
+ return _topVer;
}
/**
@@ -165,14 +165,14 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
* @return Partition.
*/
int partition() {
- return part;
+ return _part;
}
/**
* @return Update counter.
*/
long updateCounter() {
- return updateCntr;
+ return _updateCntr;
}
/**
@@ -310,67 +310,67 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
switch (writer.state()) {
case 0:
- if (!writer.writeInt("cacheId", cacheId))
+ if (!writer.writeInt("_part", _part))
return false;
writer.incrementState();
case 1:
- if (!writer.writeByte("evtType", evtType != null ? (byte)evtType.ordinal() : -1))
+ if (!writer.writeMessage("_topVer", _topVer))
return false;
writer.incrementState();
case 2:
- if (!writer.writeMessage("filteredEvts", filteredEvts))
+ if (!writer.writeLong("_updateCntr", _updateCntr))
return false;
writer.incrementState();
case 3:
- if (!writer.writeByte("flags", flags))
+ if (!writer.writeInt("cacheId", cacheId))
return false;
writer.incrementState();
case 4:
- if (!writer.writeBoolean("keepBinary", keepBinary))
+ if (!writer.writeByte("evtType", evtType != null ? (byte)evtType.ordinal() : -1))
return false;
writer.incrementState();
case 5:
- if (!writer.writeMessage("key", key))
+ if (!writer.writeMessage("filteredEvts", filteredEvts))
return false;
writer.incrementState();
case 6:
- if (!writer.writeMessage("newVal", newVal))
+ if (!writer.writeByte("flags", flags))
return false;
writer.incrementState();
case 7:
- if (!writer.writeMessage("oldVal", oldVal))
+ if (!writer.writeBoolean("keepBinary", keepBinary))
return false;
writer.incrementState();
case 8:
- if (!writer.writeInt("part", part))
+ if (!writer.writeMessage("key", key))
return false;
writer.incrementState();
case 9:
- if (!writer.writeMessage("topVer", topVer))
+ if (!writer.writeMessage("newVal", newVal))
return false;
writer.incrementState();
case 10:
- if (!writer.writeLong("updateCntr", updateCntr))
+ if (!writer.writeMessage("oldVal", oldVal))
return false;
writer.incrementState();
@@ -389,7 +389,7 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
switch (reader.state()) {
case 0:
- cacheId = reader.readInt("cacheId");
+ _part = reader.readInt("_part");
if (!reader.isLastRead())
return false;
@@ -397,19 +397,15 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
reader.incrementState();
case 1:
- byte evtTypeOrd;
-
- evtTypeOrd = reader.readByte("evtType");
+ _topVer = reader.readMessage("_topVer");
if (!reader.isLastRead())
return false;
- evtType = eventTypeFromOrdinal(evtTypeOrd);
-
reader.incrementState();
case 2:
- filteredEvts = reader.readMessage("filteredEvts");
+ _updateCntr = reader.readLong("_updateCntr");
if (!reader.isLastRead())
return false;
@@ -417,7 +413,7 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
reader.incrementState();
case 3:
- flags = reader.readByte("flags");
+ cacheId = reader.readInt("cacheId");
if (!reader.isLastRead())
return false;
@@ -425,15 +421,19 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
reader.incrementState();
case 4:
- keepBinary = reader.readBoolean("keepBinary");
+ byte evtTypeOrd;
+
+ evtTypeOrd = reader.readByte("evtType");
if (!reader.isLastRead())
return false;
+ evtType = eventTypeFromOrdinal(evtTypeOrd);
+
reader.incrementState();
case 5:
- key = reader.readMessage("key");
+ filteredEvts = reader.readMessage("filteredEvts");
if (!reader.isLastRead())
return false;
@@ -441,7 +441,7 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
reader.incrementState();
case 6:
- newVal = reader.readMessage("newVal");
+ flags = reader.readByte("flags");
if (!reader.isLastRead())
return false;
@@ -449,7 +449,7 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
reader.incrementState();
case 7:
- oldVal = reader.readMessage("oldVal");
+ keepBinary = reader.readBoolean("keepBinary");
if (!reader.isLastRead())
return false;
@@ -457,7 +457,7 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
reader.incrementState();
case 8:
- part = reader.readInt("part");
+ key = reader.readMessage("key");
if (!reader.isLastRead())
return false;
@@ -465,7 +465,7 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
reader.incrementState();
case 9:
- topVer = reader.readMessage("topVer");
+ newVal = reader.readMessage("newVal");
if (!reader.isLastRead())
return false;
@@ -473,7 +473,7 @@ public class CacheContinuousQueryEntry implements GridCacheDeployable, Message {
reader.incrementState();
case 10:
- updateCntr = reader.readLong("updateCntr");
+ oldVal = reader.readMessage("oldVal");
if (!reader.isLastRead())
return false;
[07/16] ignite git commit: IGNITE-2365 - Notify policy if swap or
offheap is enabled and rebalanced entry was not preloaded. IGNITE-2099 -
Fixing custom collections. This closes #396
Posted by sb...@apache.org.
IGNITE-2365 - Notify policy if swap or offheap is enabled and rebalanced entry was not preloaded.
IGNITE-2099 - Fixing custom collections.
This closes #396
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/6524c796
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/6524c796
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/6524c796
Branch: refs/heads/ignite-1811
Commit: 6524c79629f587fb28bc43ddbef973aa2e83f66b
Parents: 2af1d9b
Author: Alexey Goncharuk <al...@gmail.com>
Authored: Wed Jan 13 16:47:32 2016 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Fri Jan 15 10:27:48 2016 +0300
----------------------------------------------------------------------
.../ignite/internal/binary/BinaryUtils.java | 78 ++++++++
.../processors/cache/CacheObjectContext.java | 91 +++++----
.../binary/CacheObjectBinaryProcessorImpl.java | 33 ++--
.../dht/preloader/GridDhtPartitionDemander.java | 11 +-
.../binary/BinaryMarshallerSelfTest.java | 44 ++++-
.../cache/GridCacheDeploymentSelfTest.java | 3 +-
...IgniteCacheGetCustomCollectionsSelfTest.java | 128 +++++++++++++
...gniteCacheLoadRebalanceEvictionSelfTest.java | 188 +++++++++++++++++++
.../platform/PlatformComputeEchoTask.java | 6 +-
.../testsuites/IgniteCacheTestSuite4.java | 5 +
10 files changed, 515 insertions(+), 72 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
index 62a9d26..a82b65f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
@@ -590,6 +590,43 @@ public class BinaryUtils {
}
/**
+ * @param map Map to check.
+ * @return {@code True} if this map type is supported.
+ */
+ public static boolean knownMap(Object map) {
+ Class<?> cls = map == null ? null : map.getClass();
+
+ return cls == HashMap.class ||
+ cls == LinkedHashMap.class ||
+ cls == TreeMap.class ||
+ cls == ConcurrentHashMap8.class ||
+ cls == ConcurrentHashMap.class;
+ }
+
+ /**
+ * Attempts to create a new map of the same known type. Will return null if map type is not supported.
+ *
+ * @param map Map.
+ * @return New map of the same type or null.
+ */
+ public static <K, V> Map<K, V> newKnownMap(Object map) {
+ Class<?> cls = map == null ? null : map.getClass();
+
+ if (cls == HashMap.class)
+ return U.newHashMap(((Map)map).size());
+ else if (cls == LinkedHashMap.class)
+ return U.newLinkedHashMap(((Map)map).size());
+ else if (cls == TreeMap.class)
+ return new TreeMap<>(((TreeMap<Object, Object>)map).comparator());
+ else if (cls == ConcurrentHashMap8.class)
+ return new ConcurrentHashMap8<>(U.capacity(((Map)map).size()));
+ else if (cls == ConcurrentHashMap.class)
+ return new ConcurrentHashMap<>(U.capacity(((Map)map).size()));
+
+ return null;
+ }
+
+ /**
* Attempts to create a new map of the same type as {@code map} has. Otherwise returns new {@code HashMap} instance.
*
* @param map Original map.
@@ -609,6 +646,47 @@ public class BinaryUtils {
}
/**
+ * @param col Collection to check.
+ * @return True if this is a collection of a known type.
+ */
+ public static boolean knownCollection(Object col) {
+ Class<?> cls = col == null ? null : col.getClass();
+
+ return cls == HashSet.class ||
+ cls == LinkedHashSet.class ||
+ cls == TreeSet.class ||
+ cls == ConcurrentSkipListSet.class ||
+ cls == ArrayList.class ||
+ cls == LinkedList.class;
+ }
+
+ /**
+ * Attempts to create a new collection of the same known type. Will return null if collection type is
+ * unknown.
+ *
+ * @param col Collection.
+ * @return New empty collection.
+ */
+ public static <V> Collection<V> newKnownCollection(Object col) {
+ Class<?> cls = col == null ? null : col.getClass();
+
+ if (cls == HashSet.class)
+ return U.newHashSet(((Collection)col).size());
+ else if (cls == LinkedHashSet.class)
+ return U.newLinkedHashSet(((Collection)col).size());
+ else if (cls == TreeSet.class)
+ return new TreeSet<>(((TreeSet<Object>)col).comparator());
+ else if (cls == ConcurrentSkipListSet.class)
+ return new ConcurrentSkipListSet<>(((ConcurrentSkipListSet<Object>)col).comparator());
+ else if (cls == ArrayList.class)
+ return new ArrayList<>(((Collection)col).size());
+ else if (cls == LinkedList.class)
+ return new LinkedList<>();
+
+ return null;
+ }
+
+ /**
* Attempts to create a new set of the same type as {@code set} has. Otherwise returns new {@code HashSet} instance.
*
* @param set Original set.
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectContext.java
index 7401434..d22bc75 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectContext.java
@@ -19,8 +19,11 @@ package org.apache.ignite.internal.processors.cache;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
import java.util.Map;
-import java.util.Set;
import org.apache.ignite.cache.affinity.AffinityKeyMapper;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.binary.BinaryUtils;
@@ -161,13 +164,25 @@ import org.apache.ignite.internal.util.typedef.F;
* @return Unwrapped collection.
*/
public Collection<Object> unwrapBinariesIfNeeded(Collection<Object> col, boolean keepBinary, boolean cpy) {
- if (col instanceof ArrayList)
- return unwrapBinaries((ArrayList<Object>)col, keepBinary, cpy);
+ Collection<Object> col0 = BinaryUtils.newKnownCollection(col);
- if (col instanceof Set)
- return unwrapBinaries((Set<Object>)col, keepBinary, cpy);
+ if (col0 == null)
+ col0 = new ArrayList<>(col.size());
- Collection<Object> col0 = new ArrayList<>(col.size());
+ for (Object obj : col)
+ col0.add(unwrapBinary(obj, keepBinary, cpy));
+
+ return col0;
+ }
+
+ /**
+ * @param col Collection to unwrap.
+ * @param keepBinary Keep binary flag.
+ * @param cpy Copy flag.
+ * @return Unwrapped collection.
+ */
+ private Collection<Object> unwrapKnownCollection(Collection<Object> col, boolean keepBinary, boolean cpy) {
+ Collection<Object> col0 = BinaryUtils.newKnownCollection(col);
for (Object obj : col)
col0.add(unwrapBinary(obj, keepBinary, cpy));
@@ -212,44 +227,6 @@ import org.apache.ignite.internal.util.typedef.F;
}
/**
- * Unwraps array list.
- *
- * @param col List to unwrap.
- * @return Unwrapped list.
- */
- private Collection<Object> unwrapBinaries(ArrayList<Object> col, boolean keepBinary, boolean cpy) {
- int size = col.size();
-
- col = new ArrayList<>(col);
-
- for (int i = 0; i < size; i++) {
- Object o = col.get(i);
-
- Object unwrapped = unwrapBinary(o, keepBinary, cpy);
-
- if (o != unwrapped)
- col.set(i, unwrapped);
- }
-
- return col;
- }
-
- /**
- * Unwraps set with binary.
- *
- * @param set Set to unwrap.
- * @return Unwrapped set.
- */
- private Set<Object> unwrapBinaries(Set<Object> set, boolean keepBinary, boolean cpy) {
- Set<Object> set0 = BinaryUtils.newSet(set);
-
- for (Object obj : set)
- set0.add(unwrapBinary(obj, keepBinary, cpy));
-
- return set0;
- }
-
- /**
* @param o Object to unwrap.
* @return Unwrapped object.
*/
@@ -267,9 +244,9 @@ import org.apache.ignite.internal.util.typedef.F;
return (key != uKey || val != uVal) ? F.t(uKey, uVal) : o;
}
- else if (o instanceof Collection)
- return unwrapBinariesIfNeeded((Collection<Object>)o, keepBinary, cpy);
- else if (o instanceof Map)
+ else if (BinaryUtils.knownCollection(o))
+ return unwrapKnownCollection((Collection<Object>)o, keepBinary, cpy);
+ else if (BinaryUtils.knownMap(o))
return unwrapBinariesIfNeeded((Map<Object, Object>)o, keepBinary, cpy);
else if (o instanceof Object[])
return unwrapBinariesInArrayIfNeeded((Object[])o, keepBinary, cpy);
@@ -282,4 +259,24 @@ import org.apache.ignite.internal.util.typedef.F;
return o;
}
+
+ /**
+ * @param o Object to test.
+ * @return True if collection should be recursively unwrapped.
+ */
+ private boolean knownCollection(Object o) {
+ Class<?> cls = o == null ? null : o.getClass();
+
+ return cls == ArrayList.class || cls == LinkedList.class || cls == HashSet.class;
+ }
+
+ /**
+ * @param o Object to test.
+ * @return True if map should be recursively unwrapped.
+ */
+ private boolean knownMap(Object o) {
+ Class<?> cls = o == null ? null : o.getClass();
+
+ return cls == HashMap.class || cls == LinkedHashMap.class;
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
index bcc2ab7..c9d6dad 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
@@ -402,31 +402,30 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
return new IgniteBiTuple<>(marshalToBinary(tup.get1()), marshalToBinary(tup.get2()));
}
- if (obj instanceof Collection) {
- Collection<Object> col = (Collection<Object>)obj;
+ {
+ Collection<Object> pCol = BinaryUtils.newKnownCollection(obj);
- Collection<Object> pCol;
+ if (pCol != null) {
+ Collection<?> col = (Collection<?>)obj;
- if (col instanceof Set)
- pCol = (Collection<Object>)BinaryUtils.newSet((Set<?>)col);
- else
- pCol = new ArrayList<>(col.size());
+ for (Object item : col)
+ pCol.add(marshalToBinary(item));
- for (Object item : col)
- pCol.add(marshalToBinary(item));
-
- return pCol;
+ return pCol;
+ }
}
- if (obj instanceof Map) {
- Map<?, ?> map = (Map<?, ?>)obj;
+ {
+ Map<Object, Object> pMap = BinaryUtils.newKnownMap(obj);
- Map<Object, Object> pMap = BinaryUtils.newMap((Map<Object, Object>)obj);
+ if (pMap != null) {
+ Map<?, ?> map = (Map<?, ?>)obj;
- for (Map.Entry<?, ?> e : map.entrySet())
- pMap.put(marshalToBinary(e.getKey()), marshalToBinary(e.getValue()));
+ for (Map.Entry<?, ?> e : map.entrySet())
+ pMap.put(marshalToBinary(e.getKey()), marshalToBinary(e.getValue()));
- return pMap;
+ return pMap;
+ }
}
if (obj instanceof Map.Entry) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
index 998f7a2..9553656 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
@@ -696,9 +696,14 @@ public class GridDhtPartitionDemander {
(IgniteUuid)null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, entry.value(), true, null,
false, null, null, null, true);
}
- else if (log.isDebugEnabled())
- log.debug("Rebalancing entry is already in cache (will ignore) [key=" + cached.key() +
- ", part=" + p + ']');
+ else {
+ if (cctx.isSwapOrOffheapEnabled())
+ cctx.evicts().touch(cached, topVer); // Start tracking.
+
+ if (log.isDebugEnabled())
+ log.debug("Rebalancing entry is already in cache (will ignore) [key=" + cached.key() +
+ ", part=" + p + ']');
+ }
}
else if (log.isDebugEnabled())
log.debug("Rebalance predicate evaluated to false for entry (will ignore): " + entry);
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/test/java/org/apache/ignite/internal/binary/BinaryMarshallerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/binary/BinaryMarshallerSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/binary/BinaryMarshallerSelfTest.java
index 20b2258..c347b9f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/binary/BinaryMarshallerSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/binary/BinaryMarshallerSelfTest.java
@@ -113,7 +113,7 @@ public class BinaryMarshallerSelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testByte() throws Exception {
- assertEquals((byte) 100, marshalUnmarshal((byte)100).byteValue());
+ assertEquals((byte)100, marshalUnmarshal((byte)100).byteValue());
}
/**
@@ -401,7 +401,7 @@ public class BinaryMarshallerSelfTest extends GridCommonAbstractTest {
CustomCollections cc = new CustomCollections();
cc.list.add(1);
- cc.customList.add(2);
+ cc.customList.add(new Value(1));
CustomCollections copiedCc = marshalUnmarshal(cc);
@@ -415,6 +415,28 @@ public class BinaryMarshallerSelfTest extends GridCommonAbstractTest {
}
/**
+ * Test serialization of custom collections.
+ *
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("unchecked")
+ public void testCustomCollections2() throws Exception {
+ CustomArrayList arrList = new CustomArrayList();
+
+ arrList.add(1);
+
+ Object cp = marshalUnmarshal(arrList);
+
+ assert cp.getClass().equals(CustomArrayList.class);
+
+ CustomArrayList customCp = (CustomArrayList)cp;
+
+ assertEquals(customCp.size(), arrList.size());
+
+ assertEquals(customCp.get(0), arrList.get(0));
+ }
+
+ /**
* Test custom collections with factories.
*
* @throws Exception If failed.
@@ -3958,6 +3980,24 @@ public class BinaryMarshallerSelfTest extends GridCommonAbstractTest {
private Value(int val) {
this.val = val;
}
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (!(o instanceof Value))
+ return false;
+
+ Value value = (Value)o;
+
+ return val == value.val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return val;
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentSelfTest.java
index c18554e..613e98c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentSelfTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.cache;
+import java.util.ArrayList;
import java.util.Arrays;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
@@ -306,7 +307,7 @@ public class GridCacheDeploymentSelfTest extends GridCommonAbstractTest {
assert cache != null;
- cache.put(key, Arrays.asList(val1Cls.newInstance()));
+ cache.put(key, new ArrayList<>(Arrays.asList(val1Cls.newInstance())));
info(">>>>>>> First put completed.");
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheGetCustomCollectionsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheGetCustomCollectionsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheGetCustomCollectionsSelfTest.java
new file mode 100644
index 0000000..aa6fbd5
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheGetCustomCollectionsSelfTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.io.Serializable;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+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.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class IgniteCacheGetCustomCollectionsSelfTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ cfg.setMarshaller(null);
+
+ final CacheConfiguration<String, MyMap> mapCacheConfig = new CacheConfiguration<>();
+
+ mapCacheConfig.setCacheMode(CacheMode.PARTITIONED);
+ mapCacheConfig.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+ mapCacheConfig.setBackups(1);
+ mapCacheConfig.setName("cache");
+
+ cfg.setCacheConfiguration(mapCacheConfig);
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+
+ discoSpi.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(discoSpi);
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPutGet() throws Exception {
+ startGrids(3);
+
+ try {
+ IgniteEx ignite = grid(0);
+
+ IgniteCache<String, MyMap> cache = ignite.cache("cache");
+
+ Set<String> keys = new HashSet<>();
+
+ for (int i = 0; i < 100; i++) {
+ String key = "a" + i;;
+
+ MyMap map = new MyMap();
+
+ map.put("a", new Value());
+
+ cache.put(key, map);
+
+ map = cache.get(key);
+
+ keys.add(key);
+
+ Object a = map.get("a");
+
+ assertNotNull(a);
+ assertEquals(Value.class, a.getClass());
+ }
+
+ Map<String, MyMap> vals = cache.getAll(keys);
+
+ for (String key : keys) {
+ MyMap map = vals.get(key);
+
+ Object a = map.get("a");
+
+ assertNotNull(a);
+ assertEquals(Value.class, a.getClass());
+ }
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+
+ /**
+ *
+ */
+ private static class MyMap extends HashMap implements Serializable {
+
+ }
+
+ /**
+ *
+ */
+ private static class Value implements Serializable {
+ private int val;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheLoadRebalanceEvictionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheLoadRebalanceEvictionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheLoadRebalanceEvictionSelfTest.java
new file mode 100644
index 0000000..0b1e029
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheLoadRebalanceEvictionSelfTest.java
@@ -0,0 +1,188 @@
+/*
+ * 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.io.Serializable;
+import java.util.ArrayList;
+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 javax.cache.configuration.FactoryBuilder;
+import javax.cache.integration.CacheLoaderException;
+import javax.cache.integration.CacheWriterException;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.eviction.lru.LruEvictionPolicy;
+import org.apache.ignite.cache.store.CacheStoreAdapter;
+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.util.typedef.PA;
+import org.apache.ignite.lang.IgniteBiInClosure;
+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.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class IgniteCacheLoadRebalanceEvictionSelfTest extends GridCommonAbstractTest {
+ /** */
+ public static final int LRU_MAX_SIZE = 10;
+
+ /** */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int ENTRIES_CNT = 10000;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+
+ discoSpi.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(discoSpi);
+
+ LruEvictionPolicy evictionPolicy = new LruEvictionPolicy<>();
+ evictionPolicy.setMaxSize(LRU_MAX_SIZE);
+
+ CacheConfiguration<String, byte[]> cacheCfg = new CacheConfiguration<>();
+ cacheCfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
+ cacheCfg.setEvictSynchronized(false);
+ cacheCfg.setCacheMode(CacheMode.PARTITIONED);
+ cacheCfg.setBackups(1);
+ cacheCfg.setReadFromBackup(true);
+ cacheCfg.setEvictionPolicy(evictionPolicy);
+ cacheCfg.setOffHeapMaxMemory(1024 * 1024 * 1024L);
+ cacheCfg.setStatisticsEnabled(true);
+
+ cacheCfg.setWriteThrough(false);
+ cacheCfg.setReadThrough(false);
+
+ cacheCfg.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(new Storage()));
+
+ cfg.setCacheConfiguration(cacheCfg);
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testStartRebalancing() throws Exception {
+ List<IgniteInternalFuture<Object>> futs = new ArrayList<>();
+
+ int gridCnt = 4;
+
+ for (int i = 0; i < gridCnt; i++) {
+ final IgniteEx ig = startGrid(i);
+
+ futs.add(GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ ig.cache(null).localLoadCache(null);
+
+ return null;
+ }
+ }));
+ }
+
+ try {
+ for (IgniteInternalFuture<Object> fut : futs)
+ fut.get();
+
+ for (int i = 0; i < gridCnt; i++) {
+ IgniteEx grid = grid(i);
+
+ final IgniteCache<Object, Object> cache = grid.cache(null);
+
+ GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return cache.localSize(CachePeekMode.ONHEAP) <= 10;
+ }
+ }, getTestTimeout());
+ }
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+
+ /**
+ *
+ */
+ private static class Storage extends CacheStoreAdapter<Integer, byte[]> implements Serializable {
+ /** */
+ private static final byte[] data = new byte[1024];
+
+ /** {@inheritDoc} */
+ @Override public void write(Cache.Entry<? extends Integer, ? extends byte[]> e) throws CacheWriterException {
+ throw new UnsupportedOperationException("Unsupported");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeAll(Collection<Cache.Entry<? extends Integer, ? extends byte[]>> entries)
+ throws CacheWriterException {
+ throw new UnsupportedOperationException("Unsupported");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void delete(Object key) throws CacheWriterException {
+ throw new UnsupportedOperationException("Unsupported");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void deleteAll(Collection<?> keys) throws CacheWriterException {
+ throw new UnsupportedOperationException("Unsupported");
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte[] load(Integer key) throws CacheLoaderException {
+ return data;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Map<Integer, byte[]> loadAll(Iterable<? extends Integer> keys) throws CacheLoaderException {
+ Map<Integer, byte[]> res = new HashMap<>();
+
+ for (Integer key : keys)
+ res.put(key, data);
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void loadCache(IgniteBiInClosure<Integer, byte[]> clo,
+ @Nullable Object... args) throws CacheLoaderException {
+
+ for (int i = 0; i < ENTRIES_CNT; i++)
+ clo.apply(i, data);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeEchoTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeEchoTask.java b/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeEchoTask.java
index 03ab998..e945ada 100644
--- a/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeEchoTask.java
+++ b/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeEchoTask.java
@@ -17,6 +17,8 @@
package org.apache.ignite.platform;
+import java.util.ArrayList;
+import java.util.HashMap;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteException;
@@ -160,10 +162,10 @@ public class PlatformComputeEchoTask extends ComputeTaskAdapter<Integer, Object>
return new int[] { 1 };
case TYPE_COLLECTION:
- return Collections.singletonList(1);
+ return new ArrayList<>(Collections.singletonList(1));
case TYPE_MAP:
- return Collections.singletonMap(1, 1);
+ return new HashMap<>(Collections.singletonMap(1, 1));
case TYPE_BINARY:
return new PlatformComputeBinarizable(1);
http://git-wip-us.apache.org/repos/asf/ignite/blob/6524c796/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index fcc8d37..1b8eeda 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -19,6 +19,7 @@ package org.apache.ignite.testsuites;
import junit.framework.TestSuite;
import org.apache.ignite.cache.store.jdbc.CacheJdbcStoreSessionListenerSelfTest;
+import org.apache.ignite.internal.processors.cache.IgniteCacheGetCustomCollectionsSelfTest;
import org.apache.ignite.internal.processors.GridCacheTxLoadFromStoreOnLockSelfTest;
import org.apache.ignite.internal.processors.cache.CacheClientStoreSelfTest;
import org.apache.ignite.internal.processors.cache.CacheOffheapMapEntrySelfTest;
@@ -57,6 +58,7 @@ import org.apache.ignite.internal.processors.cache.IgniteCacheConfigurationDefau
import org.apache.ignite.internal.processors.cache.IgniteCacheConfigurationTemplateTest;
import org.apache.ignite.internal.processors.cache.IgniteCacheDynamicStopSelfTest;
import org.apache.ignite.internal.processors.cache.IgniteCacheInvokeReadThroughTest;
+import org.apache.ignite.internal.processors.cache.IgniteCacheLoadRebalanceEvictionSelfTest;
import org.apache.ignite.internal.processors.cache.IgniteCacheTxCopyOnReadDisabledTest;
import org.apache.ignite.internal.processors.cache.IgniteCacheTxLocalPeekModesTest;
import org.apache.ignite.internal.processors.cache.IgniteCacheTxLocalStoreValueTest;
@@ -286,6 +288,9 @@ public class IgniteCacheTestSuite4 extends TestSuite {
suite.addTestSuite(IgniteCacheSingleGetMessageTest.class);
+ suite.addTestSuite(IgniteCacheGetCustomCollectionsSelfTest.class);
+ suite.addTestSuite(IgniteCacheLoadRebalanceEvictionSelfTest.class);
+
return suite;
}
}
\ No newline at end of file
[12/16] ignite git commit: Merge remote-tracking branch
'remotes/origin/master' into ignite-1811
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/master' into ignite-1811
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/a728eccc
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/a728eccc
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/a728eccc
Branch: refs/heads/ignite-1811
Commit: a728eccc0ea44d388ecba610d2489200735b7200
Parents: 97d0c68 fa427dc
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jan 15 11:20:20 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jan 15 11:20:20 2016 +0300
----------------------------------------------------------------------
.../ignite/internal/binary/BinaryUtils.java | 78 ++++++++
.../processors/cache/CacheObjectContext.java | 91 +++++----
.../binary/CacheObjectBinaryProcessorImpl.java | 33 ++--
.../dht/preloader/GridDhtPartitionDemander.java | 11 +-
.../GridDhtPartitionsExchangeFuture.java | 8 +-
.../continuous/CacheContinuousQueryEntry.java | 70 +++----
.../binary/BinaryMarshallerSelfTest.java | 44 ++++-
.../cache/GridCacheDeploymentSelfTest.java | 3 +-
...IgniteCacheGetCustomCollectionsSelfTest.java | 128 +++++++++++++
...gniteCacheLoadRebalanceEvictionSelfTest.java | 188 +++++++++++++++++++
.../platform/PlatformComputeEchoTask.java | 6 +-
.../testsuites/IgniteCacheTestSuite4.java | 5 +
.../testsuites/IgniteCacheTestSuite5.java | 2 +
.../yardstick/cache/IgniteGetTxBenchmark.java | 30 +++
14 files changed, 588 insertions(+), 109 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/a728eccc/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index ee2d084,1b8eeda..97557c2
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@@ -286,8 -287,10 +288,11 @@@ public class IgniteCacheTestSuite4 exte
suite.addTestSuite(CacheGetFutureHangsSelfTest.class);
suite.addTestSuite(IgniteCacheSingleGetMessageTest.class);
+ suite.addTestSuite(IgniteCacheReadFromBackupTest.class);
+ suite.addTestSuite(IgniteCacheGetCustomCollectionsSelfTest.class);
+ suite.addTestSuite(IgniteCacheLoadRebalanceEvictionSelfTest.class);
+
return suite;
}
}
[02/16] ignite git commit: fixes to javadoc
Posted by sb...@apache.org.
fixes to javadoc
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/59cfbfde
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/59cfbfde
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/59cfbfde
Branch: refs/heads/ignite-1811
Commit: 59cfbfdea535539227fd02ac2966a8f5dc42ead6
Parents: 59a893c
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Thu Jan 14 12:54:48 2016 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Thu Jan 14 12:54:48 2016 +0300
----------------------------------------------------------------------
.../discovery/tcp/ipfinder/TcpDiscoveryIpFinder.java | 14 +++++++++++---
1 file changed, 11 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/59cfbfde/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinder.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinder.java
index 1e112c4..d277a3a 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinder.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinder.java
@@ -61,9 +61,17 @@ public interface TcpDiscoveryIpFinder {
/**
* Checks whether IP finder is shared or not.
* <p>
- * If it is shared then only coordinator can unregister addresses.
+ * If this property is set to {@code true} then IP finder allows to add and remove
+ * addresses in runtime and this is how, for example, IP finder should work in
+ * Amazon EC2 environment or any other environment where IPs may not be known beforehand.
* <p>
- * All nodes should register their address themselves, as early as possible on node start.
+ * If this property is set to {@code false} then IP finder is immutable and all the addresses
+ * should be listed in configuration before Ignite start. This is the most use case for IP finders
+ * local to current VM. Since, usually such IP finders are created per each Ignite instance and
+ * all the known IPs are listed right away, but there is also an option to make such IP finders shared
+ * by setting this property to {@code true} and literally share it between local VM Ignite instances.
+ * This way user does not have to list any IPs before start, instead all starting nodes add their addresses
+ * to the finder, then get the registered addresses and continue with discovery procedure.
*
* @return {@code true} if IP finder is shared.
*/
@@ -95,4 +103,4 @@ public interface TcpDiscoveryIpFinder {
* Closes this IP finder and releases any system resources associated with it.
*/
public void close();
-}
\ No newline at end of file
+}
[04/16] ignite git commit: Added more logging for direct message read
errors.
Posted by sb...@apache.org.
Added more logging for direct message read errors.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/8b0c59a2
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/8b0c59a2
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/8b0c59a2
Branch: refs/heads/ignite-1811
Commit: 8b0c59a25822e69bd2fe896685be5e921b859e26
Parents: b409b8d
Author: sboikov <sb...@gridgain.com>
Authored: Thu Jan 14 16:42:46 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Jan 14 16:42:46 2016 +0300
----------------------------------------------------------------------
.../internal/direct/DirectMessageReader.java | 11 +++++
.../internal/direct/DirectMessageWriter.java | 13 +++++
.../direct/state/DirectMessageState.java | 10 ++++
.../stream/v1/DirectByteBufferStreamImplV1.java | 8 +++
.../stream/v2/DirectByteBufferStreamImplV2.java | 8 +++
.../internal/util/nio/GridDirectParser.java | 52 +++++++++++++-------
.../communication/tcp/TcpCommunicationSpi.java | 9 +++-
7 files changed, 92 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/8b0c59a2/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
index e0b7b22..10bc7e2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageReader.java
@@ -27,6 +27,7 @@ import org.apache.ignite.internal.direct.state.DirectMessageStateItem;
import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
import org.apache.ignite.internal.direct.stream.v1.DirectByteBufferStreamImplV1;
import org.apache.ignite.internal.direct.stream.v2.DirectByteBufferStreamImplV2;
+import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -373,6 +374,11 @@ public class DirectMessageReader implements MessageReader {
state.reset();
}
+ /** {@inheritDoc} */
+ public String toString() {
+ return S.toString(DirectMessageReader.class, this);
+ }
+
/**
*/
private static class StateItem implements DirectMessageStateItem {
@@ -407,5 +413,10 @@ public class DirectMessageReader implements MessageReader {
@Override public void reset() {
state = 0;
}
+
+ /** {@inheritDoc} */
+ public String toString() {
+ return S.toString(StateItem.class, this);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8b0c59a2/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
index 085cf68..730f9bc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/DirectMessageWriter.java
@@ -27,6 +27,8 @@ import org.apache.ignite.internal.direct.state.DirectMessageStateItem;
import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
import org.apache.ignite.internal.direct.stream.v1.DirectByteBufferStreamImplV1;
import org.apache.ignite.internal.direct.stream.v2.DirectByteBufferStreamImplV2;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -39,6 +41,7 @@ import org.jetbrains.annotations.Nullable;
*/
public class DirectMessageWriter implements MessageWriter {
/** State. */
+ @GridToStringInclude
private final DirectMessageState<StateItem> state;
/**
@@ -327,6 +330,11 @@ public class DirectMessageWriter implements MessageWriter {
state.reset();
}
+ /** {@inheritDoc} */
+ public String toString() {
+ return S.toString(DirectMessageWriter.class, this);
+ }
+
/**
*/
private static class StateItem implements DirectMessageStateItem {
@@ -364,5 +372,10 @@ public class DirectMessageWriter implements MessageWriter {
state = 0;
hdrWritten = false;
}
+
+ /** {@inheritDoc} */
+ public String toString() {
+ return S.toString(StateItem.class, this);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8b0c59a2/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java
index a61bb30..8ad7fe0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/state/DirectMessageState.java
@@ -18,6 +18,9 @@
package org.apache.ignite.internal.direct.state;
import java.lang.reflect.Array;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteOutClosure;
/**
@@ -28,9 +31,11 @@ public class DirectMessageState<T extends DirectMessageStateItem> {
private static final int INIT_SIZE = 10;
/** Item factory. */
+ @GridToStringExclude
private final IgniteOutClosure<T> factory;
/** Stack array. */
+ @GridToStringInclude
private T[] stack;
/** Current position. */
@@ -95,4 +100,9 @@ public class DirectMessageState<T extends DirectMessageStateItem> {
stack[0].reset();
}
+
+ /** {@inheritDoc} */
+ public String toString() {
+ return S.toString(DirectMessageState.class, this);
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8b0c59a2/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java
index ad8671d..5292f35 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v1/DirectByteBufferStreamImplV1.java
@@ -28,6 +28,8 @@ import java.util.NoSuchElementException;
import java.util.UUID;
import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
import org.apache.ignite.internal.util.GridUnsafe;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -217,6 +219,7 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
private static final Object NULL = new Object();
/** */
+ @GridToStringExclude
private final MessageFactory msgFactory;
/** */
@@ -1347,6 +1350,11 @@ public class DirectByteBufferStreamImplV1 implements DirectByteBufferStream {
};
}
+ /** {@inheritDoc} */
+ public String toString() {
+ return S.toString(DirectByteBufferStreamImplV1.class, this);
+ }
+
/**
* Array creator.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/8b0c59a2/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java
index 89c9cc6..ed3eb7c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/direct/stream/v2/DirectByteBufferStreamImplV2.java
@@ -29,6 +29,8 @@ import java.util.RandomAccess;
import java.util.UUID;
import org.apache.ignite.internal.direct.stream.DirectByteBufferStream;
import org.apache.ignite.internal.util.GridUnsafe;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
@@ -218,6 +220,7 @@ public class DirectByteBufferStreamImplV2 implements DirectByteBufferStream {
private static final Object NULL = new Object();
/** */
+ @GridToStringExclude
private final MessageFactory msgFactory;
/** */
@@ -1570,6 +1573,11 @@ public class DirectByteBufferStreamImplV2 implements DirectByteBufferStream {
}
}
+ /** {@inheritDoc} */
+ public String toString() {
+ return S.toString(DirectByteBufferStreamImplV2.class, this);
+ }
+
/**
* Array creator.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/8b0c59a2/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
index aa88808..76e7d4d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
@@ -21,6 +21,8 @@ import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
@@ -37,19 +39,24 @@ public class GridDirectParser implements GridNioParser {
private static final int READER_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
/** */
+ private final IgniteLogger log;
+
+ /** */
private final MessageFactory msgFactory;
/** */
private final GridNioMessageReaderFactory readerFactory;
/**
+ * @param log Logger.
* @param msgFactory Message factory.
* @param readerFactory Message reader factory.
*/
- public GridDirectParser(MessageFactory msgFactory, GridNioMessageReaderFactory readerFactory) {
+ public GridDirectParser(IgniteLogger log, MessageFactory msgFactory, GridNioMessageReaderFactory readerFactory) {
assert msgFactory != null;
assert readerFactory != null;
+ this.log = log;
this.msgFactory = msgFactory;
this.readerFactory = readerFactory;
}
@@ -64,28 +71,39 @@ public class GridDirectParser implements GridNioParser {
Message msg = ses.removeMeta(MSG_META_KEY);
- if (msg == null && buf.hasRemaining())
- msg = msgFactory.create(buf.get());
+ try {
+ if (msg == null && buf.hasRemaining())
+ msg = msgFactory.create(buf.get());
- boolean finished = false;
+ boolean finished = false;
- if (buf.hasRemaining()) {
- if (reader != null)
- reader.setCurrentReadClass(msg.getClass());
+ if (buf.hasRemaining()) {
+ if (reader != null)
+ reader.setCurrentReadClass(msg.getClass());
- finished = msg.readFrom(buf, reader);
- }
+ finished = msg.readFrom(buf, reader);
+ }
- if (finished) {
- if (reader != null)
- reader.reset();
+ if (finished) {
+ if (reader != null)
+ reader.reset();
- return msg;
- }
- else {
- ses.addMeta(MSG_META_KEY, msg);
+ return msg;
+ }
+ else {
+ ses.addMeta(MSG_META_KEY, msg);
- return null;
+ return null;
+ }
+ }
+ catch (Throwable e) {
+ U.error(log, "Failed to read message [msg=" + msg +
+ ", buf=" + buf +
+ ", reader=" + reader +
+ ", ses=" + ses + "]",
+ e);
+
+ throw e;
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8b0c59a2/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index 6cdfe9a..918bc83 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -1611,7 +1611,9 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
}
};
- GridDirectParser parser = new GridDirectParser(msgFactory, readerFactory);
+ GridDirectParser parser = new GridDirectParser(log.getLogger(GridDirectParser.class),
+ msgFactory,
+ readerFactory);
IgnitePredicate<Message> skipRecoveryPred = new IgnitePredicate<Message>() {
@Override public boolean apply(Message msg) {
@@ -2983,7 +2985,10 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
endpoint,
srvLsnr,
writerFactory,
- new GridNioCodecFilter(new GridDirectParser(msgFactory, readerFactory), log, true),
+ new GridNioCodecFilter(
+ new GridDirectParser(log.getLogger(GridDirectParser.class),msgFactory, readerFactory),
+ log,
+ true),
new GridConnectionBytesVerifyFilter(log)
);
[03/16] ignite git commit: Merge remote-tracking branch
'origin/master'
Posted by sb...@apache.org.
Merge remote-tracking branch 'origin/master'
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b409b8de
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b409b8de
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b409b8de
Branch: refs/heads/ignite-1811
Commit: b409b8dee64d297fe03869533182109c8e40c7a4
Parents: 59cfbfd 2d10630
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Thu Jan 14 12:55:10 2016 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Thu Jan 14 12:55:10 2016 +0300
----------------------------------------------------------------------
.../org/apache/ignite/IgniteTransactions.java | 4 +-
.../apache/ignite/internal/GridComponent.java | 5 +-
.../ignite/internal/GridUpdateNotifier.java | 454 ------------------
.../apache/ignite/internal/IgniteKernal.java | 221 ++++-----
.../ignite/internal/IgniteVersionUtils.java | 4 +-
.../ignite/internal/MarshallerContextImpl.java | 86 +++-
.../discovery/GridDiscoveryManager.java | 6 +-
.../cache/CacheEntrySerializablePredicate.java | 3 +-
.../cache/CacheInvokeDirectResult.java | 4 +-
.../processors/cache/GridCacheIoManager.java | 23 +
.../processors/cache/GridCacheProcessor.java | 52 ++-
.../processors/cache/GridCacheReturn.java | 2 +
.../processors/cache/IgniteCacheProxy.java | 2 +-
.../GridDistributedLockResponse.java | 2 +-
.../GridDistributedTxFinishRequest.java | 11 +-
.../GridDistributedTxPrepareRequest.java | 2 +-
.../GridDistributedTxPrepareResponse.java | 4 +-
.../dht/GridDhtAffinityAssignmentResponse.java | 4 +-
.../distributed/dht/GridDhtLockFuture.java | 44 +-
.../distributed/dht/GridDhtLockRequest.java | 2 +-
.../distributed/dht/GridDhtTxFinishRequest.java | 90 ++--
.../dht/GridDhtTxFinishResponse.java | 4 +-
.../cache/distributed/dht/GridDhtTxLocal.java | 1 +
.../dht/GridDhtTxPrepareRequest.java | 2 +-
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 21 +-
.../dht/atomic/GridDhtAtomicUpdateResponse.java | 6 +-
.../dht/atomic/GridNearAtomicUpdateRequest.java | 22 +-
.../atomic/GridNearAtomicUpdateResponse.java | 4 +-
.../dht/preloader/GridDhtForceKeysResponse.java | 6 +-
.../GridDhtPartitionDemandMessage.java | 6 +-
.../GridDhtPartitionSupplyMessageV2.java | 6 +-
.../preloader/GridDhtPartitionsFullMessage.java | 2 +-
.../GridDhtPartitionsSingleMessage.java | 4 +-
.../distributed/near/GridNearGetResponse.java | 4 +-
...ridNearOptimisticTxPrepareFutureAdapter.java | 6 +-
.../near/GridNearSingleGetResponse.java | 2 +-
.../near/GridNearTxFinishFuture.java | 432 +++++++++++++-----
.../near/GridNearTxFinishRequest.java | 5 +
.../near/GridNearTxFinishResponse.java | 4 +-
.../cache/distributed/near/GridNearTxLocal.java | 2 +-
.../near/GridNearTxPrepareResponse.java | 4 +-
.../cache/query/GridCacheQueryRequest.java | 16 +-
.../cache/query/GridCacheQueryResponse.java | 18 +-
.../cache/transactions/IgniteInternalTx.java | 6 +
.../cache/transactions/IgniteTxAdapter.java | 23 +-
.../cache/transactions/IgniteTxEntry.java | 11 +-
.../cache/transactions/IgniteTxHandler.java | 26 +-
.../transactions/IgniteTxLocalAdapter.java | 6 +-
.../cache/transactions/IgniteTxManager.java | 20 +
.../processors/cluster/ClusterProcessor.java | 174 +++++++
.../processors/cluster/GridUpdateNotifier.java | 457 +++++++++++++++++++
.../datastreamer/DataStreamerRequest.java | 1 +
.../datastructures/DataStructuresProcessor.java | 11 +-
.../processors/igfs/IgfsAckMessage.java | 4 +-
.../handlers/cache/GridCacheCommandHandler.java | 6 +-
.../ignite/spi/discovery/DiscoverySpi.java | 2 +
.../ignite/stream/socket/SocketStreamer.java | 3 +-
.../internal/GridUpdateNotifierSelfTest.java | 137 ------
...UpdateNotifierPerClusterSettingSelfTest.java | 130 ++++++
...cheAbstractFullApiMultithreadedSelfTest.java | 13 +-
.../cache/GridCacheAbstractFullApiSelfTest.java | 2 +-
.../processors/cache/GridCacheStopSelfTest.java | 2 +-
.../cache/IgniteDynamicCacheStartSelfTest.java | 30 +-
...eMarshallerCacheConcurrentReadWriteTest.java | 189 ++++++++
.../IgniteClientDataStructuresAbstractTest.java | 3 +
.../dht/GridCacheTxNodeFailureSelfTest.java | 13 +-
.../IgniteCacheCommitDelayTxRecoveryTest.java | 376 +++++++++++++++
.../IgniteCachePutRetryAbstractSelfTest.java | 36 +-
...gniteCachePutRetryTransactionalSelfTest.java | 21 +
.../cluster/GridUpdateNotifierSelfTest.java | 140 ++++++
.../continuous/GridEventConsumeSelfTest.java | 3 +
.../internal/util/nio/GridNioSelfTest.java | 11 +-
...dTcpCommunicationSpiRecoveryAckSelfTest.java | 1 -
...CommunicationRecoveryAckClosureSelfTest.java | 19 +-
.../spi/discovery/tcp/TcpDiscoverySelfTest.java | 2 +
.../startup/GridRandomCommandLineLoader.java | 4 +-
.../junits/common/GridCommonAbstractTest.java | 34 +-
.../ignite/testsuites/IgniteBasicTestSuite.java | 2 +
.../IgniteCacheTxRecoverySelfTestSuite.java | 3 +
.../testsuites/IgniteKernalSelfTestSuite.java | 4 +-
.../tcp/ipfinder/zk/ZookeeperIpFinderTest.java | 69 ++-
81 files changed, 2502 insertions(+), 1094 deletions(-)
----------------------------------------------------------------------
[16/16] ignite git commit: ignite-1811
Posted by sb...@apache.org.
ignite-1811
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/caaab6e5
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/caaab6e5
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/caaab6e5
Branch: refs/heads/ignite-1811
Commit: caaab6e541a532a9815e40f27337a2c244ba21ef
Parents: 98d470a
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jan 15 15:07:16 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jan 15 15:07:16 2016 +0300
----------------------------------------------------------------------
.../cache/distributed/dht/GridClientPartitionTopology.java | 2 +-
.../cache/distributed/dht/GridPartitionedGetFuture.java | 2 ++
.../processors/cache/distributed/near/GridNearGetFuture.java | 4 +++-
.../cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java | 2 +-
4 files changed, 7 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/caaab6e5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index f0c4c90..dcfc038 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@ -882,7 +882,7 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@Override public boolean rebalanceFinished(AffinityTopologyVersion topVer) {
- assert false;
+ assert false : "Should not be called on non-affinity node";
return false;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/caaab6e5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index 2bc6869..1f2d7c5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -431,6 +431,8 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
* @return {@code True} if there is no need to further search value.
*/
private boolean localGet(KeyCacheObject key, int part, Map<K, V> locVals) {
+ assert cctx.affinityNode() : this;
+
GridDhtCacheAdapter<K, V> cache = cache();
while (true) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/caaab6e5/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 12a1b74..9291001 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -548,7 +548,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
* @param key Key.
* @param part Partition.
* @param topVer Topology version.
- * @param nearRead {@code True} if tried to read from near cache.
+ * @param nearRead {@code True} if already tried to read from near cache.
* @return {@code True} if there is no need to further search value.
*/
private boolean localDhtGet(KeyCacheObject key,
@@ -557,6 +557,8 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
boolean nearRead) {
GridDhtCacheAdapter<K, V> dht = cache().dht();
+ assert dht.context().affinityNode() : this;
+
while (true) {
GridCacheEntryEx dhtEntry = null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/caaab6e5/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java
index 7b1b6f1..ac80d69 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartNoExchangeTimeoutTest.java
@@ -326,7 +326,7 @@ public class IgniteDynamicCacheStartNoExchangeTimeoutTest extends GridCommonAbst
for (int i = 0; i < 100; i++) {
cache.put(key, key);
- assertEquals("Invalid value for node: " + ignite.name(), key, cache.get(key));
+ assertEquals(key, cache.get(key));
key++;
}
[09/16] ignite git commit: ignite-1811
Posted by sb...@apache.org.
ignite-1811
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/0823cd63
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/0823cd63
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/0823cd63
Branch: refs/heads/ignite-1811
Commit: 0823cd6342d0241ddfaebc00e88d36aad6a6358c
Parents: 969faef
Author: sboikov <sb...@gridgain.com>
Authored: Thu Jan 14 11:56:45 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jan 15 11:14:35 2016 +0300
----------------------------------------------------------------------
.../internal/TestRecordingCommunicationSpi.java | 77 +++++++++++++
.../cache/IgniteCacheNearLockValueSelfTest.java | 62 ++--------
.../IgniteCacheReadFromBackupTest.java | 114 +++++++++++++++++++
.../IgniteCacheSingleGetMessageTest.java | 68 ++---------
.../GridCacheDhtPreloadMessageCountTest.java | 62 ++--------
.../testsuites/IgniteCacheTestSuite4.java | 2 +
6 files changed, 224 insertions(+), 161 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/0823cd63/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
new file mode 100644
index 0000000..bf84387
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
@@ -0,0 +1,77 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
+ /** */
+ private Class<?> recordCls;
+
+ /** */
+ private List<Object> recordedMsgs = new ArrayList<>();
+
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
+ throws IgniteSpiException {
+ if (msg instanceof GridIoMessage) {
+ Object msg0 = ((GridIoMessage)msg).message();
+
+ synchronized (this) {
+ if (recordCls != null && msg0.getClass().equals(recordCls))
+ recordedMsgs.add(msg0);
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+
+ /**
+ * @param recordCls Message class to record.
+ */
+ public void record(@Nullable Class<?> recordCls) {
+ synchronized (this) {
+ this.recordCls = recordCls;
+ }
+ }
+
+ /**
+ * @return Recorded messages.
+ */
+ public List<Object> recordedMessages() {
+ synchronized (this) {
+ List<Object> msgs = recordedMsgs;
+
+ recordedMsgs = new ArrayList<>();
+
+ return msgs;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0823cd63/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java
index 100acfe..f106fec 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNearLockValueSelfTest.java
@@ -18,22 +18,15 @@
package org.apache.ignite.internal.processors.cache;
import java.util.Collection;
-import java.util.concurrent.ConcurrentLinkedDeque;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CacheAtomicityMode;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.IgniteKernal;
-import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.spi.IgniteSpiException;
-import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -71,7 +64,11 @@ public class IgniteCacheNearLockValueSelfTest extends GridCommonAbstractTest {
if (getTestGridName(0).equals(gridName))
cfg.setClientMode(true);
- cfg.setCommunicationSpi(new TestCommunicationSpi());
+ TestRecordingCommunicationSpi commSpi = new TestRecordingCommunicationSpi();
+
+ commSpi.record(GridNearLockRequest.class);
+
+ cfg.setCommunicationSpi(commSpi);
return cfg;
}
@@ -88,18 +85,18 @@ public class IgniteCacheNearLockValueSelfTest extends GridCommonAbstractTest {
cache.put("key1", "val1");
for (int i = 0; i < 3; i++) {
- ((TestCommunicationSpi)ignite(0).configuration().getCommunicationSpi()).clear();
- ((TestCommunicationSpi)ignite(1).configuration().getCommunicationSpi()).clear();
-
try (Transaction tx = ignite(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
cache.get("key1");
tx.commit();
}
- TestCommunicationSpi comm = (TestCommunicationSpi)ignite(0).configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi comm =
+ (TestRecordingCommunicationSpi)ignite(0).configuration().getCommunicationSpi();
+
+ Collection<GridNearLockRequest> reqs = (Collection)comm.recordedMessages();
- assertEquals(1, comm.requests().size());
+ assertEquals(1, reqs.size());
GridCacheAdapter<Object, Object> primary = ((IgniteKernal)grid(1)).internalCache("partitioned");
@@ -107,7 +104,7 @@ public class IgniteCacheNearLockValueSelfTest extends GridCommonAbstractTest {
assertNotNull(dhtEntry);
- GridNearLockRequest req = comm.requests().iterator().next();
+ GridNearLockRequest req = reqs.iterator().next();
assertEquals(dhtEntry.version(), req.dhtVersion(0));
@@ -122,39 +119,4 @@ public class IgniteCacheNearLockValueSelfTest extends GridCommonAbstractTest {
}
}
}
-
- /**
- *
- */
- private static class TestCommunicationSpi extends TcpCommunicationSpi {
- /** */
- private Collection<GridNearLockRequest> reqs = new ConcurrentLinkedDeque<>();
-
- /** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
- throws IgniteSpiException {
- if (msg instanceof GridIoMessage) {
- GridIoMessage ioMsg = (GridIoMessage)msg;
-
- if (ioMsg.message() instanceof GridNearLockRequest)
- reqs.add((GridNearLockRequest)ioMsg.message());
- }
-
- super.sendMessage(node, msg, ackC);
- }
-
- /**
- * @return Collected requests.
- */
- public Collection<GridNearLockRequest> requests() {
- return reqs;
- }
-
- /**
- *
- */
- public void clear() {
- reqs.clear();
- }
- }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/0823cd63/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
new file mode 100644
index 0000000..ee72909
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheReadFromBackupTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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 org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+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.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class IgniteCacheReadFromBackupTest extends GridCommonAbstractTest {
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int NODES = 4;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TestRecordingCommunicationSpi commSpi = new TestRecordingCommunicationSpi();
+
+ cfg.setCommunicationSpi(commSpi);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ return cfg;
+ }
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGridsMultiThreaded(NODES);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNoPrimaryReadPreloadFinished() throws Exception {
+ }
+
+ /**
+ * @param ccfg Cache configuration.
+ * @throws Exception If failed.
+ */
+ private void checkNoPrimaryRead(CacheConfiguration<Object, Object> ccfg) throws Exception {
+ ignite(0).createCache(ccfg);
+
+ try {
+ for (int i = 0; i < NODES; i++) {
+ Ignite ignite = ignite(i);
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(ccfg.getName());
+
+ TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ Integer key = nearKey(cache);
+
+ assertNull(cache.get(key));
+ }
+ }
+ finally {
+ ignite(0).destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @param cacheMode Cache mode.
+ * @param backups Number of backups.
+ * @return Cache configuration.
+ */
+ private CacheConfiguration<Object, Object> cacheConfiguration(CacheMode cacheMode, int backups) {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
+
+ ccfg.setCacheMode(cacheMode);
+
+ if (cacheMode != CacheMode.REPLICATED)
+ ccfg.setBackups(backups);
+
+ return ccfg;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0823cd63/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
index 722ee83..48fc961 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSingleGetMessageTest.java
@@ -21,26 +21,19 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.spi.IgniteSpiException;
-import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.transactions.Transaction;
-import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
@@ -69,7 +62,7 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
cfg.setClientMode(client);
- TestCommunicationSpi commSpi = new TestCommunicationSpi();
+ TestRecordingCommunicationSpi commSpi = new TestRecordingCommunicationSpi();
cfg.setCommunicationSpi(commSpi);
@@ -156,7 +149,7 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
Ignite node = cache.unwrap(Ignite.class);
- TestCommunicationSpi spi = (TestCommunicationSpi)node.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi spi = (TestRecordingCommunicationSpi)node.configuration().getCommunicationSpi();
spi.record(GridNearSingleGetRequest.class);
@@ -164,7 +157,8 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
assertNotSame(node, primary);
- TestCommunicationSpi primarySpi = (TestCommunicationSpi)primary.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi primarySpi =
+ (TestRecordingCommunicationSpi)primary.configuration().getCommunicationSpi();
primarySpi.record(GridNearSingleGetResponse.class);
@@ -253,7 +247,7 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
* @param spi Near node SPI.
* @param primarySpi Primary node SPI.
*/
- private void checkMessages(TestCommunicationSpi spi, TestCommunicationSpi primarySpi) {
+ private void checkMessages(TestRecordingCommunicationSpi spi, TestRecordingCommunicationSpi primarySpi) {
List<Object> msgs = spi.recordedMessages();
assertEquals(1, msgs.size());
@@ -269,7 +263,7 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
* @param spi Near node SPI.
* @param primarySpi Primary node SPI.
*/
- private void checkNoMessages(TestCommunicationSpi spi, TestCommunicationSpi primarySpi) {
+ private void checkNoMessages(TestRecordingCommunicationSpi spi, TestRecordingCommunicationSpi primarySpi) {
List<Object> msgs = spi.recordedMessages();
assertEquals(0, msgs.size());
@@ -318,52 +312,4 @@ public class IgniteCacheSingleGetMessageTest extends GridCommonAbstractTest {
return ccfg;
}
-
- /**
- *
- */
- private static class TestCommunicationSpi extends TcpCommunicationSpi {
- /** */
- private Class<?> recordCls;
-
- /** */
- private List<Object> recordedMsgs = new ArrayList<>();
-
- /** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
- throws IgniteSpiException {
- if (msg instanceof GridIoMessage) {
- Object msg0 = ((GridIoMessage)msg).message();
-
- synchronized (this) {
- if (recordCls != null && msg0.getClass().equals(recordCls))
- recordedMsgs.add(msg0);
- }
- }
-
- super.sendMessage(node, msg, ackC);
- }
-
- /**
- * @param recordCls Message class to record.
- */
- void record(@Nullable Class<?> recordCls) {
- synchronized (this) {
- this.recordCls = recordCls;
- }
- }
-
- /**
- * @return Recorded messages.
- */
- List<Object> recordedMessages() {
- synchronized (this) {
- List<Object> msgs = recordedMsgs;
-
- recordedMsgs = new ArrayList<>();
-
- return msgs;
- }
- }
- }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0823cd63/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java
index fd94150..0666349 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtPreloadMessageCountTest.java
@@ -17,25 +17,17 @@
package org.apache.ignite.internal.processors.cache.distributed.dht;
-import java.util.Collection;
-import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.cache.CacheRebalanceMode;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
-import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteInClosure;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.spi.IgniteSpiException;
-import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -79,7 +71,11 @@ public class GridCacheDhtPreloadMessageCountTest extends GridCommonAbstractTest
c.setDiscoverySpi(disco);
c.setCacheConfiguration(cc);
- c.setCommunicationSpi(new TestCommunicationSpi());
+ TestRecordingCommunicationSpi commSpi = new TestRecordingCommunicationSpi();
+
+ commSpi.record(GridDhtPartitionsSingleMessage.class);
+
+ c.setCommunicationSpi(commSpi);
return c;
}
@@ -110,11 +106,13 @@ public class GridCacheDhtPreloadMessageCountTest extends GridCommonAbstractTest
IgniteCache<String, Integer> c1 = g1.cache(null);
IgniteCache<String, Integer> c2 = g2.cache(null);
- TestCommunicationSpi spi0 = (TestCommunicationSpi)g0.configuration().getCommunicationSpi();
- TestCommunicationSpi spi1 = (TestCommunicationSpi)g1.configuration().getCommunicationSpi();
- TestCommunicationSpi spi2 = (TestCommunicationSpi)g2.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi spi0 = (TestRecordingCommunicationSpi)g0.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi spi1 = (TestRecordingCommunicationSpi)g1.configuration().getCommunicationSpi();
+ TestRecordingCommunicationSpi spi2 = (TestRecordingCommunicationSpi)g2.configuration().getCommunicationSpi();
- info(spi0.sentMessages().size() + " " + spi1.sentMessages().size() + " " + spi2.sentMessages().size());
+ info(spi0.recordedMessages().size() + " " +
+ spi1.recordedMessages().size() + " " +
+ spi2.recordedMessages().size());
checkCache(c0, cnt);
checkCache(c1, cnt);
@@ -137,40 +135,4 @@ public class GridCacheDhtPreloadMessageCountTest extends GridCommonAbstractTest
assertEquals(Integer.valueOf(i), c.localPeek(key, CachePeekMode.ONHEAP));
}
}
-
- /**
- * Communication SPI that will count single partition update messages.
- */
- private static class TestCommunicationSpi extends TcpCommunicationSpi {
- /** Recorded messages. */
- private Collection<GridDhtPartitionsSingleMessage> sentMsgs = new ConcurrentLinkedQueue<>();
-
- /** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackClosure)
- throws IgniteSpiException {
- recordMessage((GridIoMessage)msg);
-
- super.sendMessage(node, msg, ackClosure);
- }
-
- /**
- * @return Collection of sent messages.
- */
- public Collection<GridDhtPartitionsSingleMessage> sentMessages() {
- return sentMsgs;
- }
-
- /**
- * Adds message to a list if message is of correct type.
- *
- * @param msg Message.
- */
- private void recordMessage(GridIoMessage msg) {
- if (msg.message() instanceof GridDhtPartitionsSingleMessage) {
- GridDhtPartitionsSingleMessage partSingleMsg = (GridDhtPartitionsSingleMessage)msg.message();
-
- sentMsgs.add(partSingleMsg);
- }
- }
- }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/0823cd63/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index fcc8d37..ee2d084 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -81,6 +81,7 @@ import org.apache.ignite.internal.processors.cache.distributed.CacheGetFutureHan
import org.apache.ignite.internal.processors.cache.distributed.CacheNoValueClassOnServerNodeTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheCreatePutMultiNodeSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheCreatePutTest;
+import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheReadFromBackupTest;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheSingleGetMessageTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridCacheDhtTxPreloadSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.dht.IgniteCacheLockFailoverSelfTest;
@@ -285,6 +286,7 @@ public class IgniteCacheTestSuite4 extends TestSuite {
suite.addTestSuite(CacheGetFutureHangsSelfTest.class);
suite.addTestSuite(IgniteCacheSingleGetMessageTest.class);
+ suite.addTestSuite(IgniteCacheReadFromBackupTest.class);
return suite;
}
[14/16] ignite git commit: IGNITE-429
Posted by sb...@apache.org.
IGNITE-429
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3a0cb51f
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3a0cb51f
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3a0cb51f
Branch: refs/heads/ignite-1811
Commit: 3a0cb51f8303af5a841c2d5353f9d0ddc0e626df
Parents: fa427dc
Author: shtykh_roman <rs...@yahoo.com>
Authored: Fri Jan 15 14:24:22 2016 +0300
Committer: Anton Vinogradov <av...@apache.org>
Committed: Fri Jan 15 14:24:22 2016 +0300
----------------------------------------------------------------------
modules/storm/README.txt | 37 +++
modules/storm/licenses/apache-2.0.txt | 202 ++++++++++++
modules/storm/pom.xml | 104 +++++++
.../ignite/stream/storm/StormStreamer.java | 304 +++++++++++++++++++
.../ignite/stream/storm/package-info.java | 22 ++
.../storm/IgniteStormStreamerSelfTestSuite.java | 38 +++
.../storm/StormIgniteStreamerSelfTest.java | 184 +++++++++++
.../ignite/stream/storm/TestStormSpout.java | 141 +++++++++
.../storm/src/test/resources/example-ignite.xml | 71 +++++
pom.xml | 1 +
10 files changed, 1104 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/README.txt
----------------------------------------------------------------------
diff --git a/modules/storm/README.txt b/modules/storm/README.txt
new file mode 100644
index 0000000..eb20f25
--- /dev/null
+++ b/modules/storm/README.txt
@@ -0,0 +1,37 @@
+Apache Ignite Storm Streamer Module
+-----------------------------------
+
+Apache Ignite Storm Streamer module provides streaming via Storm to Ignite cache.
+
+Starting data transfer to Ignite cache can be done with the following steps.
+
+1. Import Ignite Kafka Streamer Module In Maven Project
+
+If you are using Maven to manage dependencies of your project, you can add Storm module
+dependency like this (replace '${ignite.version}' with actual Ignite version you are
+interested in):
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+ http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ ...
+ <dependencies>
+ ...
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-storm</artifactId>
+ <version>${ignite.version}</version>
+ </dependency>
+ ...
+ </dependencies>
+ ...
+</project>
+
+2. Create an Ignite configuration file (see example-ignite.xml) and make sure it is accessible from the streamer.
+
+3. Make sure your key-value data input to the streamer is specified with the field named "ignite"
+(or a different one you configure with StormStreamer.setIgniteTupleField(...)).
+See TestStormSpout.declareOutputFields(...) for an example.
+
+4. Create a topology with the streamer and start.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/licenses/apache-2.0.txt
----------------------------------------------------------------------
diff --git a/modules/storm/licenses/apache-2.0.txt b/modules/storm/licenses/apache-2.0.txt
new file mode 100644
index 0000000..d645695
--- /dev/null
+++ b/modules/storm/licenses/apache-2.0.txt
@@ -0,0 +1,202 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/pom.xml
----------------------------------------------------------------------
diff --git a/modules/storm/pom.xml b/modules/storm/pom.xml
new file mode 100644
index 0000000..0cd1fcb
--- /dev/null
+++ b/modules/storm/pom.xml
@@ -0,0 +1,104 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+ 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.
+-->
+
+<!--
+ POM file.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-parent</artifactId>
+ <version>1</version>
+ <relativePath>../../parent</relativePath>
+ </parent>
+
+ <artifactId>ignite-storm</artifactId>
+ <version>1.5.1.final-SNAPSHOT</version>
+ <url>http://ignite.apache.org</url>
+
+ <properties>
+ <storm.version>0.10.0</storm.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>storm-core</artifactId>
+ <version>${storm.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.zookeeper</groupId>
+ <artifactId>zookeeper</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-log4j</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-spring</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/src/main/java/org/apache/ignite/stream/storm/StormStreamer.java
----------------------------------------------------------------------
diff --git a/modules/storm/src/main/java/org/apache/ignite/stream/storm/StormStreamer.java b/modules/storm/src/main/java/org/apache/ignite/stream/storm/StormStreamer.java
new file mode 100644
index 0000000..bdaec0b
--- /dev/null
+++ b/modules/storm/src/main/java/org/apache/ignite/stream/storm/StormStreamer.java
@@ -0,0 +1,304 @@
+/*
+ * 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.stream.storm;
+
+import backtype.storm.task.OutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Tuple;
+import java.util.Map;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.ignite.stream.StreamAdapter;
+
+/**
+ * Apache Storm streamer implemented as a Storm bolt.
+ * Obtaining data from other bolts and spouts is done by the specified tuple field.
+ */
+public class StormStreamer<K, V> extends StreamAdapter<Tuple, K, V> implements IRichBolt {
+ /** Default flush frequency. */
+ private static final long DFLT_FLUSH_FREQ = 10000L;
+
+ /** Default Ignite tuple field. */
+ private static final String DFLT_TUPLE_FIELD = "ignite";
+
+ /** Logger. */
+ private IgniteLogger log;
+
+ /** Field by which tuple data is obtained in topology. */
+ private String igniteTupleField = DFLT_TUPLE_FIELD;
+
+ /** Automatic flush frequency. */
+ private long autoFlushFrequency = DFLT_FLUSH_FREQ;
+
+ /** Enables overwriting existing values in cache. */
+ private boolean allowOverwrite = false;
+
+ /** Flag for stopped state. */
+ private static volatile boolean stopped = true;
+
+ /** Storm output collector. */
+ private OutputCollector collector;
+
+ /** Ignite grid configuration file. */
+ private static String igniteConfigFile;
+
+ /** Cache name. */
+ private static String cacheName;
+
+ /**
+ * Gets Ignite tuple field, by which tuple data is obtained in topology.
+ *
+ * @return Tuple field.
+ */
+ public String getIgniteTupleField() {
+ return igniteTupleField;
+ }
+
+ /**
+ * Names Ignite tuple field, by which tuple data is obtained in topology.
+ *
+ * @param igniteTupleField Name of tuple field.
+ */
+ public void setIgniteTupleField(String igniteTupleField) {
+ this.igniteTupleField = igniteTupleField;
+ }
+
+ /**
+ * Gets the cache name.
+ *
+ * @return Cache name.
+ */
+ public String getCacheName() {
+ return cacheName;
+ }
+
+ /**
+ * Sets the cache name.
+ *
+ * @param cacheName Cache name.
+ */
+ public void setCacheName(String cacheName) {
+ StormStreamer.cacheName = cacheName;
+ }
+
+ /**
+ * Gets Ignite configuration file.
+ *
+ * @return Configuration file.
+ */
+ public String getIgniteConfigFile() {
+ return igniteConfigFile;
+ }
+
+ /**
+ * Specifies Ignite configuration file.
+ *
+ * @param igniteConfigFile Ignite config file.
+ */
+ public void setIgniteConfigFile(String igniteConfigFile) {
+ StormStreamer.igniteConfigFile = igniteConfigFile;
+ }
+
+ /**
+ * Obtains data flush frequency.
+ *
+ * @return Flush frequency.
+ */
+ public long getAutoFlushFrequency() {
+ return autoFlushFrequency;
+ }
+
+ /**
+ * Specifies data flush frequency into the grid.
+ *
+ * @param autoFlushFrequency Flush frequency.
+ */
+ public void setAutoFlushFrequency(long autoFlushFrequency) {
+ this.autoFlushFrequency = autoFlushFrequency;
+ }
+
+ /**
+ * Obtains flag for enabling overwriting existing values in cache.
+ *
+ * @return True if overwriting is allowed, false otherwise.
+ */
+ public boolean getAllowOverwrite() {
+ return allowOverwrite;
+ }
+
+ /**
+ * Enables overwriting existing values in cache.
+ *
+ * @param allowOverwrite Flag value.
+ */
+ public void setAllowOverwrite(boolean allowOverwrite) {
+ this.allowOverwrite = allowOverwrite;
+ }
+
+ /**
+ * Starts streamer.
+ *
+ * @throws IgniteException If failed.
+ */
+ @SuppressWarnings("unchecked")
+ public void start() throws IgniteException {
+ A.notNull(igniteConfigFile, "Ignite config file");
+ A.notNull(cacheName, "Cache name");
+ A.notNull(igniteTupleField, "Ignite tuple field");
+
+ setIgnite(StreamerContext.getIgnite());
+
+ final IgniteDataStreamer<K, V> dataStreamer = StreamerContext.getStreamer();
+ dataStreamer.autoFlushFrequency(autoFlushFrequency);
+ dataStreamer.allowOverwrite(allowOverwrite);
+
+ setStreamer(dataStreamer);
+
+ log = getIgnite().log();
+
+ stopped = false;
+ }
+
+ /**
+ * Stops streamer.
+ *
+ * @throws IgniteException If failed.
+ */
+ public void stop() throws IgniteException {
+ if (stopped)
+ return;
+
+ stopped = true;
+
+ getIgnite().<K, V>dataStreamer(cacheName).close(true);
+
+ getIgnite().close();
+ }
+
+ /**
+ * Initializes Ignite client instance from a configuration file and declares the output collector of the bolt.
+ *
+ * @param map Map derived from topology.
+ * @param topologyContext Context topology in storm.
+ * @param collector Output collector.
+ */
+ @Override
+ public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {
+ start();
+
+ this.collector = collector;
+ }
+
+ /**
+ * Transfers data into grid.
+ *
+ * @param tuple Storm tuple.
+ */
+ @SuppressWarnings("unchecked")
+ @Override
+ public void execute(Tuple tuple) {
+ if (stopped)
+ return;
+
+ if (!(tuple.getValueByField(igniteTupleField) instanceof Map))
+ throw new IgniteException("Map as a streamer input is expected!");
+
+ final Map<K, V> gridVals = (Map<K, V>)tuple.getValueByField(igniteTupleField);
+
+ try {
+ if (log.isDebugEnabled())
+ log.debug("Tuple (id:" + tuple.getMessageId() + ") from storm: " + gridVals);
+
+ getStreamer().addData(gridVals);
+
+ collector.ack(tuple);
+ }
+ catch (Exception e) {
+ log.error("Error while processing tuple of " + gridVals, e);
+ collector.fail(tuple);
+ }
+ }
+
+ /**
+ * Cleans up the streamer when the bolt is going to shutdown.
+ */
+ @Override
+ public void cleanup() {
+ stop();
+ }
+
+ /**
+ * Normally declares output fields for the stream of the topology.
+ * Empty because we have no tuples for any further processing.
+ *
+ * @param declarer OutputFieldsDeclarer.
+ */
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ // No-op.
+ }
+
+ /**
+ * Not used.
+ *
+ * @return Configurations.
+ */
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+
+ /**
+ * Streamer context initializing grid and data streamer instances on demand.
+ */
+ public static class StreamerContext {
+ /** Constructor. */
+ private StreamerContext() {
+ }
+
+ /** Instance holder. */
+ private static class Holder {
+ private static final Ignite IGNITE = Ignition.start(igniteConfigFile);
+ private static final IgniteDataStreamer STREAMER = IGNITE.dataStreamer(cacheName);
+ }
+
+ /**
+ * Obtains grid instance.
+ *
+ * @return Grid instance.
+ */
+ public static Ignite getIgnite() {
+ return Holder.IGNITE;
+ }
+
+ /**
+ * Obtains data streamer instance.
+ *
+ * @return Data streamer instance.
+ */
+ public static IgniteDataStreamer getStreamer() {
+ return Holder.STREAMER;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/src/main/java/org/apache/ignite/stream/storm/package-info.java
----------------------------------------------------------------------
diff --git a/modules/storm/src/main/java/org/apache/ignite/stream/storm/package-info.java b/modules/storm/src/main/java/org/apache/ignite/stream/storm/package-info.java
new file mode 100644
index 0000000..67b2d17
--- /dev/null
+++ b/modules/storm/src/main/java/org/apache/ignite/stream/storm/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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 description. -->
+ * StormStreamer -- integration with Apache Storm.
+ */
+package org.apache.ignite.stream.storm;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/src/test/java/org/apache/ignite/stream/storm/IgniteStormStreamerSelfTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/storm/src/test/java/org/apache/ignite/stream/storm/IgniteStormStreamerSelfTestSuite.java b/modules/storm/src/test/java/org/apache/ignite/stream/storm/IgniteStormStreamerSelfTestSuite.java
new file mode 100644
index 0000000..25546e5
--- /dev/null
+++ b/modules/storm/src/test/java/org/apache/ignite/stream/storm/IgniteStormStreamerSelfTestSuite.java
@@ -0,0 +1,38 @@
+/*
+ * 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.stream.storm;
+
+import junit.framework.TestSuite;
+
+/**
+ * Apache Storm streamer tests.
+ */
+public class IgniteStormStreamerSelfTestSuite extends TestSuite {
+
+ /**
+ * @return Test suite.
+ * @throws Exception Thrown in case of the failure.
+ */
+ public static TestSuite suite() throws Exception {
+ TestSuite suite = new TestSuite("Apache Storm streamer Test Suite");
+
+ suite.addTest(new TestSuite(StormIgniteStreamerSelfTest.class));
+
+ return suite;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/src/test/java/org/apache/ignite/stream/storm/StormIgniteStreamerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/storm/src/test/java/org/apache/ignite/stream/storm/StormIgniteStreamerSelfTest.java b/modules/storm/src/test/java/org/apache/ignite/stream/storm/StormIgniteStreamerSelfTest.java
new file mode 100644
index 0000000..0ce4c6e
--- /dev/null
+++ b/modules/storm/src/test/java/org/apache/ignite/stream/storm/StormIgniteStreamerSelfTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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.stream.storm;
+
+import backtype.storm.Config;
+import backtype.storm.ILocalCluster;
+import backtype.storm.Testing;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.testing.CompleteTopologyParam;
+import backtype.storm.testing.MkClusterParam;
+import backtype.storm.testing.MockedSources;
+import backtype.storm.testing.TestJob;
+import backtype.storm.topology.TopologyBuilder;
+import backtype.storm.tuple.Values;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.NotNull;
+
+import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
+
+/**
+ * Tests for {@link StormStreamer}.
+ */
+public class StormIgniteStreamerSelfTest extends GridCommonAbstractTest {
+ /** Cache name. */
+ private static final String TEST_CACHE = "testCache";
+
+ /** Ignite test configuration file. */
+ private static final String GRID_CONF_FILE = "modules/storm/src/test/resources/example-ignite.xml";
+
+ /** Ignite instance. */
+ private Ignite ignite;
+
+ /** Parallelization in Storm. */
+ private static final int STORM_EXECUTORS = 2;
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override protected void beforeTest() throws Exception {
+ IgniteConfiguration cfg = loadConfiguration(GRID_CONF_FILE);
+
+ cfg.setClientMode(false);
+
+ ignite = startGrid("igniteServerNode", cfg);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * Tests for the streamer bolt. Ignite started in bolt based on what is specified in the configuration file.
+ *
+ * @throws TimeoutException
+ * @throws InterruptedException
+ */
+ public void testStormStreamerIgniteBolt() throws TimeoutException, InterruptedException {
+ final StormStreamer<String, String> stormStreamer = new StormStreamer<>();
+
+ stormStreamer.setAutoFlushFrequency(10L);
+ stormStreamer.setAllowOverwrite(true);
+ stormStreamer.setCacheName(TEST_CACHE);
+ stormStreamer.setIgniteTupleField(TestStormSpout.IGNITE_TUPLE_FIELD);
+ stormStreamer.setIgniteConfigFile(GRID_CONF_FILE);
+
+ Config daemonConf = new Config();
+
+ daemonConf.put(Config.STORM_LOCAL_MODE_ZMQ, false);
+
+ MkClusterParam mkClusterParam = new MkClusterParam();
+
+ mkClusterParam.setDaemonConf(daemonConf);
+ mkClusterParam.setSupervisors(4);
+
+ final CountDownLatch latch = new CountDownLatch(TestStormSpout.CNT);
+
+ IgniteBiPredicate<UUID, CacheEvent> putLsnr = new IgniteBiPredicate<UUID, CacheEvent>() {
+ @Override public boolean apply(UUID uuid, CacheEvent evt) {
+ assert evt != null;
+
+ latch.countDown();
+
+ return true;
+ }
+ };
+
+ final UUID putLsnrId = ignite.events(ignite.cluster().forCacheNodes(TEST_CACHE))
+ .remoteListen(putLsnr, null, EVT_CACHE_OBJECT_PUT);
+
+ Testing.withSimulatedTimeLocalCluster(mkClusterParam, new TestJob() {
+ @Override public void run(ILocalCluster cluster) throws IOException, InterruptedException {
+ // Creates a test topology.
+ TopologyBuilder builder = new TopologyBuilder();
+
+ TestStormSpout testStormSpout = new TestStormSpout();
+
+ builder.setSpout("test-spout", testStormSpout);
+ builder.setBolt("ignite-bolt", stormStreamer, STORM_EXECUTORS).shuffleGrouping("test-spout");
+
+ StormTopology topology = builder.createTopology();
+
+ // Prepares a mock data for the spout.
+ MockedSources mockedSources = new MockedSources();
+
+ mockedSources.addMockData("test-spout", getMockData());
+
+ // Prepares the config.
+ Config conf = new Config();
+
+ conf.setMessageTimeoutSecs(10);
+
+ IgniteCache<Integer, String> cache = ignite.cache(TEST_CACHE);
+
+ CompleteTopologyParam completeTopologyParam = new CompleteTopologyParam();
+
+ completeTopologyParam.setTimeoutMs(10000);
+ completeTopologyParam.setMockedSources(mockedSources);
+ completeTopologyParam.setStormConf(conf);
+
+ // Checks the cache doesn't contain any entries yet.
+ assertEquals(0, cache.size(CachePeekMode.PRIMARY));
+
+ Testing.completeTopology(cluster, topology, completeTopologyParam);
+
+ // Checks events successfully processed in 20 seconds.
+ assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+ ignite.events(ignite.cluster().forCacheNodes(TEST_CACHE)).stopRemoteListen(putLsnrId);
+
+ // Validates all entries are in the cache.
+ assertEquals(TestStormSpout.CNT, cache.size(CachePeekMode.PRIMARY));
+
+ for (Map.Entry<Integer, String> entry : TestStormSpout.getKeyValMap().entrySet())
+ assertEquals(entry.getValue(), cache.get(entry.getKey()));
+ }
+ }
+ );
+ }
+
+ /**
+ * Prepares entry values for test input.
+ *
+ * @return Array of entry values.
+ */
+ @NotNull
+ private static Values[] getMockData() {
+ final int SIZE = 10;
+
+ ArrayList<Values> mockData = new ArrayList<>();
+
+ for (int i = 0; i < TestStormSpout.CNT; i += SIZE)
+ mockData.add(new Values(TestStormSpout.getKeyValMap().subMap(i, i + SIZE)));
+
+ return mockData.toArray(new Values[mockData.size()]);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/src/test/java/org/apache/ignite/stream/storm/TestStormSpout.java
----------------------------------------------------------------------
diff --git a/modules/storm/src/test/java/org/apache/ignite/stream/storm/TestStormSpout.java b/modules/storm/src/test/java/org/apache/ignite/stream/storm/TestStormSpout.java
new file mode 100644
index 0000000..a006ca7
--- /dev/null
+++ b/modules/storm/src/test/java/org/apache/ignite/stream/storm/TestStormSpout.java
@@ -0,0 +1,141 @@
+/*
+ * 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.stream.storm;
+
+import backtype.storm.spout.SpoutOutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * Testing Storm spout.
+ */
+public class TestStormSpout implements IRichSpout {
+
+ /** Field by which tuple data is obtained by the streamer. */
+ public static final String IGNITE_TUPLE_FIELD = "test-field";
+
+ /** Number of outgoing tuples. */
+ public static final int CNT = 500;
+
+ /** Spout message prefix. */
+ private static final String VAL_PREFIX = "v:";
+
+ /** Spout output collector. */
+ private SpoutOutputCollector collector;
+
+ /** Map of messages to be sent. */
+ private static TreeMap<Integer, String> keyValMap;
+
+ static {
+ keyValMap = generateKeyValMap();
+ }
+
+ /**
+ * Declares the output field for the component.
+ *
+ * @param outputFieldsDeclarer Declarer to declare fields on.
+ */
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+ outputFieldsDeclarer.declare(new Fields(IGNITE_TUPLE_FIELD));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
+ this.collector = spoutOutputCollector;
+ }
+
+ /**
+ * Requests to emit a tuple to the output collector.
+ */
+ @Override
+ public void nextTuple() {
+ // No-op. Sent via mocked sources.
+ }
+
+ /**
+ * Generates key,value pair to emit to bolt({@link StormStreamer}).
+ *
+ * @return Key, value pair.
+ */
+ private static TreeMap<Integer, String> generateKeyValMap() {
+ TreeMap<Integer, String> keyValMap = new TreeMap<>();
+
+ for (int evt = 0; evt < CNT; evt++) {
+ String msg = VAL_PREFIX + evt;
+
+ keyValMap.put(evt, msg);
+ }
+
+ return keyValMap;
+ }
+
+ public static TreeMap<Integer, String> getKeyValMap() {
+ return keyValMap;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void ack(Object msgId) {
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void fail(Object msgId) {
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void close() {
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void activate() {
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void deactivate() {
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/modules/storm/src/test/resources/example-ignite.xml
----------------------------------------------------------------------
diff --git a/modules/storm/src/test/resources/example-ignite.xml b/modules/storm/src/test/resources/example-ignite.xml
new file mode 100644
index 0000000..fbb05d3
--- /dev/null
+++ b/modules/storm/src/test/resources/example-ignite.xml
@@ -0,0 +1,71 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+ 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.
+-->
+
+<!--
+ Ignite configuration with all defaults and enabled p2p deployment and enabled events.
+ Used for testing IgniteSink running Ignite in a client mode.
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns:util="http://www.springframework.org/schema/util"
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans
+ http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://www.springframework.org/schema/util
+ http://www.springframework.org/schema/util/spring-util.xsd">
+ <bean id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
+ <!-- Enable client mode. -->
+ <property name="clientMode" value="true"/>
+
+ <!-- Cache accessed from IgniteSink. -->
+ <property name="cacheConfiguration">
+ <list>
+ <!-- Partitioned cache example configuration with configurations adjusted to server nodes'. -->
+ <bean class="org.apache.ignite.configuration.CacheConfiguration">
+ <property name="atomicityMode" value="ATOMIC"/>
+
+ <property name="name" value="testCache"/>
+ </bean>
+ </list>
+ </property>
+
+ <!-- Enable cache events. -->
+ <property name="includeEventTypes">
+ <list>
+ <!-- Cache events (only EVT_CACHE_OBJECT_PUT for tests). -->
+ <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
+ </list>
+ </property>
+
+ <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
+ <property name="discoverySpi">
+ <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
+ <property name="ipFinder">
+ <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">
+ <property name="addresses">
+ <list>
+ <value>127.0.0.1:47500</value>
+ </list>
+ </property>
+ </bean>
+ </property>
+ </bean>
+ </property>
+ </bean>
+</beans>
http://git-wip-us.apache.org/repos/asf/ignite/blob/3a0cb51f/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 5c6e87c..0a74c47 100644
--- a/pom.xml
+++ b/pom.xml
@@ -80,6 +80,7 @@
<module>modules/mqtt</module>
<module>modules/zookeeper</module>
<module>modules/camel</module>
+ <module>modules/storm</module>
<module>modules/osgi-paxlogging</module>
<module>modules/osgi-karaf</module>
<module>modules/osgi</module>
[08/16] ignite git commit: IGNITE-2328 - Added the ticket test to the
suite.
Posted by sb...@apache.org.
IGNITE-2328 - Added the ticket test to the suite.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c4ff7111
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c4ff7111
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c4ff7111
Branch: refs/heads/ignite-1811
Commit: c4ff7111638e939ee395111840fd786314527003
Parents: 6524c79
Author: Alexey Goncharuk <al...@gmail.com>
Authored: Fri Jan 15 10:31:37 2016 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Fri Jan 15 10:31:37 2016 +0300
----------------------------------------------------------------------
.../java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c4ff7111/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
index 0f86c4c..442e116 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
@@ -20,6 +20,7 @@ package org.apache.ignite.testsuites;
import junit.framework.TestSuite;
import org.apache.ignite.internal.processors.cache.CacheNearReaderUpdateTest;
import org.apache.ignite.internal.processors.cache.CacheSerializableTransactionsTest;
+import org.apache.ignite.internal.processors.cache.IgniteCacheStoreCollectionTest;
/**
* Test suite.
@@ -34,6 +35,7 @@ public class IgniteCacheTestSuite5 extends TestSuite {
suite.addTestSuite(CacheSerializableTransactionsTest.class);
suite.addTestSuite(CacheNearReaderUpdateTest.class);
+ suite.addTestSuite(IgniteCacheStoreCollectionTest.class);
return suite;
}
[15/16] ignite git commit: Merge remote-tracking branch
'remotes/origin/master' into ignite-1811
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/master' into ignite-1811
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/98d470a3
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/98d470a3
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/98d470a3
Branch: refs/heads/ignite-1811
Commit: 98d470a3e90440364e5cab85d32c3bb26a68d151
Parents: 56b9da0 3a0cb51
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jan 15 14:56:22 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jan 15 14:56:22 2016 +0300
----------------------------------------------------------------------
modules/storm/README.txt | 37 +++
modules/storm/licenses/apache-2.0.txt | 202 ++++++++++++
modules/storm/pom.xml | 104 +++++++
.../ignite/stream/storm/StormStreamer.java | 304 +++++++++++++++++++
.../ignite/stream/storm/package-info.java | 22 ++
.../storm/IgniteStormStreamerSelfTestSuite.java | 38 +++
.../storm/StormIgniteStreamerSelfTest.java | 184 +++++++++++
.../ignite/stream/storm/TestStormSpout.java | 141 +++++++++
.../storm/src/test/resources/example-ignite.xml | 71 +++++
pom.xml | 1 +
10 files changed, 1104 insertions(+)
----------------------------------------------------------------------
[10/16] ignite git commit: Merge remote-tracking branch
'remotes/origin/master' into ignite-1811
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/master' into ignite-1811
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/97d0c688
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/97d0c688
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/97d0c688
Branch: refs/heads/ignite-1811
Commit: 97d0c6883c62c5349b0a5b83973ab61a6ba23bcd
Parents: 0823cd6 8b0c59a
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jan 15 11:14:47 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jan 15 11:14:47 2016 +0300
----------------------------------------------------------------------
.../ignite/internal/IgniteVersionUtils.java | 4 +-
.../internal/direct/DirectMessageReader.java | 11 +++++
.../internal/direct/DirectMessageWriter.java | 13 +++++
.../direct/state/DirectMessageState.java | 10 ++++
.../stream/v1/DirectByteBufferStreamImplV1.java | 8 +++
.../stream/v2/DirectByteBufferStreamImplV2.java | 8 +++
.../internal/util/nio/GridDirectParser.java | 52 +++++++++++++-------
.../communication/tcp/TcpCommunicationSpi.java | 9 +++-
.../tcp/ipfinder/TcpDiscoveryIpFinder.java | 14 ++++--
.../startup/GridRandomCommandLineLoader.java | 4 +-
10 files changed, 107 insertions(+), 26 deletions(-)
----------------------------------------------------------------------