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 2015/06/26 18:43:15 UTC
[01/20] incubator-ignite git commit: # GG-10404 Added check for empty
topology for query cleanup task.
Repository: incubator-ignite
Updated Branches:
refs/heads/ignite-428 c1505098f -> ad5e99eb7 (forced update)
# GG-10404 Added check for empty topology for query cleanup task.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/32697ab0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/32697ab0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/32697ab0
Branch: refs/heads/ignite-428
Commit: 32697ab08071c708c9f8b4c1b70b2ee4ab62d858
Parents: f0e74c0
Author: AKuznetsov <ak...@gridgain.com>
Authored: Fri Jun 12 11:19:37 2015 +0700
Committer: AKuznetsov <ak...@gridgain.com>
Committed: Fri Jun 12 11:19:37 2015 +0700
----------------------------------------------------------------------
.../connection/GridClientNioTcpConnection.java | 7 ++++-
.../visor/query/VisorQueryCleanupTask.java | 14 +++++++++
.../visor/util/VisorEmptyTopologyException.java | 33 ++++++++++++++++++++
3 files changed, 53 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32697ab0/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
index d247e05..67709b8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
@@ -750,7 +750,12 @@ public class GridClientNioTcpConnection extends GridClientConnection {
new GridClientFutureCallback<GridClientTaskResultBean, R>() {
@Override public R onComplete(GridClientFuture<GridClientTaskResultBean> fut)
throws GridClientException {
- return fut.get().getResult();
+ GridClientTaskResultBean resBean = fut.get();
+
+ if (resBean != null)
+ return resBean.getResult();
+ else
+ return null;
}
});
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32697ab0/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
index 722ad91..b9a55e1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
@@ -22,6 +22,7 @@ import org.apache.ignite.compute.*;
import org.apache.ignite.internal.processors.task.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.internal.visor.*;
+import org.apache.ignite.internal.visor.util.*;
import org.jetbrains.annotations.*;
import java.util.*;
@@ -47,6 +48,9 @@ public class VisorQueryCleanupTask extends VisorMultiNodeTask<Map<UUID, Collecti
@Nullable VisorTaskArgument<Map<UUID, Collection<String>>> arg) {
Set<UUID> nodeIds = taskArg.keySet();
+ if (nodeIds.isEmpty())
+ throw new VisorEmptyTopologyException("Nothing to clear. List with node IDs is empty!");
+
Map<ComputeJob, ClusterNode> map = U.newHashMap(nodeIds.size());
try {
@@ -54,6 +58,16 @@ public class VisorQueryCleanupTask extends VisorMultiNodeTask<Map<UUID, Collecti
if (nodeIds.contains(node.id()))
map.put(new VisorQueryCleanupJob(taskArg.get(node.id()), debug), node);
+ if (map.isEmpty()) {
+ String notFoundNodes = "";
+
+ for (UUID nid : nodeIds)
+ notFoundNodes = notFoundNodes + (notFoundNodes.isEmpty() ? "" : ",") + U.id8(nid);
+
+ throw new VisorEmptyTopologyException("Failed to clear query results. Nodes are not available: [" +
+ notFoundNodes + "]");
+ }
+
return map;
}
finally {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32697ab0/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorEmptyTopologyException.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorEmptyTopologyException.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorEmptyTopologyException.java
new file mode 100644
index 0000000..fda1bd7
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorEmptyTopologyException.java
@@ -0,0 +1,33 @@
+/*
+ * 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.visor.util;
+
+import org.apache.ignite.*;
+
+/**
+ * Marker exception for indication of empty topology in Visor tasks.
+ */
+public class VisorEmptyTopologyException extends IgniteException {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** @inheritDoc */
+ public VisorEmptyTopologyException(String msg) {
+ super(msg);
+ }
+}
[09/20] incubator-ignite git commit: # ignite-1017
Posted by sb...@apache.org.
# ignite-1017
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/5b237e15
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/5b237e15
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/5b237e15
Branch: refs/heads/ignite-428
Commit: 5b237e15ccab0e48b69045cd1c55698ac26b8cd0
Parents: 01d842a
Author: Atri <at...@gmail.com>
Authored: Tue Jun 23 19:02:23 2015 +0530
Committer: ashutak <as...@gridgain.com>
Committed: Fri Jun 26 14:11:35 2015 +0300
----------------------------------------------------------------------
.../org/apache/ignite/cluster/ClusterGroup.java | 16 ++++++++++++++++
.../internal/cluster/ClusterGroupAdapter.java | 7 +++++++
.../internal/cluster/IgniteClusterAsyncImpl.java | 5 +++++
3 files changed, 28 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5b237e15/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java b/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
index 2f43fc6..9627f76 100644
--- a/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
+++ b/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
@@ -131,10 +131,26 @@ public interface ClusterGroup {
* @param name Name of the attribute.
* @param val Optional attribute value to match.
* @return Cluster group for nodes containing specified attribute.
+ *
+ * @deprecated use {@link ClusterGroup#forAttribute(String name, @Nullable Object val}
*/
public ClusterGroup forAttribute(String name, @Nullable String val);
/**
+ * Creates a new cluster group for nodes containing given name and value
+ * specified in user attributes.
+ * <p>
+ * User attributes for every node are optional and can be specified in
+ * grid node configuration. See {@link IgniteConfiguration#getUserAttributes()}
+ * for more information.
+ *
+ * @param name Name of the attribute.
+ * @param val Optional attribute value to match.
+ * @return Cluster group for nodes containing specified attribute.
+ */
+ public ClusterGroup forAttribute(String name, @Nullable Object val);
+
+ /**
* Creates a cluster group of nodes started in server mode.
*
* @see Ignition#setClientMode(boolean)
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5b237e15/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
index b940017..414f5ba 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
@@ -352,6 +352,13 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
}
/** {@inheritDoc} */
+ @Override public final ClusterGroup forAttribute(String name, @Nullable final Object val) {
+ A.notNull(name, "n");
+
+ return forPredicate(new AttributeFilter(name, val));
+ }
+
+ /** {@inheritDoc} */
@Override public ClusterGroup forServers() {
return forPredicate(new AttributeFilter(IgniteNodeAttributes.ATTR_CLIENT_MODE, false));
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5b237e15/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
index 6e68527..f676261 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
@@ -197,6 +197,11 @@ public class IgniteClusterAsyncImpl extends AsyncSupportAdapter<IgniteCluster>
}
/** {@inheritDoc} */
+ @Override public ClusterGroup forAttribute(String name, @Nullable Object val) {
+ return cluster.forAttribute(name, val);
+ }
+
+ /** {@inheritDoc} */
@Override public ClusterGroup forServers() {
return cluster.forServers();
}
[19/20] incubator-ignite git commit: ignite-428 Review
Posted by sb...@apache.org.
ignite-428 Review
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/6ca6a0c7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/6ca6a0c7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/6ca6a0c7
Branch: refs/heads/ignite-428
Commit: 6ca6a0c77e397a7fd9f9038f2808ce5d38b056fb
Parents: da07744
Author: agura <ag...@gridgain.com>
Authored: Thu Jun 25 22:18:00 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Fri Jun 26 19:04:41 2015 +0300
----------------------------------------------------------------------
modules/kafka/pom.xml | 18 +-
.../ignite/stream/kafka/KafkaStreamer.java | 72 +++---
.../stream/kafka/KafkaEmbeddedBroker.java | 251 ++++++++++---------
.../kafka/KafkaIgniteStreamerSelfTest.java | 131 +++++-----
.../ignite/stream/kafka/SimplePartitioner.java | 27 +-
5 files changed, 245 insertions(+), 254 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6ca6a0c7/modules/kafka/pom.xml
----------------------------------------------------------------------
diff --git a/modules/kafka/pom.xml b/modules/kafka/pom.xml
index 165ec1c..43909bc 100644
--- a/modules/kafka/pom.xml
+++ b/modules/kafka/pom.xml
@@ -31,7 +31,7 @@
</parent>
<artifactId>ignite-kafka</artifactId>
- <version>1.1.1-SNAPSHOT</version>
+ <version>1.1.6-SNAPSHOT</version>
<dependencies>
<dependency>
@@ -39,6 +39,7 @@
<artifactId>ignite-core</artifactId>
<version>${project.version}</version>
</dependency>
+
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
@@ -66,6 +67,7 @@
</exclusion>
</exclusions>
</dependency>
+
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
@@ -73,12 +75,6 @@
</dependency>
<dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- <version>2.4</version>
- </dependency>
-
- <dependency>
<groupId>org.apache.ignite</groupId>
<artifactId>ignite-log4j</artifactId>
<version>${project.version}</version>
@@ -90,7 +86,6 @@
<version>4.2</version>
</dependency>
-
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
@@ -103,13 +98,6 @@
</dependency>
<dependency>
- <groupId>commons-beanutils</groupId>
- <artifactId>commons-beanutils</artifactId>
- <version>1.8.3</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
<groupId>org.apache.ignite</groupId>
<artifactId>ignite-spring</artifactId>
<version>${project.version}</version>
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6ca6a0c7/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java b/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
index e0240ce..5761209 100644
--- a/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
+++ b/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
@@ -30,17 +30,15 @@ import java.util.*;
import java.util.concurrent.*;
/**
- * Server that subscribes to topic messages from Kafka broker, streams its to key-value pairs into {@link
- * org.apache.ignite.IgniteDataStreamer} instance.
+ * Server that subscribes to topic messages from Kafka broker and streams its to key-value pairs into
+ * {@link IgniteDataStreamer} instance.
* <p>
- * Uses Kafka's High Level Consumer API to read messages from Kafka
+ * Uses Kafka's High Level Consumer API to read messages from Kafka.
*
* @see <a href="https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example">Consumer Consumer Group
* Example</a>
*/
-public class KafkaStreamer<T, K, V>
- extends StreamAdapter<T, K, V> {
-
+public class KafkaStreamer<T, K, V> extends StreamAdapter<T, K, V> {
/** Logger. */
private IgniteLogger log;
@@ -53,61 +51,61 @@ public class KafkaStreamer<T, K, V>
/** Number of threads to process kafka streams. */
private int threads;
- /** Kafka Consumer Config. */
- private ConsumerConfig consumerConfig;
+ /** Kafka consumer config. */
+ private ConsumerConfig consumerCfg;
- /** Key Decoder. */
+ /** Key decoder. */
private Decoder<K> keyDecoder;
- /** Value Decoder. */
- private Decoder<V> valueDecoder;
+ /** Value decoder. */
+ private Decoder<V> valDecoder;
- /** Kafka Consumer connector. */
+ /** Kafka consumer connector. */
private ConsumerConnector consumer;
/**
- * Sets the topic.
+ * Sets the topic name.
*
- * @param topic Topic Name.
+ * @param topic Topic name.
*/
- public void setTopic(final String topic) {
+ public void setTopic(String topic) {
this.topic = topic;
}
/**
* Sets the threads.
*
- * @param threads Number of Threads.
+ * @param threads Number of threads.
*/
- public void setThreads(final int threads) {
+ public void setThreads(int threads) {
this.threads = threads;
}
/**
* Sets the consumer config.
*
- * @param consumerConfig Consumer configuration.
+ * @param consumerCfg Consumer configuration.
*/
- public void setConsumerConfig(final ConsumerConfig consumerConfig) {
- this.consumerConfig = consumerConfig;
+ public void setConsumerConfig(ConsumerConfig consumerCfg) {
+ this.consumerCfg = consumerCfg;
}
/**
* Sets the key decoder.
*
- * @param keyDecoder Key Decoder.
+ * @param keyDecoder Key decoder.
*/
- public void setKeyDecoder(final Decoder<K> keyDecoder) {
+ public void setKeyDecoder(Decoder<K> keyDecoder) {
this.keyDecoder = keyDecoder;
}
/**
* Sets the value decoder.
*
- * @param valueDecoder Value Decoder
+ * @param valDecoder Value decoder.
*/
- public void setValueDecoder(final Decoder<V> valueDecoder) {
- this.valueDecoder = valueDecoder;
+ public void setValueDecoder(Decoder<V> valDecoder) {
+ this.valDecoder = valDecoder;
}
/**
@@ -120,19 +118,20 @@ public class KafkaStreamer<T, K, V>
A.notNull(getIgnite(), "ignite");
A.notNull(topic, "topic");
A.notNull(keyDecoder, "key decoder");
- A.notNull(valueDecoder, "value decoder");
- A.notNull(consumerConfig, "kafka consumer config");
+ A.notNull(valDecoder, "value decoder");
+ A.notNull(consumerCfg, "kafka consumer config");
A.ensure(threads > 0, "threads > 0");
log = getIgnite().log();
- consumer = kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);
+ consumer = kafka.consumer.Consumer.createJavaConsumerConnector(consumerCfg);
+
+ Map<String, Integer> topicCntMap = new HashMap<>();
- Map<String, Integer> topicCountMap = new HashMap<>();
- topicCountMap.put(topic, threads);
+ topicCntMap.put(topic, threads);
- Map<String, List<KafkaStream<K, V>>> consumerMap = consumer.createMessageStreams(topicCountMap, keyDecoder,
- valueDecoder);
+ Map<String, List<KafkaStream<K, V>>> consumerMap =
+ consumer.createMessageStreams(topicCntMap, keyDecoder, valDecoder);
List<KafkaStream<K, V>> streams = consumerMap.get(topic);
@@ -140,16 +139,11 @@ public class KafkaStreamer<T, K, V>
executor = Executors.newFixedThreadPool(threads);
// Now create an object to consume the messages.
- for (final KafkaStream<K,V> stream : streams) {
+ for (final KafkaStream<K, V> stream : streams) {
executor.submit(new Runnable() {
@Override public void run() {
-
- ConsumerIterator<K, V> it = stream.iterator();
-
- while (it.hasNext()) {
- final MessageAndMetadata<K, V> messageAndMetadata = it.next();
+ for (MessageAndMetadata<K, V> messageAndMetadata : stream)
getStreamer().addData(messageAndMetadata.key(), messageAndMetadata.message());
- }
}
});
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6ca6a0c7/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
index 28533f7..98b9e4c 100644
--- a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
@@ -17,7 +17,6 @@
package org.apache.ignite.stream.kafka;
-import org.apache.commons.io.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.zookeeper.server.*;
@@ -25,6 +24,7 @@ import kafka.admin.*;
import kafka.api.*;
import kafka.api.Request;
import kafka.producer.*;
+import kafka.serializer.*;
import kafka.server.*;
import kafka.utils.*;
import org.I0Itec.zkclient.*;
@@ -39,106 +39,104 @@ import java.util.concurrent.*;
* Kafka Embedded Broker.
*/
public class KafkaEmbeddedBroker {
-
- /** Default ZooKeeper Host. */
+ /** Default ZooKeeper host. */
private static final String ZK_HOST = "localhost";
- /** Broker Port. */
+ /** Broker port. */
private static final int BROKER_PORT = 9092;
- /** ZooKeeper Connection Timeout. */
+ /** ZooKeeper connection timeout. */
private static final int ZK_CONNECTION_TIMEOUT = 6000;
- /** ZooKeeper Session Timeout. */
+ /** ZooKeeper session timeout. */
private static final int ZK_SESSION_TIMEOUT = 6000;
/** ZooKeeper port. */
private static int zkPort = 0;
- /** Is ZooKeeper Ready. */
+ /** Is ZooKeeper ready. */
private boolean zkReady;
- /** Kafka Config. */
- private KafkaConfig brokerConfig;
+ /** Kafka config. */
+ private KafkaConfig brokerCfg;
- /** Kafka Server. */
- private KafkaServer kafkaServer;
+ /** Kafka server. */
+ private KafkaServer kafkaSrv;
- /** ZooKeeper Client. */
+ /** ZooKeeper client. */
private ZkClient zkClient;
/** Embedded ZooKeeper. */
private EmbeddedZooKeeper zooKeeper;
/**
- * Creates an embedded Kafka Broker.
+ * Creates an embedded Kafka broker.
*/
public KafkaEmbeddedBroker() {
try {
setupEmbeddedZooKeeper();
+
setupEmbeddedKafkaServer();
}
catch (IOException | InterruptedException e) {
- throw new RuntimeException("failed to start Kafka Broker " + e);
+ throw new RuntimeException("Failed to start Kafka broker " + e);
}
-
}
/**
- * @return ZooKeeper Address.
+ * @return ZooKeeper address.
*/
public static String getZKAddress() {
- return ZK_HOST + ":" + zkPort;
+ return ZK_HOST + ':' + zkPort;
}
/**
* Creates a Topic.
*
- * @param topic topic name
- * @param partitions number of paritions for the topic
- * @param replicationFactor replication factor
- * @throws TimeoutException
- * @throws InterruptedException
+ * @param topic Topic name.
+ * @param partitions Number of partitions for the topic.
+ * @param replicationFactor Replication factor.
+ * @throws TimeoutException If operation is timed out.
+ * @throws InterruptedException If interrupted.
*/
- public void createTopic(String topic, final int partitions, final int replicationFactor)
+ public void createTopic(String topic, int partitions, int replicationFactor)
throws TimeoutException, InterruptedException {
AdminUtils.createTopic(zkClient, topic, partitions, replicationFactor, new Properties());
+
waitUntilMetadataIsPropagated(topic, 0, 10000, 100);
}
/**
- * Sends message to Kafka Broker.
+ * Sends message to Kafka broker.
*
- * @param keyedMessages List of Keyed Messages.
+ * @param keyedMessages List of keyed messages.
* @return Producer used to send the message.
*/
- public Producer sendMessages(List<KeyedMessage<String, String>> keyedMessages) {
+ public Producer<String, String> sendMessages(List<KeyedMessage<String, String>> keyedMessages) {
Producer<String, String> producer = new Producer<>(getProducerConfig());
+
producer.send(scala.collection.JavaConversions.asScalaBuffer(keyedMessages));
+
return producer;
}
/**
- * Shuts down Kafka Broker.
- *
- * @throws IOException
+ * Shuts down Kafka broker.
*/
- public void shutdown()
- throws IOException {
-
+ public void shutdown() {
zkReady = false;
- if (kafkaServer != null)
- kafkaServer.shutdown();
+ if (kafkaSrv != null)
+ kafkaSrv.shutdown();
- List<String> logDirs = scala.collection.JavaConversions.asJavaList(brokerConfig.logDirs());
+ List<String> logDirs = scala.collection.JavaConversions.asJavaList(brokerCfg.logDirs());
- for (String logDir : logDirs) {
- FileUtils.deleteDirectory(new File(logDir));
- }
+ for (String logDir : logDirs)
+ U.delete(new File(logDir));
if (zkClient != null) {
zkClient.close();
+
zkClient = null;
}
@@ -148,16 +146,15 @@ public class KafkaEmbeddedBroker {
zooKeeper.shutdown();
}
catch (IOException e) {
- // ignore
+ // No-op.
}
zooKeeper = null;
}
-
}
/**
- * @return the Zookeeper Client
+ * @return ZooKeeper client.
*/
private ZkClient getZkClient() {
A.ensure(zkReady, "Zookeeper not setup yet");
@@ -169,102 +166,105 @@ public class KafkaEmbeddedBroker {
/**
* Checks if topic metadata is propagated.
*
- * @param topic topic name
- * @param partition partition
- * @return true if propagated otherwise false
+ * @param topic Topic name.
+ * @param part Partition.
+ * @return {@code True} if propagated, otherwise {@code false}.
*/
- private boolean isMetadataPropagated(final String topic, final int partition) {
- final scala.Option<PartitionStateInfo> partitionStateOption = kafkaServer.apis().metadataCache().getPartitionInfo(
- topic, partition);
- if (partitionStateOption.isDefined()) {
- final PartitionStateInfo partitionState = partitionStateOption.get();
- final LeaderAndIsr leaderAndInSyncReplicas = partitionState.leaderIsrAndControllerEpoch().leaderAndIsr();
-
- if (ZkUtils.getLeaderForPartition(getZkClient(), topic, partition) != null
- && Request.isValidBrokerId(leaderAndInSyncReplicas.leader())
- && leaderAndInSyncReplicas.isr().size() >= 1)
- return true;
+ private boolean isMetadataPropagated(String topic, int part) {
+ scala.Option<PartitionStateInfo> partStateOption =
+ kafkaSrv.apis().metadataCache().getPartitionInfo(topic, part);
- }
- return false;
+ if (!partStateOption.isDefined())
+ return false;
+
+ PartitionStateInfo partState = partStateOption.get();
+
+ LeaderAndIsr LeaderAndIsr = partState.leaderIsrAndControllerEpoch().leaderAndIsr();
+
+ return ZkUtils.getLeaderForPartition(getZkClient(), topic, part) != null &&
+ Request.isValidBrokerId(LeaderAndIsr.leader()) && LeaderAndIsr.isr().size() >= 1;
}
/**
* Waits until metadata is propagated.
*
- * @param topic topic name
- * @param partition partition
- * @param timeout timeout value in millis
- * @param interval interval in millis to sleep
- * @throws TimeoutException
- * @throws InterruptedException
+ * @param topic Topic name.
+ * @param part Partition.
+ * @param timeout Timeout value in millis.
+ * @param interval Interval in millis to sleep.
+ * @throws TimeoutException If operation is timed out.
+ * @throws InterruptedException If interrupted.
*/
- private void waitUntilMetadataIsPropagated(final String topic, final int partition, final long timeout,
- final long interval) throws TimeoutException, InterruptedException {
+ private void waitUntilMetadataIsPropagated(String topic, int part, long timeout, long interval)
+ throws TimeoutException, InterruptedException {
int attempt = 1;
- final long startTime = System.currentTimeMillis();
+
+ long startTime = System.currentTimeMillis();
while (true) {
- if (isMetadataPropagated(topic, partition))
+ if (isMetadataPropagated(topic, part))
return;
- final long duration = System.currentTimeMillis() - startTime;
+ long duration = System.currentTimeMillis() - startTime;
if (duration < timeout)
Thread.sleep(interval);
else
- throw new TimeoutException("metadata propagate timed out, attempt=" + attempt);
+ throw new TimeoutException("Metadata propagation is timed out, attempt " + attempt);
attempt++;
}
-
}
/**
- * Sets up embedded Kafka Server
+ * Sets up embedded Kafka server.
*
- * @throws IOException
+ * @throws IOException If failed.
*/
- private void setupEmbeddedKafkaServer()
- throws IOException {
+ private void setupEmbeddedKafkaServer() throws IOException {
A.ensure(zkReady, "Zookeeper should be setup before hand");
- brokerConfig = new KafkaConfig(getBrokerConfig());
- kafkaServer = new KafkaServer(brokerConfig, SystemTime$.MODULE$);
- kafkaServer.startup();
+ brokerCfg = new KafkaConfig(getBrokerConfig());
+
+ kafkaSrv = new KafkaServer(brokerCfg, SystemTime$.MODULE$);
+
+ kafkaSrv.startup();
}
/**
- * Sets up embedded zooKeeper
+ * Sets up embedded ZooKeeper.
*
- * @throws IOException
- * @throws InterruptedException
+ * @throws IOException If failed.
+ * @throws InterruptedException If interrupted.
*/
- private void setupEmbeddedZooKeeper()
- throws IOException, InterruptedException {
+ private void setupEmbeddedZooKeeper() throws IOException, InterruptedException {
EmbeddedZooKeeper zooKeeper = new EmbeddedZooKeeper(ZK_HOST, zkPort);
+
zooKeeper.startup();
+
zkPort = zooKeeper.getActualPort();
+
zkClient = new ZkClient(getZKAddress(), ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT, ZKStringSerializer$.MODULE$);
+
zkReady = true;
}
/**
- * @return Kafka Broker Address.
+ * @return Kafka broker address.
*/
private static String getBrokerAddress() {
- return ZK_HOST + ":" + BROKER_PORT;
+ return ZK_HOST + ':' + BROKER_PORT;
}
/**
- * Gets KafKa Brofer Config
+ * Gets Kafka broker config.
*
- * @return Kafka Broker Config
- * @throws IOException
+ * @return Kafka broker config.
+ * @throws IOException If failed.
*/
- private static Properties getBrokerConfig()
- throws IOException {
+ private static Properties getBrokerConfig() throws IOException {
Properties props = new Properties();
+
props.put("broker.id", "0");
props.put("host.name", ZK_HOST);
props.put("port", "" + BROKER_PORT);
@@ -272,60 +272,63 @@ public class KafkaEmbeddedBroker {
props.put("zookeeper.connect", getZKAddress());
props.put("log.flush.interval.messages", "1");
props.put("replica.socket.timeout.ms", "1500");
+
return props;
}
/**
- * @return Kafka Producer Config
+ * @return Kafka Producer config.
*/
private static ProducerConfig getProducerConfig() {
Properties props = new Properties();
+
props.put("metadata.broker.list", getBrokerAddress());
- props.put("serializer.class", "kafka.serializer.StringEncoder");
- props.put("key.serializer.class", "kafka.serializer.StringEncoder");
- props.put("partitioner.class", "org.apache.ignite.kafka.SimplePartitioner");
+ props.put("serializer.class", StringEncoder.class.getName());
+ props.put("key.serializer.class", StringEncoder.class.getName());
+ props.put("partitioner.class", SimplePartitioner.class.getName());
+
return new ProducerConfig(props);
}
/**
- * Creates Temp Directory
+ * Creates temp directory.
*
- * @param prefix prefix
- * @return Created File.
- * @throws IOException
+ * @param prefix Prefix.
+ * @return Created file.
+ * @throws IOException If failed.
*/
- private static File createTempDir(final String prefix)
- throws IOException {
- final Path path = Files.createTempDirectory(prefix);
- return path.toFile();
+ private static File createTempDir( String prefix) throws IOException {
+ Path path = Files.createTempDirectory(prefix);
+ return path.toFile();
}
/**
- * Creates Embedded ZooKeeper.
+ * Creates embedded ZooKeeper.
*/
private static class EmbeddedZooKeeper {
- /** Default ZooKeeper Host. */
+ /** Default ZooKeeper host. */
private final String zkHost;
- /** Default ZooKeeper Port. */
+ /** Default ZooKeeper port. */
private final int zkPort;
- /** NIO Context Factory. */
+ /** NIO context factory. */
private NIOServerCnxnFactory factory;
- /** Snapshot Directory. */
+ /** Snapshot directory. */
private File snapshotDir;
- /** Log Directory. */
+ /** Log directory. */
private File logDir;
/**
- * Creates an embedded Zookeeper
- * @param zkHost zookeeper host
- * @param zkPort zookeeper port
+ * Creates an embedded ZooKeeper.
+ *
+ * @param zkHost ZooKeeper host.
+ * @param zkPort ZooKeeper port.
*/
- EmbeddedZooKeeper(final String zkHost, final int zkPort) {
+ EmbeddedZooKeeper(String zkHost, int zkPort) {
this.zkHost = zkHost;
this.zkPort = zkPort;
}
@@ -333,22 +336,25 @@ public class KafkaEmbeddedBroker {
/**
* Starts up ZooKeeper.
*
- * @throws IOException
- * @throws InterruptedException
+ * @throws IOException If failed.
+ * @throws InterruptedException If interrupted.
*/
- void startup()
- throws IOException, InterruptedException {
+ void startup() throws IOException, InterruptedException {
snapshotDir = createTempDir("_ss");
+
logDir = createTempDir("_log");
- ZooKeeperServer zooServer = new ZooKeeperServer(snapshotDir, logDir, 500);
+
+ ZooKeeperServer zkSrv = new ZooKeeperServer(snapshotDir, logDir, 500);
+
factory = new NIOServerCnxnFactory();
+
factory.configure(new InetSocketAddress(zkHost, zkPort), 16);
- factory.startup(zooServer);
+
+ factory.startup(zkSrv);
}
/**
- *
- * @return actual port zookeeper is started
+ * @return Actual port ZooKeeper is started.
*/
int getActualPort() {
return factory.getLocalPort();
@@ -357,17 +363,16 @@ public class KafkaEmbeddedBroker {
/**
* Shuts down ZooKeeper.
*
- * @throws IOException
+ * @throws IOException If failed.
*/
- void shutdown()
- throws IOException {
+ void shutdown() throws IOException {
if (factory != null) {
factory.shutdown();
U.delete(snapshotDir);
+
U.delete(logDir);
}
}
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6ca6a0c7/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
index 5972639..2473990 100644
--- a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
@@ -32,31 +32,31 @@ import java.util.*;
import java.util.concurrent.*;
import static org.apache.ignite.events.EventType.*;
+import static org.apache.ignite.stream.kafka.KafkaEmbeddedBroker.*;
/**
* Tests {@link KafkaStreamer}.
*/
-public class KafkaIgniteStreamerSelfTest
- extends GridCommonAbstractTest {
+public class KafkaIgniteStreamerSelfTest extends GridCommonAbstractTest {
/** Embedded Kafka. */
private KafkaEmbeddedBroker embeddedBroker;
/** Count. */
private static final int CNT = 100;
- /** Test Topic. */
+ /** Test topic. */
private static final String TOPIC_NAME = "page_visits";
- /** Kafka Partition. */
+ /** Kafka partition. */
private static final int PARTITIONS = 4;
- /** Kafka Replication Factor. */
+ /** Kafka replication factor. */
private static final int REPLICATION_FACTOR = 1;
- /** Topic Message Key Prefix. */
+ /** Topic message key prefix. */
private static final String KEY_PREFIX = "192.168.2.";
- /** Topic Message Value Url. */
+ /** Topic message value URL. */
private static final String VALUE_URL = ",www.example.com,";
/** Constructor. */
@@ -65,18 +65,15 @@ public class KafkaIgniteStreamerSelfTest
}
/** {@inheritDoc} */
- @Override
- protected void beforeTest()
- throws Exception {
+ @SuppressWarnings("unchecked")
+ @Override protected void beforeTest() throws Exception {
grid().<Integer, String>getOrCreateCache(defaultCacheConfiguration());
embeddedBroker = new KafkaEmbeddedBroker();
}
/** {@inheritDoc} */
- @Override
- protected void afterTest()
- throws Exception {
+ @Override protected void afterTest() throws Exception {
grid().cache(null).clear();
embeddedBroker.shutdown();
@@ -85,76 +82,80 @@ public class KafkaIgniteStreamerSelfTest
/**
* Tests Kafka streamer.
*
- * @throws TimeoutException
- * @throws InterruptedException
+ * @throws TimeoutException If timed out.
+ * @throws InterruptedException If interrupted.
*/
- public void testKafkaStreamer()
- throws TimeoutException, InterruptedException {
+ public void testKafkaStreamer() throws TimeoutException, InterruptedException {
embeddedBroker.createTopic(TOPIC_NAME, PARTITIONS, REPLICATION_FACTOR);
- Map<String, String> keyValueMap = produceStream(TOPIC_NAME);
- consumerStream(TOPIC_NAME, keyValueMap);
+ Map<String, String> keyValMap = produceStream(TOPIC_NAME);
+
+ consumerStream(TOPIC_NAME, keyValMap);
}
/**
- * Produces/Sends messages to Kafka.
+ * Sends messages to Kafka.
*
* @param topic Topic name.
* @return Map of key value messages.
*/
- private Map<String, String> produceStream(final String topic) {
- final Map<String, String> keyValueMap = new HashMap<>();
-
+ private Map<String, String> produceStream(String topic) {
// Generate random subnets.
List<Integer> subnet = new ArrayList<>();
- int i = 0;
- while (i <= CNT)
- subnet.add(++i);
+ for (int i = 1; i <= CNT; i++)
+ subnet.add(i);
Collections.shuffle(subnet);
- final List<KeyedMessage<String, String>> messages = new ArrayList<>();
- for (int event = 0; event < CNT; event++) {
- long runtime = new Date().getTime();
- String ip = KEY_PREFIX + subnet.get(event);
+ List<KeyedMessage<String, String>> messages = new ArrayList<>(CNT);
+
+ Map<String, String> keyValMap = new HashMap<>();
+
+ for (int evt = 0; evt < CNT; evt++) {
+ long runtime = System.currentTimeMillis();
+
+ String ip = KEY_PREFIX + subnet.get(evt);
+
String msg = runtime + VALUE_URL + ip;
+
messages.add(new KeyedMessage<>(topic, ip, msg));
- keyValueMap.put(ip, msg);
+
+ keyValMap.put(ip, msg);
}
- final Producer<String, String> producer = embeddedBroker.sendMessages(messages);
+ Producer<String, String> producer = embeddedBroker.sendMessages(messages);
+
producer.close();
- return keyValueMap;
+ return keyValMap;
}
/**
- * Consumes Kafka Stream via ignite.
+ * Consumes Kafka stream via Ignite.
*
* @param topic Topic name.
- * @param keyValueMap Expected key value map.
- * @throws TimeoutException TimeoutException.
- * @throws InterruptedException InterruptedException.
+ * @param keyValMap Expected key value map.
+ * @throws TimeoutException If timed out.
+ * @throws InterruptedException If interrupted.
*/
- private void consumerStream(final String topic, final Map<String, String> keyValueMap)
+ private void consumerStream(String topic, Map<String, String> keyValMap)
throws TimeoutException, InterruptedException {
-
KafkaStreamer<String, String, String> kafkaStmr = null;
- final Ignite ignite = grid();
- try (IgniteDataStreamer<String, String> stmr = ignite.dataStreamer(null)) {
+ Ignite ignite = grid();
+ try (IgniteDataStreamer<String, String> stmr = ignite.dataStreamer(null)) {
stmr.allowOverwrite(true);
stmr.autoFlushFrequency(10);
- // Configure socket streamer.
+ // Configure Kafka streamer.
kafkaStmr = new KafkaStreamer<>();
// Get the cache.
IgniteCache<String, String> cache = ignite.cache(null);
- // Set ignite instance.
+ // Set Ignite instance.
kafkaStmr.setIgnite(ignite);
// Set data streamer instance.
@@ -167,58 +168,55 @@ public class KafkaIgniteStreamerSelfTest
kafkaStmr.setThreads(4);
// Set the consumer configuration.
- kafkaStmr.setConsumerConfig(createDefaultConsumerConfig(KafkaEmbeddedBroker.getZKAddress(),
- "groupX"));
+ kafkaStmr.setConsumerConfig(createDefaultConsumerConfig(getZKAddress(), "groupX"));
// Set the decoders.
- StringDecoder stringDecoder = new StringDecoder(new VerifiableProperties());
- kafkaStmr.setKeyDecoder(stringDecoder);
- kafkaStmr.setValueDecoder(stringDecoder);
+ StringDecoder strDecoder = new StringDecoder(new VerifiableProperties());
+
+ kafkaStmr.setKeyDecoder(strDecoder);
+ kafkaStmr.setValueDecoder(strDecoder);
// Start kafka streamer.
kafkaStmr.start();
final CountDownLatch latch = new CountDownLatch(CNT);
+
IgniteBiPredicate<UUID, CacheEvent> locLsnr = new IgniteBiPredicate<UUID, CacheEvent>() {
- @Override
- public boolean apply(UUID uuid, CacheEvent evt) {
+ @Override public boolean apply(UUID uuid, CacheEvent evt) {
latch.countDown();
+
return true;
}
};
ignite.events(ignite.cluster().forCacheNodes(null)).remoteListen(locLsnr, null, EVT_CACHE_OBJECT_PUT);
- latch.await();
- for (Map.Entry<String, String> entry : keyValueMap.entrySet()) {
- final String key = entry.getKey();
- final String value = entry.getValue();
+ latch.await();
- final String cacheValue = cache.get(key);
- assertEquals(value, cacheValue);
- }
+ for (Map.Entry<String, String> entry : keyValMap.entrySet())
+ assertEquals(entry.getValue(), cache.get(entry.getKey()));
}
-
finally {
- // Shutdown kafka streamer.
- kafkaStmr.stop();
+ if (kafkaStmr != null)
+ kafkaStmr.stop();
}
}
/**
* Creates default consumer config.
*
- * @param zooKeeper Zookeeper address <server:port>.
- * @param groupId Group Id for kafka subscriber.
- * @return {@link ConsumerConfig} kafka consumer configuration.
+ * @param zooKeeper ZooKeeper address <server:port>.
+ * @param grpId Group Id for kafka subscriber.
+ * @return Kafka consumer configuration.
*/
- private ConsumerConfig createDefaultConsumerConfig(String zooKeeper, String groupId) {
+ private ConsumerConfig createDefaultConsumerConfig(String zooKeeper, String grpId) {
A.notNull(zooKeeper, "zookeeper");
- A.notNull(groupId, "groupId");
+ A.notNull(grpId, "groupId");
Properties props = new Properties();
+
props.put("zookeeper.connect", zooKeeper);
- props.put("group.id", groupId);
+ props.put("group.id", grpId);
props.put("zookeeper.session.timeout.ms", "400");
props.put("zookeeper.sync.time.ms", "200");
props.put("auto.commit.interval.ms", "1000");
@@ -226,5 +224,4 @@ public class KafkaIgniteStreamerSelfTest
return new ConsumerConfig(props);
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6ca6a0c7/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
index b836b44..1ef4f77 100644
--- a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
@@ -18,29 +18,36 @@
package org.apache.ignite.stream.kafka;
import kafka.producer.*;
+import kafka.utils.*;
/**
- * Simple Partitioner for Kafka.
+ * Simple partitioner for Kafka.
*/
@SuppressWarnings("UnusedDeclaration")
-public class SimplePartitioner
- implements Partitioner {
+public class SimplePartitioner implements Partitioner {
+ /**
+ * Constructs instance.
+ *
+ * @param props Properties.
+ */
+ public SimplePartitioner(VerifiableProperties props) {
+ // No-op.
+ }
/**
* Partitions the key based on the key value.
*
* @param key Key.
- * @param partitionSize Partition size.
+ * @param partSize Partition size.
* @return partition Partition.
*/
- public int partition(Object key, int partitionSize) {
- int partition = 0;
+ public int partition(Object key, int partSize) {
String keyStr = (String)key;
+
String[] keyValues = keyStr.split("\\.");
+
Integer intKey = Integer.parseInt(keyValues[3]);
- if (intKey > 0) {
- partition = intKey % partitionSize;
- }
- return partition;
+
+ return intKey > 0 ? intKey % partSize : 0;
}
}
[17/20] incubator-ignite git commit: Merge branch 'ignite-1006' into
ignite-sprint-7
Posted by sb...@apache.org.
Merge branch 'ignite-1006' into ignite-sprint-7
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/2ee616de
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/2ee616de
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/2ee616de
Branch: refs/heads/ignite-428
Commit: 2ee616def5f532b2a049f0ba3a1582cd0bbcbaa0
Parents: 70238a6 8fd909d
Author: ashutak <as...@gridgain.com>
Authored: Fri Jun 26 18:02:08 2015 +0300
Committer: ashutak <as...@gridgain.com>
Committed: Fri Jun 26 18:02:08 2015 +0300
----------------------------------------------------------------------
.../org/apache/ignite/cluster/ClusterGroup.java | 7 +
.../internal/cluster/ClusterGroupAdapter.java | 10 ++
.../cluster/IgniteClusterAsyncImpl.java | 5 +
.../internal/ClusterForHostsSelfTest.java | 113 ---------------
.../internal/ClusterGroupHostsSelfTest.java | 141 +++++++++++++++++++
.../ignite/testsuites/IgniteBasicTestSuite.java | 2 +-
6 files changed, 164 insertions(+), 114 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2ee616de/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2ee616de/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2ee616de/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
----------------------------------------------------------------------
[07/20] incubator-ignite git commit: # ignite-883 issues with client
connect/reconnect
Posted by sb...@apache.org.
# ignite-883 issues with client connect/reconnect
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/efa92c54
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/efa92c54
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/efa92c54
Branch: refs/heads/ignite-428
Commit: efa92c54ab07d7f72b9c83aa3a09e03627d72e4a
Parents: 3e8ddb4
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jun 26 11:06:41 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jun 26 11:06:41 2015 +0300
----------------------------------------------------------------------
.../GridCachePartitionExchangeManager.java | 6 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 151 ++++++-----
.../ignite/spi/discovery/tcp/ServerImpl.java | 103 +++++--
.../spi/discovery/tcp/TcpDiscoverySpi.java | 3 +-
.../tcp/TcpClientDiscoverySpiSelfTest.java | 265 ++++++++++++++++++-
5 files changed, 448 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/efa92c54/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index edd0ad7..af87685 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -826,7 +826,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
updated |= top.update(null, entry.getValue()) != null;
}
- if (updated)
+ if (!cctx.kernalContext().clientNode() && updated)
refreshPartitions();
}
else
@@ -985,7 +985,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
// If not first preloading and no more topology events present,
// then we periodically refresh partition map.
- if (futQ.isEmpty() && preloadFinished) {
+ if (!cctx.kernalContext().clientNode() && futQ.isEmpty() && preloadFinished) {
refreshPartitions(timeout);
timeout = cctx.gridConfig().getNetworkTimeout();
@@ -1051,7 +1051,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
startEvtFired = true;
- if (changed && futQ.isEmpty())
+ if (!cctx.kernalContext().clientNode() && changed && futQ.isEmpty())
refreshPartitions();
}
else {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/efa92c54/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index 0c2c059..04276d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
@@ -1123,7 +1123,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (joinLatch.getCount() > 0) {
joinError(new IgniteSpiException("Join process timed out, did not receive response for " +
"join request (consider increasing 'joinTimeout' configuration property) " +
- "[joinTimeout=" + spi.joinTimeout + ", sock=" + sock +']'));
+ "[joinTimeout=" + spi.joinTimeout + ", sock=" + sock + ']'));
break;
}
@@ -1282,17 +1282,21 @@ class ClientImpl extends TcpDiscoveryImpl {
"[msg=" + msg + ", locNode=" + locNode + ']');
}
else {
- boolean topChanged = rmtNodes.putIfAbsent(newNodeId, node) == null;
+ if (nodeAdded()) {
+ boolean topChanged = rmtNodes.putIfAbsent(newNodeId, node) == null;
- if (topChanged) {
- if (log.isDebugEnabled())
- log.debug("Added new node to topology: " + node);
+ if (topChanged) {
+ if (log.isDebugEnabled())
+ log.debug("Added new node to topology: " + node);
- Map<Integer, byte[]> data = msg.newNodeDiscoveryData();
+ Map<Integer, byte[]> data = msg.newNodeDiscoveryData();
- if (data != null)
- spi.onExchange(newNodeId, newNodeId, data, null);
+ if (data != null)
+ spi.onExchange(newNodeId, newNodeId, data, null);
+ }
}
+ else if (log.isDebugEnabled())
+ log.debug("Ignore topology message, local node not added to topology: " + msg);
}
}
@@ -1332,54 +1336,58 @@ class ClientImpl extends TcpDiscoveryImpl {
"[msg=" + msg + ", locNode=" + locNode + ']');
}
else {
- TcpDiscoveryNode node = rmtNodes.get(msg.nodeId());
+ if (nodeAdded()) {
+ TcpDiscoveryNode node = rmtNodes.get(msg.nodeId());
- if (node == null) {
- if (log.isDebugEnabled())
- log.debug("Discarding node add finished message since node is not found [msg=" + msg + ']');
+ if (node == null) {
+ if (log.isDebugEnabled())
+ log.debug("Discarding node add finished message since node is not found [msg=" + msg + ']');
- return;
- }
+ return;
+ }
- boolean evt = false;
+ boolean evt = false;
- long topVer = msg.topologyVersion();
+ long topVer = msg.topologyVersion();
- assert topVer > 0 : msg;
+ assert topVer > 0 : msg;
- if (!node.visible()) {
- node.order(topVer);
- node.visible(true);
+ if (!node.visible()) {
+ node.order(topVer);
+ node.visible(true);
- if (spi.locNodeVer.equals(node.version()))
- node.version(spi.locNodeVer);
+ if (spi.locNodeVer.equals(node.version()))
+ node.version(spi.locNodeVer);
- evt = true;
- }
- else {
- if (log.isDebugEnabled())
- log.debug("Skip node join event, node already joined [msg=" + msg + ", node=" + node + ']');
+ evt = true;
+ }
+ else {
+ if (log.isDebugEnabled())
+ log.debug("Skip node join event, node already joined [msg=" + msg + ", node=" + node + ']');
- assert node.order() == topVer : node;
- }
+ assert node.order() == topVer : node;
+ }
- Collection<ClusterNode> top = updateTopologyHistory(topVer, msg);
+ Collection<ClusterNode> top = updateTopologyHistory(topVer, msg);
- assert top != null && top.contains(node) : "Topology does not contain node [msg=" + msg +
- ", node=" + node + ", top=" + top + ']';
+ assert top != null && top.contains(node) : "Topology does not contain node [msg=" + msg +
+ ", node=" + node + ", top=" + top + ']';
- if (!pending && joinLatch.getCount() > 0) {
- if (log.isDebugEnabled())
- log.debug("Discarding node add finished message (join process is not finished): " + msg);
+ if (!pending && joinLatch.getCount() > 0) {
+ if (log.isDebugEnabled())
+ log.debug("Discarding node add finished message (join process is not finished): " + msg);
- return;
- }
+ return;
+ }
- if (evt) {
- notifyDiscovery(EVT_NODE_JOINED, topVer, node, top);
+ if (evt) {
+ notifyDiscovery(EVT_NODE_JOINED, topVer, node, top);
- spi.stats.onNodeJoined();
+ spi.stats.onNodeJoined();
+ }
}
+ else if (log.isDebugEnabled())
+ log.debug("Ignore topology message, local node not added to topology: " + msg);
}
}
@@ -1397,31 +1405,42 @@ class ClientImpl extends TcpDiscoveryImpl {
if (spi.getSpiContext().isStopping())
return;
- TcpDiscoveryNode node = rmtNodes.remove(msg.creatorNodeId());
+ if (nodeAdded()) {
+ TcpDiscoveryNode node = rmtNodes.remove(msg.creatorNodeId());
- if (node == null) {
- if (log.isDebugEnabled())
- log.debug("Discarding node left message since node is not found [msg=" + msg + ']');
+ if (node == null) {
+ if (log.isDebugEnabled())
+ log.debug("Discarding node left message since node is not found [msg=" + msg + ']');
- return;
- }
+ return;
+ }
- Collection<ClusterNode> top = updateTopologyHistory(msg.topologyVersion(), msg);
+ Collection<ClusterNode> top = updateTopologyHistory(msg.topologyVersion(), msg);
- if (!pending && joinLatch.getCount() > 0) {
- if (log.isDebugEnabled())
- log.debug("Discarding node left message (join process is not finished): " + msg);
+ if (!pending && joinLatch.getCount() > 0) {
+ if (log.isDebugEnabled())
+ log.debug("Discarding node left message (join process is not finished): " + msg);
- return;
- }
+ return;
+ }
- notifyDiscovery(EVT_NODE_LEFT, msg.topologyVersion(), node, top);
+ notifyDiscovery(EVT_NODE_LEFT, msg.topologyVersion(), node, top);
- spi.stats.onNodeLeft();
+ spi.stats.onNodeLeft();
+ }
+ else if (log.isDebugEnabled())
+ log.debug("Ignore topology message, local node not added to topology: " + msg);
}
}
/**
+ * @return {@code True} if received node added message for local node.
+ */
+ private boolean nodeAdded() {
+ return !topHist.isEmpty();
+ }
+
+ /**
* @param msg Message.
*/
private void processNodeFailedMessage(TcpDiscoveryNodeFailedMessage msg) {
@@ -1514,9 +1533,9 @@ class ClientImpl extends TcpDiscoveryImpl {
return;
if (getLocalNodeId().equals(msg.creatorNodeId())) {
- assert msg.success() : msg;
-
if (reconnector != null) {
+ assert msg.success() : msg;
+
currSock = reconnector.sock;
sockWriter.setSocket(currSock);
@@ -1529,7 +1548,7 @@ class ClientImpl extends TcpDiscoveryImpl {
try {
for (TcpDiscoveryAbstractMessage pendingMsg : msg.pendingMessages()) {
if (log.isDebugEnabled())
- log.debug("Process message on reconnect [msg=" + pendingMsg + ']');
+ log.debug("Process pending message on reconnect [msg=" + pendingMsg + ']');
processDiscoveryMessage(pendingMsg);
}
@@ -1538,8 +1557,22 @@ class ClientImpl extends TcpDiscoveryImpl {
pending = false;
}
}
- else if (log.isDebugEnabled())
- log.debug("Discarding reconnect message, reconnect is completed: " + msg);
+ else {
+ if (joinLatch.getCount() > 0) {
+ if (msg.success()) {
+ for (TcpDiscoveryAbstractMessage pendingMsg : msg.pendingMessages()) {
+ if (log.isDebugEnabled())
+ log.debug("Process pending message on connect [msg=" + pendingMsg + ']');
+
+ processDiscoveryMessage(pendingMsg);
+ }
+
+ assert joinLatch.getCount() == 0 : msg;
+ }
+ }
+ else if (log.isDebugEnabled())
+ log.debug("Discarding reconnect message, reconnect is completed: " + msg);
+ }
}
else if (log.isDebugEnabled())
log.debug("Discarding reconnect message for another client: " + msg);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/efa92c54/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 2458f85..fa3e564 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -2452,7 +2452,40 @@ class ServerImpl extends TcpDiscoveryImpl {
return;
}
- if (log.isDebugEnabled())
+ if (msg.client()) {
+ TcpDiscoveryClientReconnectMessage reconMsg = new TcpDiscoveryClientReconnectMessage(node.id(),
+ node.clientRouterNodeId(),
+ null);
+
+ reconMsg.verify(getLocalNodeId());
+
+ Collection<TcpDiscoveryAbstractMessage> msgs = msgHist.messages(null, node);
+
+ if (msgs != null) {
+ reconMsg.pendingMessages(msgs);
+
+ reconMsg.success(true);
+ }
+
+ if (log.isDebugEnabled())
+ log.debug("Send reconnect message to already joined client " +
+ "[clientNode=" + existingNode + ", msg=" + reconMsg + ']');
+
+ if (getLocalNodeId().equals(node.clientRouterNodeId())) {
+ ClientMessageWorker wrk = clientMsgWorkers.get(node.id());
+
+ if (wrk != null)
+ wrk.addMessage(reconMsg);
+ else if (log.isDebugEnabled())
+ log.debug("Failed to find client message worker " +
+ "[clientNode=" + existingNode + ", msg=" + reconMsg + ']');
+ }
+ else {
+ if (ring.hasRemoteNodes())
+ sendMessageAcrossRing(reconMsg);
+ }
+ }
+ else if (log.isDebugEnabled())
log.debug("Ignoring join request message since node is already in topology: " + msg);
return;
@@ -4104,15 +4137,44 @@ class ServerImpl extends TcpDiscoveryImpl {
}
if (req.client()) {
+ ClientMessageWorker clientMsgWrk0 = new ClientMessageWorker(sock, nodeId);
+
+ while (true) {
+ ClientMessageWorker old = clientMsgWorkers.putIfAbsent(nodeId, clientMsgWrk0);
+
+ if (old == null)
+ break;
+
+ if (old.isInterrupted()) {
+ clientMsgWorkers.remove(nodeId, old);
+
+ continue;
+ }
+
+ old.join(500);
+
+ old = clientMsgWorkers.putIfAbsent(nodeId, clientMsgWrk0);
+
+ if (old == null)
+ break;
+
+ if (log.isDebugEnabled())
+ log.debug("Already have client message worker, closing connection " +
+ "[locNodeId=" + locNodeId +
+ ", rmtNodeId=" + nodeId +
+ ", workerSock=" + old.sock +
+ ", sock=" + sock + ']');
+
+ return;
+ }
+
if (log.isDebugEnabled())
log.debug("Created client message worker [locNodeId=" + locNodeId +
", rmtNodeId=" + nodeId + ", sock=" + sock + ']');
- clientMsgWrk = new ClientMessageWorker(sock, nodeId);
-
- clientMsgWrk.start();
+ assert clientMsgWrk0 == clientMsgWorkers.get(nodeId);
- clientMsgWorkers.put(nodeId, clientMsgWrk);
+ clientMsgWrk = clientMsgWrk0;
}
if (log.isDebugEnabled())
@@ -4188,7 +4250,7 @@ class ServerImpl extends TcpDiscoveryImpl {
TcpDiscoveryJoinRequestMessage req = (TcpDiscoveryJoinRequestMessage)msg;
if (!req.responded()) {
- boolean ok = processJoinRequestMessage(req);
+ boolean ok = processJoinRequestMessage(req, clientMsgWrk);
if (clientMsgWrk != null && ok)
continue;
@@ -4202,14 +4264,17 @@ class ServerImpl extends TcpDiscoveryImpl {
TcpDiscoverySpiState state = spiStateCopy();
if (state == CONNECTED) {
- spi.writeToSocket(sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK);
+
+ if (clientMsgWrk != null && clientMsgWrk.getState() == State.NEW)
+ clientMsgWrk.start();
msgWorker.addMessage(msg);
continue;
}
else {
- spi.writeToSocket(sock, RES_CONTINUE_JOIN);
+ spi.writeToSocket(msg, sock, RES_CONTINUE_JOIN);
break;
}
@@ -4217,7 +4282,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else if (msg instanceof TcpDiscoveryDuplicateIdMessage) {
// Send receipt back.
- spi.writeToSocket(sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK);
boolean ignored = false;
@@ -4246,7 +4311,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else if (msg instanceof TcpDiscoveryAuthFailedMessage) {
// Send receipt back.
- spi.writeToSocket(sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK);
boolean ignored = false;
@@ -4275,7 +4340,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else if (msg instanceof TcpDiscoveryCheckFailedMessage) {
// Send receipt back.
- spi.writeToSocket(sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK);
boolean ignored = false;
@@ -4304,7 +4369,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else if (msg instanceof TcpDiscoveryLoopbackProblemMessage) {
// Send receipt back.
- spi.writeToSocket(sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK);
boolean ignored = false;
@@ -4346,7 +4411,7 @@ class ServerImpl extends TcpDiscoveryImpl {
// Send receipt back.
if (clientMsgWrk == null)
- spi.writeToSocket(sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK);
}
catch (IgniteCheckedException e) {
if (log.isDebugEnabled())
@@ -4435,24 +4500,29 @@ class ServerImpl extends TcpDiscoveryImpl {
/**
* @param msg Join request message.
+ * @param clientMsgWrk Client message worker to start.
* @return Whether connection was successful.
* @throws IOException If IO failed.
*/
@SuppressWarnings({"IfMayBeConditional"})
- private boolean processJoinRequestMessage(TcpDiscoveryJoinRequestMessage msg) throws IOException {
+ private boolean processJoinRequestMessage(TcpDiscoveryJoinRequestMessage msg,
+ @Nullable ClientMessageWorker clientMsgWrk) throws IOException {
assert msg != null;
assert !msg.responded();
TcpDiscoverySpiState state = spiStateCopy();
if (state == CONNECTED) {
- spi.writeToSocket(sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK);
if (log.isDebugEnabled())
log.debug("Responded to join request message [msg=" + msg + ", res=" + RES_OK + ']');
msg.responded(true);
+ if (clientMsgWrk != null && clientMsgWrk.getState() == State.NEW)
+ clientMsgWrk.start();
+
msgWorker.addMessage(msg);
return true;
@@ -4477,7 +4547,7 @@ class ServerImpl extends TcpDiscoveryImpl {
// Local node is stopping. Remote node should try next one.
res = RES_CONTINUE_JOIN;
- spi.writeToSocket(sock, res);
+ spi.writeToSocket(msg, sock, res);
if (log.isDebugEnabled())
log.debug("Responded to join request message [msg=" + msg + ", res=" + res + ']');
@@ -4632,6 +4702,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/**
+ * @return Ping result.
* @throws InterruptedException If interrupted.
*/
public boolean ping() throws InterruptedException {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/efa92c54/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index 1d1916a..7663fe6 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -1227,12 +1227,13 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements DiscoverySpi, T
/**
* Writes response to the socket.
*
+ * @param msg Received message.
* @param sock Socket.
* @param res Integer response.
* @throws IOException If IO failed or write timed out.
*/
@SuppressWarnings("ThrowFromFinallyBlock")
- protected void writeToSocket(Socket sock, int res) throws IOException {
+ protected void writeToSocket(TcpDiscoveryAbstractMessage msg, Socket sock, int res) throws IOException {
assert sock != null;
SocketTimeoutObject obj = new SocketTimeoutObject(sock, U.currentTimeMillis() + sockTimeout);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/efa92c54/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
index 8147958..ec6a526 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
@@ -24,7 +24,9 @@ import org.apache.ignite.events.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.util.*;
import org.apache.ignite.internal.util.io.*;
+import org.apache.ignite.internal.util.lang.*;
import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.resources.*;
import org.apache.ignite.spi.*;
@@ -106,11 +108,14 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
/** */
private int maxMissedClientHbs = TcpDiscoverySpi.DFLT_MAX_MISSED_CLIENT_HEARTBEATS;
+ /** */
+ private IgniteInClosure2X<TcpDiscoveryAbstractMessage, Socket> afterWrite;
+
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
- TcpDiscoverySpi disco = new TestTcpDiscoverySpi();
+ TestTcpDiscoverySpi disco = new TestTcpDiscoverySpi();
disco.setMaxMissedClientHeartbeats(maxMissedClientHbs);
@@ -154,6 +159,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
disco.setJoinTimeout(joinTimeout);
disco.setNetworkTimeout(netTimeout);
+ disco.afterWrite(afterWrite);
+
cfg.setDiscoverySpi(disco);
if (nodeId != null)
@@ -1016,6 +1023,189 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
+ * @throws Exception If failed.
+ */
+ public void testJoinError3() throws Exception {
+ startServerNodes(1);
+
+ Ignite ignite = G.ignite("server-0");
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)ignite.configuration().getDiscoverySpi());
+
+ srvSpi.failNodeAddFinishedMessage();
+
+ startClientNodes(1);
+
+ checkNodes(1, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testJoinErrorMissedAddFinishedMessage1() throws Exception {
+ missedAddFinishedMessage(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testJoinErrorMissedAddFinishedMessage2() throws Exception {
+ missedAddFinishedMessage(false);
+ }
+
+ /**
+ * @param singleSrv If {@code true} starts one server node two otherwise.
+ * @throws Exception If failed.
+ */
+ private void missedAddFinishedMessage(boolean singleSrv) throws Exception {
+ int srvs = singleSrv ? 1 : 2;
+
+ startServerNodes(srvs);
+
+ afterWrite = new CIX2<TcpDiscoveryAbstractMessage, Socket>() {
+ private boolean first = true;
+
+ @Override public void applyx(TcpDiscoveryAbstractMessage msg, Socket sock) throws IgniteCheckedException {
+ if (first && (msg instanceof TcpDiscoveryJoinRequestMessage)) {
+ first = false;
+
+ log.info("Close socket after message write [msg=" + msg + "]");
+
+ try {
+ sock.close();
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException(e);
+ }
+
+ log.info("Delay after message write [msg=" + msg + "]");
+
+ U.sleep(5000); // Wait when server process join request.
+ }
+ }
+ };
+
+ Ignite srv = singleSrv ? G.ignite("server-0") : G.ignite("server-1");
+
+ TcpDiscoveryNode srvNode = (TcpDiscoveryNode)srv.cluster().localNode();
+
+ assertEquals(singleSrv ? 1 : 2, srvNode.order());
+
+ clientIpFinder = new TcpDiscoveryVmIpFinder();
+
+ clientIpFinder.setAddresses(Collections.singleton("localhost:" + srvNode.discoveryPort()));
+
+ startClientNodes(1);
+
+ TcpDiscoveryNode clientNode = (TcpDiscoveryNode)G.ignite("client-0").cluster().localNode();
+
+ assertEquals(srvNode.id(), clientNode.clientRouterNodeId());
+
+ checkNodes(srvs, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientMessageWorkerStartSingleServer() throws Exception {
+ clientMessageWorkerStart(1, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientMessageWorkerStartTwoServers1() throws Exception {
+ clientMessageWorkerStart(2, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientMessageWorkerStartTwoServers2() throws Exception {
+ clientMessageWorkerStart(2, 2);
+ }
+
+ /**
+ * @param srvs Number of server nodes.
+ * @param connectTo What server connect to.
+ * @throws Exception If failed.
+ */
+ private void clientMessageWorkerStart(int srvs, int connectTo) throws Exception {
+ startServerNodes(srvs);
+
+ Ignite srv = G.ignite("server-" + (connectTo - 1));
+
+ final TcpDiscoveryNode srvNode = (TcpDiscoveryNode)srv.cluster().localNode();
+
+ assertEquals((long)connectTo, srvNode.order());
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)srv.configuration().getDiscoverySpi());
+
+ final String client0 = "client-" + clientIdx.getAndIncrement();
+
+ srvSpi.delayJoinAckFor = client0;
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ clientIpFinder = new TcpDiscoveryVmIpFinder();
+
+ clientIpFinder.setAddresses(Collections.singleton("localhost:" + srvNode.discoveryPort()));
+
+ Ignite client = startGrid(client0);
+
+ clientIpFinder = null;
+
+ clientNodeIds.add(client.cluster().localNode().id());
+
+ TestTcpDiscoverySpi clientSpi = ((TestTcpDiscoverySpi)client.configuration().getDiscoverySpi());
+
+ assertFalse(clientSpi.invalidResponse());
+
+ TcpDiscoveryNode clientNode = (TcpDiscoveryNode)client.cluster().localNode();
+
+ assertEquals(srvNode.id(), clientNode.clientRouterNodeId());
+
+ return null;
+ }
+ });
+
+ final String client1 = "client-" + clientIdx.getAndIncrement();
+
+ while (!fut.isDone()) {
+ startGrid(client1);
+
+ stopGrid(client1);
+ }
+
+ fut.get();
+
+ checkNodes(srvs, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testJoinMutlithreaded() throws Exception {
+ startServerNodes(1);
+
+ final int CLIENTS = 30;
+
+ clientsPerSrv = CLIENTS;
+
+ GridTestUtils.runMultiThreaded(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ Ignite g = startGrid("client-" + clientIdx.getAndIncrement());
+
+ clientNodeIds.add(g.cluster().localNode().id());
+
+ return null;
+ }
+ }, CLIENTS, "start-client");
+
+ checkNodes(1, CLIENTS);
+ }
+
+ /**
* @param clientIdx Client index.
* @param srvIdx Server index.
* @throws Exception In case of error.
@@ -1267,8 +1457,20 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
private AtomicInteger failNodeAdded = new AtomicInteger();
/** */
+ private AtomicInteger failNodeAddFinished = new AtomicInteger();
+
+ /** */
private AtomicInteger failClientReconnect = new AtomicInteger();
+ /** */
+ private IgniteInClosure2X<TcpDiscoveryAbstractMessage, Socket> afterWrite;
+
+ /** */
+ private volatile boolean invalidRes;
+
+ /** */
+ private volatile String delayJoinAckFor;
+
/**
* @param lock Lock.
*/
@@ -1287,6 +1489,20 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
+ * @param afterWrite After write callback.
+ */
+ void afterWrite(IgniteInClosure2X<TcpDiscoveryAbstractMessage, Socket> afterWrite) {
+ this.afterWrite = afterWrite;
+ }
+
+ /**
+ * @return {@code True} if received unexpected ack.
+ */
+ boolean invalidResponse() {
+ return invalidRes;
+ }
+
+ /**
*
*/
void failNodeAddedMessage() {
@@ -1296,6 +1512,13 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
/**
*
*/
+ void failNodeAddFinishedMessage() {
+ failNodeAddFinished.set(1);
+ }
+
+ /**
+ *
+ */
void failClientReconnectMessage() {
failClientReconnect.set(1);
}
@@ -1322,6 +1545,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
if (msg instanceof TcpDiscoveryNodeAddedMessage)
fail = failNodeAdded.getAndDecrement() > 0;
+ else if (msg instanceof TcpDiscoveryNodeAddFinishedMessage)
+ fail = failNodeAddFinished.getAndDecrement() > 0;
else if (msg instanceof TcpDiscoveryClientReconnectMessage)
fail = failClientReconnect.getAndDecrement() > 0;
@@ -1332,6 +1557,9 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
super.writeToSocket(sock, msg, bout);
+
+ if (afterWrite != null)
+ afterWrite.apply(msg, sock);
}
/** {@inheritDoc} */
@@ -1365,5 +1593,40 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
impl.workerThread().resume();
}
+
+ /** {@inheritDoc} */
+ @Override protected void writeToSocket(TcpDiscoveryAbstractMessage msg, Socket sock, int res) throws IOException {
+ if (delayJoinAckFor != null && msg instanceof TcpDiscoveryJoinRequestMessage) {
+ TcpDiscoveryJoinRequestMessage msg0 = (TcpDiscoveryJoinRequestMessage)msg;
+
+ if (delayJoinAckFor.equals(msg0.node().attribute(IgniteNodeAttributes.ATTR_GRID_NAME))) {
+ log.info("Delay response [sock=" + sock + ", msg=" + msg0 + ", res=" + res + ']');
+
+ delayJoinAckFor = null;
+
+ try {
+ Thread.sleep(2000);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ super.writeToSocket(msg, sock, res);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int readReceipt(Socket sock, long timeout) throws IOException {
+ int res = super.readReceipt(sock, timeout);
+
+ if (res != TcpDiscoveryImpl.RES_OK) {
+ invalidRes = true;
+
+ log.info("Received unexpected response: " + res);
+ }
+
+ return res;
+ }
}
}
[03/20] incubator-ignite git commit: # GG-10404 Suppress Visor
exceptions.
Posted by sb...@apache.org.
# GG-10404 Suppress Visor exceptions.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/c62c410b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/c62c410b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/c62c410b
Branch: refs/heads/ignite-428
Commit: c62c410bcf1a1f353fa586e515f846746cb4a482
Parents: 6779071
Author: AKuznetsov <ak...@gridgain.com>
Authored: Fri Jun 12 17:39:09 2015 +0700
Committer: AKuznetsov <ak...@gridgain.com>
Committed: Fri Jun 12 17:39:09 2015 +0700
----------------------------------------------------------------------
.../connection/GridClientNioTcpConnection.java | 7 +----
.../processors/rest/GridRestProcessor.java | 4 ++-
.../handlers/task/GridTaskCommandHandler.java | 12 ++++---
.../processors/task/GridTaskWorker.java | 4 ++-
.../visor/query/VisorQueryCleanupTask.java | 4 +--
.../util/VisorClusterGroupEmptyException.java | 33 ++++++++++++++++++++
.../visor/util/VisorEmptyTopologyException.java | 33 --------------------
7 files changed, 50 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c62c410b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
index 67709b8..d247e05 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/impl/connection/GridClientNioTcpConnection.java
@@ -750,12 +750,7 @@ public class GridClientNioTcpConnection extends GridClientConnection {
new GridClientFutureCallback<GridClientTaskResultBean, R>() {
@Override public R onComplete(GridClientFuture<GridClientTaskResultBean> fut)
throws GridClientException {
- GridClientTaskResultBean resBean = fut.get();
-
- if (resBean != null)
- return resBean.getResult();
- else
- return null;
+ return fut.get().getResult();
}
});
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c62c410b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
index 52ca610..2d1d802 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/GridRestProcessor.java
@@ -36,6 +36,7 @@ import org.apache.ignite.internal.util.future.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.internal.util.worker.*;
+import org.apache.ignite.internal.visor.util.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.plugin.security.*;
import org.apache.ignite.plugin.security.SecurityException;
@@ -214,7 +215,8 @@ public class GridRestProcessor extends GridProcessorAdapter {
res = f.get();
}
catch (Exception e) {
- LT.error(log, e, "Failed to handle request: " + req.command());
+ if (!X.hasCause(e, VisorClusterGroupEmptyException.class))
+ LT.error(log, e, "Failed to handle request: " + req.command());
if (log.isDebugEnabled())
log.debug("Failed to handle request [req=" + req + ", e=" + e + "]");
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c62c410b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java
index a647cd1..d832b21 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java
@@ -33,6 +33,7 @@ import org.apache.ignite.internal.util.*;
import org.apache.ignite.internal.util.future.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.internal.visor.util.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.resources.*;
import org.jetbrains.annotations.*;
@@ -134,7 +135,8 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter {
return handleAsyncUnsafe(req);
}
catch (IgniteCheckedException e) {
- U.error(log, "Failed to execute task command: " + req, e);
+ if (!X.hasCause(e, VisorClusterGroupEmptyException.class))
+ U.error(log, "Failed to execute task command: " + req, e);
return new GridFinishedFuture<>(e);
}
@@ -237,9 +239,11 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter {
U.warn(log, "Failed to execute task due to topology issues (are all mapped " +
"nodes alive?) [name=" + name + ", clientId=" + req.clientId() +
", err=" + e + ']');
- else
- U.error(log, "Failed to execute task [name=" + name + ", clientId=" +
- req.clientId() + ']', e);
+ else {
+ if (!X.hasCause(e, VisorClusterGroupEmptyException.class))
+ U.error(log, "Failed to execute task [name=" + name + ", clientId=" +
+ req.clientId() + ']', e);
+ }
desc = new TaskDescriptor(true, null, e);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c62c410b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
index f6d686c..eb5fa77 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
@@ -30,6 +30,7 @@ import org.apache.ignite.internal.processors.timeout.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.internal.util.worker.*;
+import org.apache.ignite.internal.visor.util.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.marshaller.*;
import org.apache.ignite.resources.*;
@@ -443,7 +444,8 @@ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject {
}
catch (IgniteException | IgniteCheckedException e) {
if (!fut.isCancelled()) {
- U.error(log, "Failed to map task jobs to nodes: " + ses, e);
+ if (!(e instanceof VisorClusterGroupEmptyException))
+ U.error(log, "Failed to map task jobs to nodes: " + ses, e);
finishTask(null, e);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c62c410b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
index b9a55e1..5ceb300 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCleanupTask.java
@@ -49,7 +49,7 @@ public class VisorQueryCleanupTask extends VisorMultiNodeTask<Map<UUID, Collecti
Set<UUID> nodeIds = taskArg.keySet();
if (nodeIds.isEmpty())
- throw new VisorEmptyTopologyException("Nothing to clear. List with node IDs is empty!");
+ throw new VisorClusterGroupEmptyException("Nothing to clear. List with node IDs is empty!");
Map<ComputeJob, ClusterNode> map = U.newHashMap(nodeIds.size());
@@ -64,7 +64,7 @@ public class VisorQueryCleanupTask extends VisorMultiNodeTask<Map<UUID, Collecti
for (UUID nid : nodeIds)
notFoundNodes = notFoundNodes + (notFoundNodes.isEmpty() ? "" : ",") + U.id8(nid);
- throw new VisorEmptyTopologyException("Failed to clear query results. Nodes are not available: [" +
+ throw new VisorClusterGroupEmptyException("Failed to clear query results. Nodes are not available: [" +
notFoundNodes + "]");
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c62c410b/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorClusterGroupEmptyException.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorClusterGroupEmptyException.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorClusterGroupEmptyException.java
new file mode 100644
index 0000000..b969178
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorClusterGroupEmptyException.java
@@ -0,0 +1,33 @@
+/*
+ * 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.visor.util;
+
+import org.apache.ignite.cluster.*;
+
+/**
+ * Exception to throw from Visor tasks in case of empty topology.
+ */
+public class VisorClusterGroupEmptyException extends ClusterGroupEmptyException {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** @inheritDoc */
+ public VisorClusterGroupEmptyException(String msg) {
+ super(msg);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c62c410b/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorEmptyTopologyException.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorEmptyTopologyException.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorEmptyTopologyException.java
deleted file mode 100644
index fda1bd7..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/util/VisorEmptyTopologyException.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.visor.util;
-
-import org.apache.ignite.*;
-
-/**
- * Marker exception for indication of empty topology in Visor tasks.
- */
-public class VisorEmptyTopologyException extends IgniteException {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** @inheritDoc */
- public VisorEmptyTopologyException(String msg) {
- super(msg);
- }
-}
[18/20] incubator-ignite git commit: ignite-428 Implement
IgniteKafkaStreamer to stream data from Apache Kafka
Posted by sb...@apache.org.
ignite-428 Implement IgniteKafkaStreamer to stream data from Apache Kafka
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/da07744c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/da07744c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/da07744c
Branch: refs/heads/ignite-428
Commit: da07744c4411a6fb0a8cb4ccd3bbfd365887a8f2
Parents: 2ee616d
Author: vishal.garg <vi...@workday.com>
Authored: Mon Jun 22 19:35:08 2015 -0700
Committer: agura <ag...@gridgain.com>
Committed: Fri Jun 26 19:04:40 2015 +0300
----------------------------------------------------------------------
modules/kafka/pom.xml | 128 +++++++
.../ignite/stream/kafka/KafkaStreamer.java | 179 +++++++++
.../stream/kafka/KafkaEmbeddedBroker.java | 373 +++++++++++++++++++
.../kafka/KafkaIgniteStreamerSelfTest.java | 230 ++++++++++++
.../ignite/stream/kafka/SimplePartitioner.java | 46 +++
pom.xml | 1 +
6 files changed, 957 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da07744c/modules/kafka/pom.xml
----------------------------------------------------------------------
diff --git a/modules/kafka/pom.xml b/modules/kafka/pom.xml
new file mode 100644
index 0000000..165ec1c
--- /dev/null
+++ b/modules/kafka/pom.xml
@@ -0,0 +1,128 @@
+<?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-kafka</artifactId>
+ <version>1.1.1-SNAPSHOT</version>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_2.10</artifactId>
+ <version>0.8.2.1</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.sun.jmx</groupId>
+ <artifactId>jmxri</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.sun.jdmk</groupId>
+ <artifactId>jmxtools</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>net.sf.jopt-simple</groupId>
+ <artifactId>jopt-simple</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.zookeeper</groupId>
+ <artifactId>zookeeper</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.zookeeper</groupId>
+ <artifactId>zookeeper</artifactId>
+ <version>3.4.5</version>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>2.4</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-log4j</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.ow2.asm</groupId>
+ <artifactId>asm-all</artifactId>
+ <version>4.2</version>
+ </dependency>
+
+
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.gridgain</groupId>
+ <artifactId>ignite-shmem</artifactId>
+ <version>1.0.0</version>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-beanutils</groupId>
+ <artifactId>commons-beanutils</artifactId>
+ <version>1.8.3</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-spring</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da07744c/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java b/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
new file mode 100644
index 0000000..e0240ce
--- /dev/null
+++ b/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
@@ -0,0 +1,179 @@
+/*
+ * 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.kafka;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.stream.*;
+
+import kafka.consumer.*;
+import kafka.javaapi.consumer.ConsumerConnector;
+import kafka.message.*;
+import kafka.serializer.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+/**
+ * Server that subscribes to topic messages from Kafka broker, streams its to key-value pairs into {@link
+ * org.apache.ignite.IgniteDataStreamer} instance.
+ * <p>
+ * Uses Kafka's High Level Consumer API to read messages from Kafka
+ *
+ * @see <a href="https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example">Consumer Consumer Group
+ * Example</a>
+ */
+public class KafkaStreamer<T, K, V>
+ extends StreamAdapter<T, K, V> {
+
+ /** Logger. */
+ private IgniteLogger log;
+
+ /** Executor used to submit kafka streams. */
+ private ExecutorService executor;
+
+ /** Topic. */
+ private String topic;
+
+ /** Number of threads to process kafka streams. */
+ private int threads;
+
+ /** Kafka Consumer Config. */
+ private ConsumerConfig consumerConfig;
+
+ /** Key Decoder. */
+ private Decoder<K> keyDecoder;
+
+ /** Value Decoder. */
+ private Decoder<V> valueDecoder;
+
+ /** Kafka Consumer connector. */
+ private ConsumerConnector consumer;
+
+ /**
+ * Sets the topic.
+ *
+ * @param topic Topic Name.
+ */
+ public void setTopic(final String topic) {
+ this.topic = topic;
+ }
+
+ /**
+ * Sets the threads.
+ *
+ * @param threads Number of Threads.
+ */
+ public void setThreads(final int threads) {
+ this.threads = threads;
+ }
+
+ /**
+ * Sets the consumer config.
+ *
+ * @param consumerConfig Consumer configuration.
+ */
+ public void setConsumerConfig(final ConsumerConfig consumerConfig) {
+ this.consumerConfig = consumerConfig;
+ }
+
+ /**
+ * Sets the key decoder.
+ *
+ * @param keyDecoder Key Decoder.
+ */
+ public void setKeyDecoder(final Decoder<K> keyDecoder) {
+ this.keyDecoder = keyDecoder;
+ }
+
+ /**
+ * Sets the value decoder.
+ *
+ * @param valueDecoder Value Decoder
+ */
+ public void setValueDecoder(final Decoder<V> valueDecoder) {
+ this.valueDecoder = valueDecoder;
+ }
+
+ /**
+ * Starts streamer.
+ *
+ * @throws IgniteException If failed.
+ */
+ public void start() {
+ A.notNull(getStreamer(), "streamer");
+ A.notNull(getIgnite(), "ignite");
+ A.notNull(topic, "topic");
+ A.notNull(keyDecoder, "key decoder");
+ A.notNull(valueDecoder, "value decoder");
+ A.notNull(consumerConfig, "kafka consumer config");
+ A.ensure(threads > 0, "threads > 0");
+
+ log = getIgnite().log();
+
+ consumer = kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);
+
+ Map<String, Integer> topicCountMap = new HashMap<>();
+ topicCountMap.put(topic, threads);
+
+ Map<String, List<KafkaStream<K, V>>> consumerMap = consumer.createMessageStreams(topicCountMap, keyDecoder,
+ valueDecoder);
+
+ List<KafkaStream<K, V>> streams = consumerMap.get(topic);
+
+ // Now launch all the consumer threads.
+ executor = Executors.newFixedThreadPool(threads);
+
+ // Now create an object to consume the messages.
+ for (final KafkaStream<K,V> stream : streams) {
+ executor.submit(new Runnable() {
+ @Override public void run() {
+
+ ConsumerIterator<K, V> it = stream.iterator();
+
+ while (it.hasNext()) {
+ final MessageAndMetadata<K, V> messageAndMetadata = it.next();
+ getStreamer().addData(messageAndMetadata.key(), messageAndMetadata.message());
+ }
+ }
+ });
+ }
+ }
+
+ /**
+ * Stops streamer.
+ */
+ public void stop() {
+ if (consumer != null)
+ consumer.shutdown();
+
+ if (executor != null) {
+ executor.shutdown();
+
+ try {
+ if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS))
+ if (log.isDebugEnabled())
+ log.debug("Timed out waiting for consumer threads to shut down, exiting uncleanly");
+ }
+ catch (InterruptedException e) {
+ if (log.isDebugEnabled())
+ log.debug("Interrupted during shutdown, exiting uncleanly");
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da07744c/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
new file mode 100644
index 0000000..28533f7
--- /dev/null
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaEmbeddedBroker.java
@@ -0,0 +1,373 @@
+/*
+ * 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.kafka;
+
+import org.apache.commons.io.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.zookeeper.server.*;
+
+import kafka.admin.*;
+import kafka.api.*;
+import kafka.api.Request;
+import kafka.producer.*;
+import kafka.server.*;
+import kafka.utils.*;
+import org.I0Itec.zkclient.*;
+
+import java.io.*;
+import java.net.*;
+import java.nio.file.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+/**
+ * Kafka Embedded Broker.
+ */
+public class KafkaEmbeddedBroker {
+
+ /** Default ZooKeeper Host. */
+ private static final String ZK_HOST = "localhost";
+
+ /** Broker Port. */
+ private static final int BROKER_PORT = 9092;
+
+ /** ZooKeeper Connection Timeout. */
+ private static final int ZK_CONNECTION_TIMEOUT = 6000;
+
+ /** ZooKeeper Session Timeout. */
+ private static final int ZK_SESSION_TIMEOUT = 6000;
+
+ /** ZooKeeper port. */
+ private static int zkPort = 0;
+
+ /** Is ZooKeeper Ready. */
+ private boolean zkReady;
+
+ /** Kafka Config. */
+ private KafkaConfig brokerConfig;
+
+ /** Kafka Server. */
+ private KafkaServer kafkaServer;
+
+ /** ZooKeeper Client. */
+ private ZkClient zkClient;
+
+ /** Embedded ZooKeeper. */
+ private EmbeddedZooKeeper zooKeeper;
+
+ /**
+ * Creates an embedded Kafka Broker.
+ */
+ public KafkaEmbeddedBroker() {
+ try {
+ setupEmbeddedZooKeeper();
+ setupEmbeddedKafkaServer();
+ }
+ catch (IOException | InterruptedException e) {
+ throw new RuntimeException("failed to start Kafka Broker " + e);
+ }
+
+ }
+
+ /**
+ * @return ZooKeeper Address.
+ */
+ public static String getZKAddress() {
+ return ZK_HOST + ":" + zkPort;
+ }
+
+ /**
+ * Creates a Topic.
+ *
+ * @param topic topic name
+ * @param partitions number of paritions for the topic
+ * @param replicationFactor replication factor
+ * @throws TimeoutException
+ * @throws InterruptedException
+ */
+ public void createTopic(String topic, final int partitions, final int replicationFactor)
+ throws TimeoutException, InterruptedException {
+ AdminUtils.createTopic(zkClient, topic, partitions, replicationFactor, new Properties());
+ waitUntilMetadataIsPropagated(topic, 0, 10000, 100);
+ }
+
+ /**
+ * Sends message to Kafka Broker.
+ *
+ * @param keyedMessages List of Keyed Messages.
+ * @return Producer used to send the message.
+ */
+ public Producer sendMessages(List<KeyedMessage<String, String>> keyedMessages) {
+ Producer<String, String> producer = new Producer<>(getProducerConfig());
+ producer.send(scala.collection.JavaConversions.asScalaBuffer(keyedMessages));
+ return producer;
+ }
+
+ /**
+ * Shuts down Kafka Broker.
+ *
+ * @throws IOException
+ */
+ public void shutdown()
+ throws IOException {
+
+ zkReady = false;
+
+ if (kafkaServer != null)
+ kafkaServer.shutdown();
+
+ List<String> logDirs = scala.collection.JavaConversions.asJavaList(brokerConfig.logDirs());
+
+ for (String logDir : logDirs) {
+ FileUtils.deleteDirectory(new File(logDir));
+ }
+
+ if (zkClient != null) {
+ zkClient.close();
+ zkClient = null;
+ }
+
+ if (zooKeeper != null) {
+
+ try {
+ zooKeeper.shutdown();
+ }
+ catch (IOException e) {
+ // ignore
+ }
+
+ zooKeeper = null;
+ }
+
+ }
+
+ /**
+ * @return the Zookeeper Client
+ */
+ private ZkClient getZkClient() {
+ A.ensure(zkReady, "Zookeeper not setup yet");
+ A.notNull(zkClient, "Zookeeper client is not yet initialized");
+
+ return zkClient;
+ }
+
+ /**
+ * Checks if topic metadata is propagated.
+ *
+ * @param topic topic name
+ * @param partition partition
+ * @return true if propagated otherwise false
+ */
+ private boolean isMetadataPropagated(final String topic, final int partition) {
+ final scala.Option<PartitionStateInfo> partitionStateOption = kafkaServer.apis().metadataCache().getPartitionInfo(
+ topic, partition);
+ if (partitionStateOption.isDefined()) {
+ final PartitionStateInfo partitionState = partitionStateOption.get();
+ final LeaderAndIsr leaderAndInSyncReplicas = partitionState.leaderIsrAndControllerEpoch().leaderAndIsr();
+
+ if (ZkUtils.getLeaderForPartition(getZkClient(), topic, partition) != null
+ && Request.isValidBrokerId(leaderAndInSyncReplicas.leader())
+ && leaderAndInSyncReplicas.isr().size() >= 1)
+ return true;
+
+ }
+ return false;
+ }
+
+ /**
+ * Waits until metadata is propagated.
+ *
+ * @param topic topic name
+ * @param partition partition
+ * @param timeout timeout value in millis
+ * @param interval interval in millis to sleep
+ * @throws TimeoutException
+ * @throws InterruptedException
+ */
+ private void waitUntilMetadataIsPropagated(final String topic, final int partition, final long timeout,
+ final long interval) throws TimeoutException, InterruptedException {
+ int attempt = 1;
+ final long startTime = System.currentTimeMillis();
+
+ while (true) {
+ if (isMetadataPropagated(topic, partition))
+ return;
+
+ final long duration = System.currentTimeMillis() - startTime;
+
+ if (duration < timeout)
+ Thread.sleep(interval);
+ else
+ throw new TimeoutException("metadata propagate timed out, attempt=" + attempt);
+
+ attempt++;
+ }
+
+ }
+
+ /**
+ * Sets up embedded Kafka Server
+ *
+ * @throws IOException
+ */
+ private void setupEmbeddedKafkaServer()
+ throws IOException {
+ A.ensure(zkReady, "Zookeeper should be setup before hand");
+
+ brokerConfig = new KafkaConfig(getBrokerConfig());
+ kafkaServer = new KafkaServer(brokerConfig, SystemTime$.MODULE$);
+ kafkaServer.startup();
+ }
+
+ /**
+ * Sets up embedded zooKeeper
+ *
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ private void setupEmbeddedZooKeeper()
+ throws IOException, InterruptedException {
+ EmbeddedZooKeeper zooKeeper = new EmbeddedZooKeeper(ZK_HOST, zkPort);
+ zooKeeper.startup();
+ zkPort = zooKeeper.getActualPort();
+ zkClient = new ZkClient(getZKAddress(), ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT, ZKStringSerializer$.MODULE$);
+ zkReady = true;
+ }
+
+ /**
+ * @return Kafka Broker Address.
+ */
+ private static String getBrokerAddress() {
+ return ZK_HOST + ":" + BROKER_PORT;
+ }
+
+ /**
+ * Gets KafKa Brofer Config
+ *
+ * @return Kafka Broker Config
+ * @throws IOException
+ */
+ private static Properties getBrokerConfig()
+ throws IOException {
+ Properties props = new Properties();
+ props.put("broker.id", "0");
+ props.put("host.name", ZK_HOST);
+ props.put("port", "" + BROKER_PORT);
+ props.put("log.dir", createTempDir("_cfg").getAbsolutePath());
+ props.put("zookeeper.connect", getZKAddress());
+ props.put("log.flush.interval.messages", "1");
+ props.put("replica.socket.timeout.ms", "1500");
+ return props;
+ }
+
+ /**
+ * @return Kafka Producer Config
+ */
+ private static ProducerConfig getProducerConfig() {
+ Properties props = new Properties();
+ props.put("metadata.broker.list", getBrokerAddress());
+ props.put("serializer.class", "kafka.serializer.StringEncoder");
+ props.put("key.serializer.class", "kafka.serializer.StringEncoder");
+ props.put("partitioner.class", "org.apache.ignite.kafka.SimplePartitioner");
+ return new ProducerConfig(props);
+ }
+
+ /**
+ * Creates Temp Directory
+ *
+ * @param prefix prefix
+ * @return Created File.
+ * @throws IOException
+ */
+ private static File createTempDir(final String prefix)
+ throws IOException {
+ final Path path = Files.createTempDirectory(prefix);
+ return path.toFile();
+
+ }
+
+ /**
+ * Creates Embedded ZooKeeper.
+ */
+ private static class EmbeddedZooKeeper {
+ /** Default ZooKeeper Host. */
+ private final String zkHost;
+
+ /** Default ZooKeeper Port. */
+ private final int zkPort;
+
+ /** NIO Context Factory. */
+ private NIOServerCnxnFactory factory;
+
+ /** Snapshot Directory. */
+ private File snapshotDir;
+
+ /** Log Directory. */
+ private File logDir;
+
+ /**
+ * Creates an embedded Zookeeper
+ * @param zkHost zookeeper host
+ * @param zkPort zookeeper port
+ */
+ EmbeddedZooKeeper(final String zkHost, final int zkPort) {
+ this.zkHost = zkHost;
+ this.zkPort = zkPort;
+ }
+
+ /**
+ * Starts up ZooKeeper.
+ *
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ void startup()
+ throws IOException, InterruptedException {
+ snapshotDir = createTempDir("_ss");
+ logDir = createTempDir("_log");
+ ZooKeeperServer zooServer = new ZooKeeperServer(snapshotDir, logDir, 500);
+ factory = new NIOServerCnxnFactory();
+ factory.configure(new InetSocketAddress(zkHost, zkPort), 16);
+ factory.startup(zooServer);
+ }
+
+ /**
+ *
+ * @return actual port zookeeper is started
+ */
+ int getActualPort() {
+ return factory.getLocalPort();
+ }
+
+ /**
+ * Shuts down ZooKeeper.
+ *
+ * @throws IOException
+ */
+ void shutdown()
+ throws IOException {
+ if (factory != null) {
+ factory.shutdown();
+
+ U.delete(snapshotDir);
+ U.delete(logDir);
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da07744c/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
new file mode 100644
index 0000000..5972639
--- /dev/null
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.kafka;
+
+import org.apache.ignite.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import kafka.consumer.*;
+import kafka.producer.*;
+import kafka.serializer.*;
+import kafka.utils.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.events.EventType.*;
+
+/**
+ * Tests {@link KafkaStreamer}.
+ */
+public class KafkaIgniteStreamerSelfTest
+ extends GridCommonAbstractTest {
+ /** Embedded Kafka. */
+ private KafkaEmbeddedBroker embeddedBroker;
+
+ /** Count. */
+ private static final int CNT = 100;
+
+ /** Test Topic. */
+ private static final String TOPIC_NAME = "page_visits";
+
+ /** Kafka Partition. */
+ private static final int PARTITIONS = 4;
+
+ /** Kafka Replication Factor. */
+ private static final int REPLICATION_FACTOR = 1;
+
+ /** Topic Message Key Prefix. */
+ private static final String KEY_PREFIX = "192.168.2.";
+
+ /** Topic Message Value Url. */
+ private static final String VALUE_URL = ",www.example.com,";
+
+ /** Constructor. */
+ public KafkaIgniteStreamerSelfTest() {
+ super(true);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected void beforeTest()
+ throws Exception {
+ grid().<Integer, String>getOrCreateCache(defaultCacheConfiguration());
+
+ embeddedBroker = new KafkaEmbeddedBroker();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected void afterTest()
+ throws Exception {
+ grid().cache(null).clear();
+
+ embeddedBroker.shutdown();
+ }
+
+ /**
+ * Tests Kafka streamer.
+ *
+ * @throws TimeoutException
+ * @throws InterruptedException
+ */
+ public void testKafkaStreamer()
+ throws TimeoutException, InterruptedException {
+ embeddedBroker.createTopic(TOPIC_NAME, PARTITIONS, REPLICATION_FACTOR);
+
+ Map<String, String> keyValueMap = produceStream(TOPIC_NAME);
+ consumerStream(TOPIC_NAME, keyValueMap);
+ }
+
+ /**
+ * Produces/Sends messages to Kafka.
+ *
+ * @param topic Topic name.
+ * @return Map of key value messages.
+ */
+ private Map<String, String> produceStream(final String topic) {
+ final Map<String, String> keyValueMap = new HashMap<>();
+
+ // Generate random subnets.
+ List<Integer> subnet = new ArrayList<>();
+
+ int i = 0;
+ while (i <= CNT)
+ subnet.add(++i);
+
+ Collections.shuffle(subnet);
+
+ final List<KeyedMessage<String, String>> messages = new ArrayList<>();
+ for (int event = 0; event < CNT; event++) {
+ long runtime = new Date().getTime();
+ String ip = KEY_PREFIX + subnet.get(event);
+ String msg = runtime + VALUE_URL + ip;
+ messages.add(new KeyedMessage<>(topic, ip, msg));
+ keyValueMap.put(ip, msg);
+ }
+
+ final Producer<String, String> producer = embeddedBroker.sendMessages(messages);
+ producer.close();
+
+ return keyValueMap;
+ }
+
+ /**
+ * Consumes Kafka Stream via ignite.
+ *
+ * @param topic Topic name.
+ * @param keyValueMap Expected key value map.
+ * @throws TimeoutException TimeoutException.
+ * @throws InterruptedException InterruptedException.
+ */
+ private void consumerStream(final String topic, final Map<String, String> keyValueMap)
+ throws TimeoutException, InterruptedException {
+
+ KafkaStreamer<String, String, String> kafkaStmr = null;
+
+ final Ignite ignite = grid();
+ try (IgniteDataStreamer<String, String> stmr = ignite.dataStreamer(null)) {
+
+ stmr.allowOverwrite(true);
+ stmr.autoFlushFrequency(10);
+
+ // Configure socket streamer.
+ kafkaStmr = new KafkaStreamer<>();
+
+ // Get the cache.
+ IgniteCache<String, String> cache = ignite.cache(null);
+
+ // Set ignite instance.
+ kafkaStmr.setIgnite(ignite);
+
+ // Set data streamer instance.
+ kafkaStmr.setStreamer(stmr);
+
+ // Set the topic.
+ kafkaStmr.setTopic(topic);
+
+ // Set the number of threads.
+ kafkaStmr.setThreads(4);
+
+ // Set the consumer configuration.
+ kafkaStmr.setConsumerConfig(createDefaultConsumerConfig(KafkaEmbeddedBroker.getZKAddress(),
+ "groupX"));
+
+ // Set the decoders.
+ StringDecoder stringDecoder = new StringDecoder(new VerifiableProperties());
+ kafkaStmr.setKeyDecoder(stringDecoder);
+ kafkaStmr.setValueDecoder(stringDecoder);
+
+ // Start kafka streamer.
+ kafkaStmr.start();
+
+ final CountDownLatch latch = new CountDownLatch(CNT);
+ IgniteBiPredicate<UUID, CacheEvent> locLsnr = new IgniteBiPredicate<UUID, CacheEvent>() {
+ @Override
+ public boolean apply(UUID uuid, CacheEvent evt) {
+ latch.countDown();
+ return true;
+ }
+ };
+
+ ignite.events(ignite.cluster().forCacheNodes(null)).remoteListen(locLsnr, null, EVT_CACHE_OBJECT_PUT);
+ latch.await();
+
+ for (Map.Entry<String, String> entry : keyValueMap.entrySet()) {
+ final String key = entry.getKey();
+ final String value = entry.getValue();
+
+ final String cacheValue = cache.get(key);
+ assertEquals(value, cacheValue);
+ }
+ }
+
+ finally {
+ // Shutdown kafka streamer.
+ kafkaStmr.stop();
+ }
+ }
+
+ /**
+ * Creates default consumer config.
+ *
+ * @param zooKeeper Zookeeper address <server:port>.
+ * @param groupId Group Id for kafka subscriber.
+ * @return {@link ConsumerConfig} kafka consumer configuration.
+ */
+ private ConsumerConfig createDefaultConsumerConfig(String zooKeeper, String groupId) {
+ A.notNull(zooKeeper, "zookeeper");
+ A.notNull(groupId, "groupId");
+
+ Properties props = new Properties();
+ props.put("zookeeper.connect", zooKeeper);
+ props.put("group.id", groupId);
+ props.put("zookeeper.session.timeout.ms", "400");
+ props.put("zookeeper.sync.time.ms", "200");
+ props.put("auto.commit.interval.ms", "1000");
+ props.put("auto.offset.reset", "smallest");
+
+ return new ConsumerConfig(props);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da07744c/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
new file mode 100644
index 0000000..b836b44
--- /dev/null
+++ b/modules/kafka/src/test/java/org/apache/ignite/stream/kafka/SimplePartitioner.java
@@ -0,0 +1,46 @@
+/*
+ * 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.kafka;
+
+import kafka.producer.*;
+
+/**
+ * Simple Partitioner for Kafka.
+ */
+@SuppressWarnings("UnusedDeclaration")
+public class SimplePartitioner
+ implements Partitioner {
+
+ /**
+ * Partitions the key based on the key value.
+ *
+ * @param key Key.
+ * @param partitionSize Partition size.
+ * @return partition Partition.
+ */
+ public int partition(Object key, int partitionSize) {
+ int partition = 0;
+ String keyStr = (String)key;
+ String[] keyValues = keyStr.split("\\.");
+ Integer intKey = Integer.parseInt(keyValues[3]);
+ if (intKey > 0) {
+ partition = intKey % partitionSize;
+ }
+ return partition;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/da07744c/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 44cb0ce..5017499 100644
--- a/pom.xml
+++ b/pom.xml
@@ -70,6 +70,7 @@
<module>modules/gce</module>
<module>modules/cloud</module>
<module>modules/mesos</module>
+ <module>modules/kafka</module>
</modules>
<profiles>
[20/20] incubator-ignite git commit: ignite-428 Retry message
consuming in case of error
Posted by sb...@apache.org.
ignite-428 Retry message consuming in case of error
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/ad5e99eb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/ad5e99eb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/ad5e99eb
Branch: refs/heads/ignite-428
Commit: ad5e99eb7678134ccae8af68db82f05a8b495b42
Parents: 6ca6a0c
Author: agura <ag...@gridgain.com>
Authored: Fri Jun 26 19:39:41 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Fri Jun 26 19:39:41 2015 +0300
----------------------------------------------------------------------
.../ignite/stream/kafka/KafkaStreamer.java | 37 ++++++++++++++++++--
1 file changed, 35 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ad5e99eb/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
----------------------------------------------------------------------
diff --git a/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java b/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
index 5761209..bc618e3 100644
--- a/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
+++ b/modules/kafka/src/main/java/org/apache/ignite/stream/kafka/KafkaStreamer.java
@@ -39,6 +39,9 @@ import java.util.concurrent.*;
* Example</a>
*/
public class KafkaStreamer<T, K, V> extends StreamAdapter<T, K, V> {
+ /** Retry timeout. */
+ private static final int RETRY_TIMEOUT = 10000;
+
/** Logger. */
private IgniteLogger log;
@@ -63,6 +66,9 @@ public class KafkaStreamer<T, K, V> extends StreamAdapter<T, K, V> {
/** Kafka consumer connector. */
private ConsumerConnector consumer;
+ /** Stopped. */
+ private volatile boolean stopped;
+
/**
* Sets the topic name.
*
@@ -138,12 +144,37 @@ public class KafkaStreamer<T, K, V> extends StreamAdapter<T, K, V> {
// Now launch all the consumer threads.
executor = Executors.newFixedThreadPool(threads);
+ stopped = false;
+
// Now create an object to consume the messages.
for (final KafkaStream<K, V> stream : streams) {
executor.submit(new Runnable() {
@Override public void run() {
- for (MessageAndMetadata<K, V> messageAndMetadata : stream)
- getStreamer().addData(messageAndMetadata.key(), messageAndMetadata.message());
+ while (!stopped) {
+ try {
+ for (ConsumerIterator<K, V> it = stream.iterator(); it.hasNext() && !stopped; ) {
+ MessageAndMetadata<K, V> msg = it.next();
+
+ try {
+ getStreamer().addData(msg.key(), msg.message());
+ }
+ catch (Exception e) {
+ log.warning("Message is ignored due to an error, msg = [" + msg + ']', e);
+ }
+ }
+ }
+ catch (Exception e) {
+ log.warning("Message can't be consumed from stream. Retry after " +
+ RETRY_TIMEOUT + " ms.", e);
+
+ try {
+ Thread.sleep(RETRY_TIMEOUT);
+ }
+ catch (InterruptedException ie) {
+ // No-op.
+ }
+ }
+ }
}
});
}
@@ -153,6 +184,8 @@ public class KafkaStreamer<T, K, V> extends StreamAdapter<T, K, V> {
* Stops streamer.
*/
public void stop() {
+ stopped = true;
+
if (consumer != null)
consumer.shutdown();
[10/20] incubator-ignite git commit: # ignite-sprint-7 fixed test
Posted by sb...@apache.org.
# ignite-sprint-7 fixed test
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/847ddf54
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/847ddf54
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/847ddf54
Branch: refs/heads/ignite-428
Commit: 847ddf543d635144d8c33f4afc33e76d0de92143
Parents: 01d842a
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jun 26 16:02:50 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jun 26 16:02:50 2015 +0300
----------------------------------------------------------------------
.../near/GridCachePartitionedMultiNodeFullApiSelfTest.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/847ddf54/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedMultiNodeFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedMultiNodeFullApiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedMultiNodeFullApiSelfTest.java
index 9445f9c..30c9e8a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedMultiNodeFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCachePartitionedMultiNodeFullApiSelfTest.java
@@ -358,7 +358,9 @@ public class GridCachePartitionedMultiNodeFullApiSelfTest extends GridCacheParti
assertEquals(0, cache1.localSize(NEAR));
assertEquals(5, cache1.localSize(CachePeekMode.ALL) - cache1.localSize(NEAR));
- assertEquals(nearEnabled() ? 2 : 0, cache2.localSize(NEAR));
+ boolean nearEnabled = cache2.getConfiguration(CacheConfiguration.class).getNearConfiguration() != null;
+
+ assertEquals(nearEnabled ? 2 : 0, cache2.localSize(NEAR));
assertEquals(0, cache2.localSize(CachePeekMode.ALL) - cache2.localSize(NEAR));
}
[12/20] incubator-ignite git commit: Merge branches 'ignite-gg-10404'
and 'ignite-sprint-7' of
https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-gg-10404
Posted by sb...@apache.org.
Merge branches 'ignite-gg-10404' and 'ignite-sprint-7' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-gg-10404
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/50a46264
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/50a46264
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/50a46264
Branch: refs/heads/ignite-428
Commit: 50a4626439afcf9c8113476da0a425d35409f153
Parents: c62c410 6e23608
Author: AKuznetsov <ak...@gridgain.com>
Authored: Fri Jun 26 20:53:15 2015 +0700
Committer: AKuznetsov <ak...@gridgain.com>
Committed: Fri Jun 26 20:53:15 2015 +0700
----------------------------------------------------------------------
RELEASE_NOTES.txt | 12 +
examples/pom.xml | 2 +-
modules/aop/pom.xml | 2 +-
modules/aws/pom.xml | 2 +-
.../s3/S3CheckpointManagerSelfTest.java | 2 +-
.../checkpoint/s3/S3CheckpointSpiSelfTest.java | 4 +-
.../s3/S3CheckpointSpiStartStopSelfTest.java | 2 +-
.../s3/S3SessionCheckpointSelfTest.java | 2 +-
.../s3/TcpDiscoveryS3IpFinderSelfTest.java | 2 +-
modules/clients/pom.xml | 2 +-
.../ClientAbstractConnectivitySelfTest.java | 4 +-
modules/cloud/pom.xml | 2 +-
modules/codegen/pom.xml | 2 +-
modules/core/pom.xml | 2 +-
.../apache/ignite/IgniteSystemProperties.java | 3 +
.../apache/ignite/cache/query/ScanQuery.java | 23 +-
.../cache/store/jdbc/CacheJdbcBlobStore.java | 22 +-
.../store/jdbc/CacheJdbcBlobStoreFactory.java | 290 +++++++
.../cache/store/jdbc/CacheJdbcPojoStore.java | 6 +-
.../store/jdbc/CacheJdbcPojoStoreFactory.java | 148 ++++
.../org/apache/ignite/cluster/ClusterGroup.java | 9 +
.../org/apache/ignite/cluster/ClusterNode.java | 2 +
.../ignite/compute/ComputeTaskSplitAdapter.java | 2 +-
.../configuration/CacheConfiguration.java | 3 +-
.../configuration/IgniteReflectionFactory.java | 81 +-
.../ignite/internal/GridKernalContextImpl.java | 8 +-
.../apache/ignite/internal/IgniteKernal.java | 11 +-
.../internal/MarshallerContextAdapter.java | 18 +-
.../ignite/internal/MarshallerContextImpl.java | 26 +-
.../client/GridClientConfiguration.java | 2 +-
.../GridClientOptimizedMarshaller.java | 26 +
.../impl/GridTcpRouterNioListenerAdapter.java | 2 +-
.../internal/cluster/ClusterGroupAdapter.java | 38 +
.../cluster/IgniteClusterAsyncImpl.java | 5 +
.../internal/interop/InteropBootstrap.java | 3 +-
.../internal/interop/InteropIgnition.java | 5 +-
.../internal/managers/GridManagerAdapter.java | 8 +-
.../discovery/GridDiscoveryManager.java | 39 +-
.../affinity/AffinityTopologyVersion.java | 7 -
.../processors/cache/GridCacheAdapter.java | 4 +
.../processors/cache/GridCacheContext.java | 2 +-
.../processors/cache/GridCacheIoManager.java | 64 +-
.../GridCachePartitionExchangeManager.java | 77 +-
.../processors/cache/GridCacheProcessor.java | 34 +-
.../processors/cache/GridCacheSwapManager.java | 12 +-
.../processors/cache/GridCacheUtils.java | 9 +
.../processors/cache/IgniteCacheProxy.java | 12 +
.../distributed/GridCacheTxRecoveryRequest.java | 26 +-
.../GridCacheTxRecoveryResponse.java | 14 +-
.../distributed/GridDistributedBaseMessage.java | 77 +-
.../distributed/GridDistributedLockRequest.java | 54 +-
.../GridDistributedLockResponse.java | 14 +-
.../GridDistributedTxFinishRequest.java | 46 +-
.../GridDistributedTxPrepareRequest.java | 62 +-
.../GridDistributedTxPrepareResponse.java | 64 +-
.../GridDistributedUnlockRequest.java | 6 +-
.../distributed/dht/GridDhtLocalPartition.java | 59 +-
.../distributed/dht/GridDhtLockFuture.java | 2 +-
.../distributed/dht/GridDhtLockRequest.java | 72 +-
.../distributed/dht/GridDhtLockResponse.java | 18 +-
.../dht/GridDhtPartitionTopologyImpl.java | 4 +-
.../dht/GridDhtPartitionsReservation.java | 292 +++++++
.../dht/GridDhtTransactionalCacheAdapter.java | 2 +-
.../distributed/dht/GridDhtTxFinishRequest.java | 38 +-
.../dht/GridDhtTxPrepareRequest.java | 54 +-
.../dht/GridDhtTxPrepareResponse.java | 22 +-
.../distributed/dht/GridDhtUnlockRequest.java | 6 +-
.../cache/distributed/dht/GridReservable.java | 35 +
.../dht/atomic/GridDhtAtomicCache.java | 9 +-
.../dht/preloader/GridDhtPartitionMap.java | 26 +-
.../GridDhtPartitionsExchangeFuture.java | 95 ++-
.../dht/preloader/GridDhtPreloader.java | 2 +-
.../distributed/near/GridNearLockRequest.java | 58 +-
.../distributed/near/GridNearLockResponse.java | 26 +-
.../near/GridNearTxFinishRequest.java | 26 +-
.../near/GridNearTxPrepareRequest.java | 50 +-
.../near/GridNearTxPrepareResponse.java | 46 +-
.../distributed/near/GridNearUnlockRequest.java | 2 +-
.../cache/query/GridCacheQueryManager.java | 33 -
.../cache/query/GridCacheTwoStepQuery.java | 22 +-
.../continuous/CacheContinuousQueryHandler.java | 8 +
.../cache/transactions/IgniteTxHandler.java | 5 +-
.../transactions/IgniteTxLocalAdapter.java | 12 +-
.../datastructures/DataStructuresProcessor.java | 129 +--
.../dr/IgniteDrDataStreamerCacheUpdater.java | 7 +-
.../processors/hadoop/HadoopJobInfo.java | 4 +-
.../hadoop/counter/HadoopCounterWriter.java | 5 +-
.../offheap/GridOffHeapProcessor.java | 19 +-
.../processors/plugin/CachePluginManager.java | 10 +-
.../plugin/IgnitePluginProcessor.java | 16 +-
.../processors/query/GridQueryIndexing.java | 14 +-
.../processors/query/GridQueryProcessor.java | 123 ++-
.../messages/GridQueryNextPageResponse.java | 35 +-
.../h2/twostep/messages/GridQueryRequest.java | 111 ++-
.../rest/client/message/GridRouterRequest.java | 18 +
.../rest/client/message/GridRouterResponse.java | 18 +
.../rest/protocols/tcp/GridTcpRestProtocol.java | 3 +-
.../processors/task/GridTaskProcessor.java | 23 +-
.../internal/util/GridConfigurationFinder.java | 55 +-
.../apache/ignite/internal/util/GridDebug.java | 48 +-
.../ignite/internal/util/IgniteUtils.java | 27 +-
.../ignite/internal/util/nio/GridNioServer.java | 64 +-
.../util/spring/IgniteSpringHelper.java | 10 +
.../SpringApplicationContextResource.java | 4 +-
.../apache/ignite/resources/SpringResource.java | 6 +-
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 35 +-
.../org/apache/ignite/spi/IgniteSpiContext.java | 9 +-
.../communication/tcp/TcpCommunicationSpi.java | 148 +++-
.../tcp/TcpCommunicationSpiMBean.java | 19 +
.../ignite/spi/discovery/DiscoverySpi.java | 3 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 657 +++++++++++-----
.../ignite/spi/discovery/tcp/ServerImpl.java | 376 ++++++---
.../spi/discovery/tcp/TcpDiscoveryImpl.java | 69 +-
.../spi/discovery/tcp/TcpDiscoverySpi.java | 51 +-
.../tcp/internal/TcpDiscoveryNode.java | 18 +
.../ipfinder/TcpDiscoveryIpFinderAdapter.java | 34 +-
.../TcpDiscoveryMulticastIpFinder.java | 19 +-
.../messages/TcpDiscoveryAbstractMessage.java | 10 +-
.../messages/TcpDiscoveryNodeFailedMessage.java | 18 +
.../core/src/main/resources/ignite.properties | 2 +-
.../internal/ClusterForHostsSelfTest.java | 113 +++
.../internal/ClusterGroupAbstractTest.java | 777 ++++++++++++++++++
.../ignite/internal/ClusterGroupSelfTest.java | 251 ++++++
.../internal/GridDiscoveryEventSelfTest.java | 12 +-
...ridFailFastNodeFailureDetectionSelfTest.java | 17 +-
.../internal/GridProjectionAbstractTest.java | 784 -------------------
.../ignite/internal/GridProjectionSelfTest.java | 251 ------
.../apache/ignite/internal/GridSelfTest.java | 30 +-
.../GridTaskFailoverAffinityRunTest.java | 170 ++++
.../IgniteSlowClientDetectionSelfTest.java | 187 +++++
.../GridDiscoveryManagerAliveCacheSelfTest.java | 17 +-
.../cache/CacheClientStoreSelfTest.java | 228 ++++++
...acheReadOnlyTransactionalClientSelfTest.java | 327 --------
.../CacheReadThroughAtomicRestartSelfTest.java | 32 +
...heReadThroughLocalAtomicRestartSelfTest.java | 32 +
.../CacheReadThroughLocalRestartSelfTest.java | 32 +
...dThroughReplicatedAtomicRestartSelfTest.java | 32 +
...cheReadThroughReplicatedRestartSelfTest.java | 32 +
.../cache/CacheReadThroughRestartSelfTest.java | 133 ++++
.../CacheStoreUsageMultinodeAbstractTest.java | 305 ++++++++
...eUsageMultinodeDynamicStartAbstractTest.java | 169 ++++
...oreUsageMultinodeDynamicStartAtomicTest.java | 32 +
...heStoreUsageMultinodeDynamicStartTxTest.java | 32 +
...reUsageMultinodeStaticStartAbstractTest.java | 158 ++++
...toreUsageMultinodeStaticStartAtomicTest.java | 32 +
...cheStoreUsageMultinodeStaticStartTxTest.java | 32 +
.../GridCacheAbstractFailoverSelfTest.java | 8 +-
.../cache/GridCacheAbstractFullApiSelfTest.java | 24 +-
.../cache/GridCacheAbstractSelfTest.java | 2 +-
.../cache/GridCacheDaemonNodeStopSelfTest.java | 119 +++
...ridCacheMultinodeUpdateAbstractSelfTest.java | 9 +
.../cache/GridCachePutAllFailoverSelfTest.java | 5 -
.../cache/GridCacheVersionMultinodeTest.java | 8 +-
.../IgniteCacheAbstractStopBusySelfTest.java | 30 +-
.../IgniteCacheAtomicStopBusySelfTest.java | 8 +-
...CacheP2pUnmarshallingRebalanceErrorTest.java | 15 +-
.../IgniteCacheP2pUnmarshallingTxErrorTest.java | 25 +-
...gniteCacheTransactionalStopBusySelfTest.java | 8 +-
...eDynamicCacheStartNoExchangeTimeoutTest.java | 466 +++++++++++
.../cache/IgniteDynamicCacheStartSelfTest.java | 37 +
...GridCacheQueueMultiNodeAbstractSelfTest.java | 4 +-
.../GridCacheSetAbstractSelfTest.java | 22 +-
.../IgniteDataStructureWithJobTest.java | 111 +++
...ridCachePartitionNotLoadedEventSelfTest.java | 82 ++
.../distributed/IgniteCache150ClientsTest.java | 189 +++++
.../IgniteCacheClientNodeConcurrentStart.java | 14 +-
...teCacheClientNodePartitionsExchangeTest.java | 1 +
.../distributed/IgniteCacheManyClientsTest.java | 191 ++++-
.../IgniteCacheTxMessageRecoveryTest.java | 5 +
.../dht/GridCacheColocatedFailoverSelfTest.java | 5 -
...idCacheNearOnlyMultiNodeFullApiSelfTest.java | 5 -
.../GridCachePartitionedFailoverSelfTest.java | 5 -
...achePartitionedMultiNodeFullApiSelfTest.java | 53 +-
.../GridCachePartitionedTxSalvageSelfTest.java | 37 +-
.../GridCacheReplicatedFailoverSelfTest.java | 2 +-
.../IgniteCacheTxStoreSessionTest.java | 4 +
...CacheClientWriteBehindStoreAbstractTest.java | 104 +++
...teCacheClientWriteBehindStoreAtomicTest.java | 38 +
.../IgnteCacheClientWriteBehindStoreTxTest.java | 32 +
.../DataStreamProcessorSelfTest.java | 3 +-
.../DataStreamerMultiThreadedSelfTest.java | 3 +
.../internal/util/IgniteUtilsSelfTest.java | 22 +
.../marshaller/MarshallerContextTestImpl.java | 18 +
.../GridTcpCommunicationSpiConfigSelfTest.java | 1 -
.../tcp/TcpClientDiscoverySpiSelfTest.java | 338 +++++++-
.../spi/discovery/tcp/TcpDiscoverySelfTest.java | 44 +-
.../testframework/GridSpiTestContext.java | 7 +-
.../testframework/junits/GridAbstractTest.java | 2 +-
.../junits/GridTestKernalContext.java | 3 +-
.../junits/common/GridCommonAbstractTest.java | 15 +-
.../ignite/testsuites/IgniteBasicTestSuite.java | 7 +-
.../IgniteCacheDataStructuresSelfTestSuite.java | 1 +
.../IgniteCacheFailoverTestSuite.java | 8 -
.../IgniteCacheFailoverTestSuite2.java | 47 ++
.../ignite/testsuites/IgniteCacheTestSuite.java | 4 +-
.../testsuites/IgniteCacheTestSuite3.java | 1 +
.../testsuites/IgniteCacheTestSuite4.java | 18 +-
.../IgniteCacheWriteBehindTestSuite.java | 2 +
.../testsuites/IgniteClientTestSuite.java | 38 +
.../testsuites/IgniteComputeGridTestSuite.java | 1 +
.../testsuites/IgniteKernalSelfTestSuite.java | 2 +-
.../ignite/util/TestTcpCommunicationSpi.java | 21 +
modules/extdata/p2p/pom.xml | 2 +-
.../p2p/GridP2PContinuousDeploymentTask1.java | 2 +-
modules/extdata/uri/pom.xml | 2 +-
modules/gce/pom.xml | 2 +-
modules/geospatial/pom.xml | 2 +-
modules/hadoop/pom.xml | 80 +-
.../fs/IgniteHadoopFileSystemCounterWriter.java | 9 +-
.../processors/hadoop/HadoopClassLoader.java | 29 +
.../processors/hadoop/HadoopDefaultJobInfo.java | 27 +-
.../internal/processors/hadoop/HadoopUtils.java | 237 ------
.../hadoop/SecondaryFileSystemProvider.java | 3 +-
.../hadoop/fs/HadoopFileSystemCacheUtils.java | 241 ++++++
.../hadoop/fs/HadoopFileSystemsUtils.java | 11 +
.../hadoop/fs/HadoopLazyConcurrentMap.java | 5 +
.../hadoop/jobtracker/HadoopJobTracker.java | 25 +-
.../child/HadoopChildProcessRunner.java | 3 +-
.../processors/hadoop/v2/HadoopV2Job.java | 84 +-
.../hadoop/v2/HadoopV2JobResourceManager.java | 22 +-
.../hadoop/v2/HadoopV2TaskContext.java | 37 +-
.../apache/ignite/igfs/IgfsEventsTestSuite.java | 5 +-
.../processors/hadoop/HadoopMapReduceTest.java | 2 +-
.../processors/hadoop/HadoopTasksV1Test.java | 7 +-
.../processors/hadoop/HadoopTasksV2Test.java | 7 +-
.../processors/hadoop/HadoopV2JobSelfTest.java | 6 +-
.../collections/HadoopAbstractMapTest.java | 3 +-
.../testsuites/IgniteHadoopTestSuite.java | 2 +-
.../IgniteIgfsLinuxAndMacOSTestSuite.java | 3 +-
modules/hibernate/pom.xml | 16 +-
.../hibernate/CacheHibernateBlobStore.java | 87 +-
.../CacheHibernateBlobStoreFactory.java | 235 ++++++
.../hibernate/src/test/config/factory-cache.xml | 59 ++
.../src/test/config/factory-cache1.xml | 61 ++
.../config/factory-incorrect-store-cache.xml | 56 ++
.../CacheHibernateStoreFactorySelfTest.java | 273 +++++++
.../testsuites/IgniteHibernateTestSuite.java | 2 +
modules/indexing/pom.xml | 2 +-
.../processors/query/h2/IgniteH2Indexing.java | 81 +-
.../query/h2/sql/GridSqlQuerySplitter.java | 49 +-
.../query/h2/twostep/GridMapQueryExecutor.java | 321 ++++++--
.../query/h2/twostep/GridMergeIndex.java | 17 +-
.../h2/twostep/GridMergeIndexUnsorted.java | 7 +-
.../h2/twostep/GridReduceQueryExecutor.java | 650 ++++++++++++---
.../query/h2/twostep/GridResultPage.java | 21 +-
.../CacheAbstractQueryMetricsSelfTest.java | 205 +++++
.../CachePartitionedQueryMetricsSelfTest.java | 32 +
.../CacheReplicatedQueryMetricsSelfTest.java | 32 +
.../cache/GridCacheCrossCacheQuerySelfTest.java | 3 +-
.../cache/GridCacheQueryMetricsSelfTest.java | 206 -----
.../cache/IgniteCacheOffheapEvictQueryTest.java | 196 +++++
.../IgniteCacheQueryMultiThreadedSelfTest.java | 1 -
...QueryOffheapEvictsMultiThreadedSelfTest.java | 5 +
...lientQueryReplicatedNodeRestartSelfTest.java | 419 ++++++++++
.../IgniteCacheQueryNodeRestartSelfTest.java | 36 +-
.../IgniteCacheQueryNodeRestartSelfTest2.java | 383 +++++++++
.../query/h2/sql/BaseH2CompareQueryTest.java | 2 +-
.../IgniteCacheQuerySelfTestSuite.java | 9 +-
modules/jcl/pom.xml | 2 +-
modules/jta/pom.xml | 2 +-
.../cache/jta/GridCacheXAResource.java | 18 +-
.../processors/cache/GridCacheJtaSelfTest.java | 2 +-
modules/log4j/pom.xml | 2 +-
modules/mesos/pom.xml | 2 +-
modules/rest-http/pom.xml | 2 +-
modules/scalar-2.10/pom.xml | 2 +-
modules/scalar/pom.xml | 2 +-
modules/schedule/pom.xml | 2 +-
modules/schema-import/pom.xml | 2 +-
modules/slf4j/pom.xml | 2 +-
modules/spark-2.10/pom.xml | 2 +-
modules/spark/pom.xml | 2 +-
modules/spring/pom.xml | 9 +-
.../GridResourceSpringBeanInjector.java | 2 +-
.../util/spring/IgniteSpringHelperImpl.java | 17 +
.../src/test/config/incorrect-store-cache.xml | 57 ++
modules/spring/src/test/config/node.xml | 43 +
modules/spring/src/test/config/node1.xml | 45 ++
.../test/config/pojo-incorrect-store-cache.xml | 56 ++
modules/spring/src/test/config/store-cache.xml | 59 ++
modules/spring/src/test/config/store-cache1.xml | 62 ++
.../jdbc/CacheJdbcBlobStoreFactorySelfTest.java | 172 ++++
.../jdbc/CacheJdbcPojoStoreFactorySelfTest.java | 193 +++++
.../testsuites/IgniteSpringTestSuite.java | 5 +
modules/ssh/pom.xml | 2 +-
modules/tools/pom.xml | 2 +-
.../ignite/tools/classgen/ClassesGenerator.java | 30 +-
modules/urideploy/pom.xml | 2 +-
modules/visor-console-2.10/pom.xml | 2 +-
modules/visor-console/pom.xml | 2 +-
.../commands/cache/VisorCacheCommand.scala | 7 +-
modules/visor-plugins/pom.xml | 2 +-
modules/web/pom.xml | 2 +-
.../IgniteWebSessionSelfTestSuite.java | 2 +-
modules/yardstick/pom.xml | 2 +-
pom.xml | 2 +-
scripts/git-patch-prop.sh | 2 +-
297 files changed, 12522 insertions(+), 3763 deletions(-)
----------------------------------------------------------------------
[04/20] incubator-ignite git commit: # ignite-sprint-7: update sprint
version at git-patch-prop.sh
Posted by sb...@apache.org.
# ignite-sprint-7: update sprint version at git-patch-prop.sh
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/3e8ddb47
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/3e8ddb47
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/3e8ddb47
Branch: refs/heads/ignite-428
Commit: 3e8ddb47dcef245596e9b7028a3f37a43e568185
Parents: e1c49b7
Author: ashutak <as...@gridgain.com>
Authored: Thu Jun 25 21:04:11 2015 +0300
Committer: ashutak <as...@gridgain.com>
Committed: Thu Jun 25 21:04:11 2015 +0300
----------------------------------------------------------------------
scripts/git-patch-prop.sh | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3e8ddb47/scripts/git-patch-prop.sh
----------------------------------------------------------------------
diff --git a/scripts/git-patch-prop.sh b/scripts/git-patch-prop.sh
index c856fb4..16f907e 100644
--- a/scripts/git-patch-prop.sh
+++ b/scripts/git-patch-prop.sh
@@ -19,6 +19,6 @@
#
# Git patch-file maker/applier properties.
#
-IGNITE_DEFAULT_BRANCH='ignite-sprint-5'
+IGNITE_DEFAULT_BRANCH='ignite-sprint-7'
PATCHES_HOME=${IGNITE_HOME}
[02/20] incubator-ignite git commit: Merge branches 'ignite-gg-10404'
and 'ignite-sprint-6' of
https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-gg-10404
Posted by sb...@apache.org.
Merge branches 'ignite-gg-10404' and 'ignite-sprint-6' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-gg-10404
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/67790714
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/67790714
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/67790714
Branch: refs/heads/ignite-428
Commit: 67790714098c9c413a06b855c5b51e665a39e755
Parents: 32697ab b087aca
Author: AKuznetsov <ak...@gridgain.com>
Authored: Fri Jun 12 14:15:58 2015 +0700
Committer: AKuznetsov <ak...@gridgain.com>
Committed: Fri Jun 12 14:15:58 2015 +0700
----------------------------------------------------------------------
.../processors/cache/GridCacheProcessor.java | 17 ++
.../continuous/GridContinuousProcessor.java | 15 +-
...teStartCacheInTransactionAtomicSelfTest.java | 32 +++
.../IgniteStartCacheInTransactionSelfTest.java | 254 +++++++++++++++++++
.../testsuites/IgniteCacheTestSuite4.java | 3 +
5 files changed, 315 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
[14/20] incubator-ignite git commit: Merge branch 'ignite-1017' into
ignite-sprint-7
Posted by sb...@apache.org.
Merge branch 'ignite-1017' into ignite-sprint-7
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/5fb9f2fc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/5fb9f2fc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/5fb9f2fc
Branch: refs/heads/ignite-428
Commit: 5fb9f2fc14e51ce96bc884acf4e169d7dcc43fe8
Parents: 6e23608 5d09ed1
Author: ashutak <as...@gridgain.com>
Authored: Fri Jun 26 17:55:04 2015 +0300
Committer: ashutak <as...@gridgain.com>
Committed: Fri Jun 26 17:55:04 2015 +0300
----------------------------------------------------------------------
.../core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java | 2 +-
.../org/apache/ignite/internal/cluster/ClusterGroupAdapter.java | 2 +-
.../org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java | 2 +-
3 files changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
[13/20] incubator-ignite git commit: # ignite-1017: review
Posted by sb...@apache.org.
# ignite-1017: review
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/5d09ed1b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/5d09ed1b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/5d09ed1b
Branch: refs/heads/ignite-428
Commit: 5d09ed1b3d295efb0670aaa943a0c3eb9c3a7267
Parents: 5b237e1
Author: ashutak <as...@gridgain.com>
Authored: Fri Jun 26 17:53:22 2015 +0300
Committer: ashutak <as...@gridgain.com>
Committed: Fri Jun 26 17:53:22 2015 +0300
----------------------------------------------------------------------
.../org/apache/ignite/cluster/ClusterGroup.java | 16 ----------------
.../internal/cluster/ClusterGroupAdapter.java | 7 -------
.../internal/cluster/IgniteClusterAsyncImpl.java | 5 -----
3 files changed, 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5d09ed1b/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java b/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
index 9627f76..62f3027 100644
--- a/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
+++ b/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
@@ -131,22 +131,6 @@ public interface ClusterGroup {
* @param name Name of the attribute.
* @param val Optional attribute value to match.
* @return Cluster group for nodes containing specified attribute.
- *
- * @deprecated use {@link ClusterGroup#forAttribute(String name, @Nullable Object val}
- */
- public ClusterGroup forAttribute(String name, @Nullable String val);
-
- /**
- * Creates a new cluster group for nodes containing given name and value
- * specified in user attributes.
- * <p>
- * User attributes for every node are optional and can be specified in
- * grid node configuration. See {@link IgniteConfiguration#getUserAttributes()}
- * for more information.
- *
- * @param name Name of the attribute.
- * @param val Optional attribute value to match.
- * @return Cluster group for nodes containing specified attribute.
*/
public ClusterGroup forAttribute(String name, @Nullable Object val);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5d09ed1b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
index 414f5ba..6c42067 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
@@ -345,13 +345,6 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
}
/** {@inheritDoc} */
- @Override public final ClusterGroup forAttribute(String name, @Nullable final String val) {
- A.notNull(name, "n");
-
- return forPredicate(new AttributeFilter(name, val));
- }
-
- /** {@inheritDoc} */
@Override public final ClusterGroup forAttribute(String name, @Nullable final Object val) {
A.notNull(name, "n");
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5d09ed1b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
index f676261..01c62b7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
@@ -192,11 +192,6 @@ public class IgniteClusterAsyncImpl extends AsyncSupportAdapter<IgniteCluster>
}
/** {@inheritDoc} */
- @Override public ClusterGroup forAttribute(String name, @Nullable String val) {
- return cluster.forAttribute(name, val);
- }
-
- /** {@inheritDoc} */
@Override public ClusterGroup forAttribute(String name, @Nullable Object val) {
return cluster.forAttribute(name, val);
}
[06/20] incubator-ignite git commit: Merge branch 'ignite-sprint-7'
into ignite-1006
Posted by sb...@apache.org.
Merge branch 'ignite-sprint-7' into ignite-1006
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/a8232be6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/a8232be6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/a8232be6
Branch: refs/heads/ignite-428
Commit: a8232be601851ca9658502de12f38ac86c1e6758
Parents: c6f66c6 3e8ddb4
Author: ashutak <as...@gridgain.com>
Authored: Thu Jun 25 21:12:09 2015 +0300
Committer: ashutak <as...@gridgain.com>
Committed: Thu Jun 25 21:12:09 2015 +0300
----------------------------------------------------------------------
DEVNOTES.txt | 21 +
RELEASE_NOTES.txt | 12 +
assembly/dependencies-fabric.xml | 1 +
examples/config/example-cache.xml | 2 +
examples/pom.xml | 36 +-
idea/ignite_codeStyle.xml | 147 ++++
modules/aop/pom.xml | 2 +-
modules/aws/pom.xml | 2 +-
.../s3/S3CheckpointManagerSelfTest.java | 2 +-
.../checkpoint/s3/S3CheckpointSpiSelfTest.java | 4 +-
.../s3/S3CheckpointSpiStartStopSelfTest.java | 2 +-
.../s3/S3SessionCheckpointSelfTest.java | 2 +-
.../s3/TcpDiscoveryS3IpFinderSelfTest.java | 2 +-
modules/clients/pom.xml | 2 +-
.../ClientAbstractConnectivitySelfTest.java | 4 +-
.../client/router/TcpSslRouterSelfTest.java | 5 +
.../client/suite/IgniteClientTestSuite.java | 3 +-
modules/cloud/pom.xml | 2 +-
.../cloud/TcpDiscoveryCloudIpFinder.java | 25 +-
.../TcpDiscoveryCloudIpFinderSelfTest.java | 3 +-
modules/codegen/pom.xml | 2 +-
modules/core/pom.xml | 3 +-
.../apache/ignite/IgniteSystemProperties.java | 6 +
.../apache/ignite/cache/query/ScanQuery.java | 48 +-
.../cache/store/jdbc/CacheJdbcBlobStore.java | 22 +-
.../store/jdbc/CacheJdbcBlobStoreFactory.java | 290 +++++++
.../cache/store/jdbc/CacheJdbcPojoStore.java | 6 +-
.../store/jdbc/CacheJdbcPojoStoreFactory.java | 148 ++++
.../org/apache/ignite/cluster/ClusterGroup.java | 9 +
.../org/apache/ignite/cluster/ClusterNode.java | 2 +
.../ignite/compute/ComputeTaskSplitAdapter.java | 2 +-
.../configuration/CacheConfiguration.java | 4 +-
.../configuration/IgniteReflectionFactory.java | 81 +-
.../ignite/internal/GridKernalContextImpl.java | 8 +-
.../ignite/internal/GridPluginContext.java | 6 +
.../apache/ignite/internal/IgniteKernal.java | 13 +-
.../internal/MarshallerContextAdapter.java | 48 +-
.../ignite/internal/MarshallerContextImpl.java | 36 +-
.../client/GridClientConfiguration.java | 2 +-
.../GridClientOptimizedMarshaller.java | 26 +
.../impl/GridTcpRouterNioListenerAdapter.java | 2 +-
.../internal/cluster/ClusterGroupAdapter.java | 38 +
.../cluster/IgniteClusterAsyncImpl.java | 5 +
.../internal/interop/InteropBootstrap.java | 3 +-
.../internal/interop/InteropIgnition.java | 57 +-
.../internal/interop/InteropProcessor.java | 8 +
.../internal/managers/GridManagerAdapter.java | 8 +-
.../discovery/GridDiscoveryManager.java | 39 +-
.../affinity/AffinityTopologyVersion.java | 7 -
.../affinity/GridAffinityAssignmentCache.java | 5 +-
.../processors/cache/GridCacheAdapter.java | 19 +-
.../processors/cache/GridCacheContext.java | 2 +-
.../processors/cache/GridCacheIoManager.java | 64 +-
.../processors/cache/GridCacheMessage.java | 51 --
.../GridCachePartitionExchangeManager.java | 73 +-
.../processors/cache/GridCacheProcessor.java | 91 ++-
.../processors/cache/GridCacheSwapManager.java | 67 +-
.../processors/cache/GridCacheUtils.java | 9 +
.../processors/cache/IgniteCacheProxy.java | 23 +-
.../processors/cache/KeyCacheObjectImpl.java | 11 +-
.../processors/cache/QueryCursorImpl.java | 23 +-
.../distributed/GridCacheTxRecoveryRequest.java | 26 +-
.../GridCacheTxRecoveryResponse.java | 14 +-
.../distributed/GridDistributedBaseMessage.java | 77 +-
.../distributed/GridDistributedLockRequest.java | 54 +-
.../GridDistributedLockResponse.java | 14 +-
.../GridDistributedTxFinishRequest.java | 46 +-
.../GridDistributedTxPrepareRequest.java | 62 +-
.../GridDistributedTxPrepareResponse.java | 64 +-
.../GridDistributedUnlockRequest.java | 6 +-
.../distributed/dht/GridDhtLocalPartition.java | 66 +-
.../distributed/dht/GridDhtLockFuture.java | 2 +-
.../distributed/dht/GridDhtLockRequest.java | 72 +-
.../distributed/dht/GridDhtLockResponse.java | 18 +-
.../dht/GridDhtPartitionTopologyImpl.java | 4 +-
.../dht/GridDhtPartitionsReservation.java | 292 +++++++
.../dht/GridDhtTransactionalCacheAdapter.java | 2 +-
.../distributed/dht/GridDhtTxFinishRequest.java | 38 +-
.../dht/GridDhtTxPrepareRequest.java | 54 +-
.../dht/GridDhtTxPrepareResponse.java | 22 +-
.../distributed/dht/GridDhtUnlockRequest.java | 6 +-
.../cache/distributed/dht/GridReservable.java | 35 +
.../dht/atomic/GridDhtAtomicCache.java | 9 +-
.../dht/preloader/GridDhtPartitionMap.java | 26 +-
.../GridDhtPartitionsExchangeFuture.java | 95 ++-
.../dht/preloader/GridDhtPreloader.java | 2 +-
.../distributed/near/GridNearLockRequest.java | 58 +-
.../distributed/near/GridNearLockResponse.java | 26 +-
.../near/GridNearTxFinishRequest.java | 26 +-
.../near/GridNearTxPrepareRequest.java | 50 +-
.../near/GridNearTxPrepareResponse.java | 46 +-
.../distributed/near/GridNearUnlockRequest.java | 2 +-
.../processors/cache/query/CacheQuery.java | 2 +-
.../query/GridCacheDistributedQueryManager.java | 3 +
.../cache/query/GridCacheQueryAdapter.java | 147 +++-
.../cache/query/GridCacheQueryManager.java | 243 +++---
.../cache/query/GridCacheQueryRequest.java | 47 +-
.../cache/query/GridCacheTwoStepQuery.java | 22 +-
.../processors/cache/query/QueryCursorEx.java | 8 +
.../continuous/CacheContinuousQueryHandler.java | 8 +
.../cache/transactions/IgniteTxHandler.java | 5 +-
.../transactions/IgniteTxLocalAdapter.java | 12 +-
.../cacheobject/IgniteCacheObjectProcessor.java | 9 +-
.../IgniteCacheObjectProcessorImpl.java | 12 +-
.../continuous/GridContinuousProcessor.java | 15 +-
.../datastreamer/DataStreamerCacheUpdaters.java | 2 +-
.../datastreamer/DataStreamerImpl.java | 8 +-
.../datastructures/DataStructuresProcessor.java | 129 +--
.../datastructures/GridCacheSetImpl.java | 4 +-
.../dr/IgniteDrDataStreamerCacheUpdater.java | 7 +-
.../processors/hadoop/HadoopJobInfo.java | 4 +-
.../hadoop/counter/HadoopCounterWriter.java | 5 +-
.../offheap/GridOffHeapProcessor.java | 19 +-
.../processors/plugin/CachePluginManager.java | 10 +-
.../plugin/IgnitePluginProcessor.java | 16 +-
.../portable/GridPortableInputStream.java | 10 +
.../processors/query/GridQueryIndexing.java | 18 +-
.../processors/query/GridQueryProcessor.java | 129 ++-
.../messages/GridQueryNextPageResponse.java | 35 +-
.../h2/twostep/messages/GridQueryRequest.java | 111 ++-
.../rest/client/message/GridRouterRequest.java | 18 +
.../rest/client/message/GridRouterResponse.java | 18 +
.../rest/protocols/tcp/GridTcpRestProtocol.java | 3 +-
.../service/GridServiceProcessor.java | 2 +-
.../processors/task/GridTaskProcessor.java | 23 +-
.../internal/util/GridConfigurationFinder.java | 55 +-
.../apache/ignite/internal/util/GridDebug.java | 48 +-
.../ignite/internal/util/GridJavaProcess.java | 2 +-
.../ignite/internal/util/IgniteUtils.java | 34 +-
.../shmem/IpcSharedMemoryClientEndpoint.java | 2 +-
.../ipc/shmem/IpcSharedMemoryNativeLoader.java | 151 +++-
.../shmem/IpcSharedMemoryServerEndpoint.java | 2 +-
.../util/ipc/shmem/IpcSharedMemoryUtils.java | 4 +-
.../ignite/internal/util/nio/GridNioServer.java | 64 +-
.../util/nio/GridShmemCommunicationClient.java | 146 ++++
.../util/spring/IgniteSpringHelper.java | 10 +
.../internal/visor/VisorMultiNodeTask.java | 2 +-
.../ignite/marshaller/MarshallerContext.java | 8 +
.../org/apache/ignite/plugin/PluginContext.java | 6 +
.../SpringApplicationContextResource.java | 4 +-
.../apache/ignite/resources/SpringResource.java | 6 +-
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 35 +-
.../org/apache/ignite/spi/IgniteSpiContext.java | 9 +-
.../communication/tcp/TcpCommunicationSpi.java | 561 +++++++++++--
.../tcp/TcpCommunicationSpiMBean.java | 27 +
.../ignite/spi/discovery/DiscoverySpi.java | 3 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 562 ++++++++-----
.../ignite/spi/discovery/tcp/ServerImpl.java | 275 ++++---
.../spi/discovery/tcp/TcpDiscoveryImpl.java | 69 +-
.../spi/discovery/tcp/TcpDiscoverySpi.java | 48 +-
.../tcp/internal/TcpDiscoveryNode.java | 18 +
.../ipfinder/TcpDiscoveryIpFinderAdapter.java | 34 +-
.../TcpDiscoveryMulticastIpFinder.java | 19 +-
.../messages/TcpDiscoveryAbstractMessage.java | 10 +-
.../messages/TcpDiscoveryNodeFailedMessage.java | 18 +
.../java/org/jsr166/ConcurrentHashMap8.java | 8 +-
.../java/org/jsr166/ConcurrentLinkedDeque8.java | 586 +++++---------
.../src/main/java/org/jsr166/LongAdder8.java | 35 +-
.../core/src/main/java/org/jsr166/README.txt | 11 +
.../src/main/java/org/jsr166/Striped64_8.java | 22 +-
.../java/org/jsr166/ThreadLocalRandom8.java | 19 +-
.../src/main/java/org/jsr166/package-info.java | 12 +-
.../core/src/main/resources/ignite.properties | 2 +-
modules/core/src/test/config/tests.properties | 2 +-
.../ignite/GridSuppressedExceptionSelfTest.java | 4 +-
.../internal/ClusterForHostsSelfTest.java | 113 +++
.../internal/ClusterGroupAbstractTest.java | 777 ++++++++++++++++++
.../ignite/internal/ClusterGroupSelfTest.java | 273 +++++++
.../internal/GridDiscoveryEventSelfTest.java | 6 +-
...ridFailFastNodeFailureDetectionSelfTest.java | 15 +-
.../GridFailoverTaskWithPredicateSelfTest.java | 3 -
.../GridJobMasterLeaveAwareSelfTest.java | 2 -
.../internal/GridJobStealingSelfTest.java | 3 -
.../internal/GridProjectionAbstractTest.java | 784 -------------------
...ectionLocalJobMultipleArgumentsSelfTest.java | 2 -
.../ignite/internal/GridProjectionSelfTest.java | 273 -------
.../apache/ignite/internal/GridSelfTest.java | 30 +-
.../GridTaskExecutionContextSelfTest.java | 9 -
.../GridTaskFailoverAffinityRunTest.java | 170 ++++
.../IgniteComputeEmptyClusterGroupTest.java | 3 -
.../IgniteComputeTopologyExceptionTest.java | 9 -
.../IgniteSlowClientDetectionSelfTest.java | 187 +++++
.../GridDiscoveryManagerAliveCacheSelfTest.java | 22 +-
.../cache/CacheClientStoreSelfTest.java | 228 ++++++
...acheReadOnlyTransactionalClientSelfTest.java | 327 --------
.../CacheReadThroughAtomicRestartSelfTest.java | 32 +
...heReadThroughLocalAtomicRestartSelfTest.java | 32 +
.../CacheReadThroughLocalRestartSelfTest.java | 32 +
...dThroughReplicatedAtomicRestartSelfTest.java | 32 +
...cheReadThroughReplicatedRestartSelfTest.java | 32 +
.../cache/CacheReadThroughRestartSelfTest.java | 133 ++++
.../CacheStoreUsageMultinodeAbstractTest.java | 305 ++++++++
...eUsageMultinodeDynamicStartAbstractTest.java | 169 ++++
...oreUsageMultinodeDynamicStartAtomicTest.java | 32 +
...heStoreUsageMultinodeDynamicStartTxTest.java | 32 +
...reUsageMultinodeStaticStartAbstractTest.java | 158 ++++
...toreUsageMultinodeStaticStartAtomicTest.java | 32 +
...cheStoreUsageMultinodeStaticStartTxTest.java | 32 +
.../GridCacheAbstractFailoverSelfTest.java | 8 +-
.../cache/GridCacheAbstractFullApiSelfTest.java | 39 +-
.../cache/GridCacheAbstractSelfTest.java | 5 +-
.../cache/GridCacheAffinityRoutingSelfTest.java | 4 +-
.../cache/GridCacheDaemonNodeStopSelfTest.java | 119 +++
.../cache/GridCacheDeploymentSelfTest.java | 3 -
.../cache/GridCacheEntryMemorySizeSelfTest.java | 91 ++-
.../cache/GridCacheMemoryModeSelfTest.java | 2 -
...ridCacheMultinodeUpdateAbstractSelfTest.java | 9 +
...inodeUpdateNearEnabledNoBackupsSelfTest.java | 2 +-
...CacheMultinodeUpdateNearEnabledSelfTest.java | 2 +-
.../processors/cache/GridCacheOffHeapTest.java | 28 +-
.../GridCacheReferenceCleanupSelfTest.java | 3 -
.../processors/cache/GridCacheStopSelfTest.java | 5 +
.../cache/GridCacheVersionMultinodeTest.java | 6 +-
.../IgniteCacheAbstractStopBusySelfTest.java | 30 +-
.../cache/IgniteCacheAbstractTest.java | 3 -
.../IgniteCacheAtomicStopBusySelfTest.java | 8 +-
.../IgniteCacheEntryListenerAbstractTest.java | 14 +-
.../IgniteCacheInterceptorSelfTestSuite.java | 2 +-
.../cache/IgniteCacheInvokeReadThroughTest.java | 5 +
...CacheP2pUnmarshallingRebalanceErrorTest.java | 15 +-
.../IgniteCacheP2pUnmarshallingTxErrorTest.java | 25 +-
...gniteCacheTransactionalStopBusySelfTest.java | 13 +-
...eDynamicCacheStartNoExchangeTimeoutTest.java | 466 +++++++++++
.../cache/IgniteDynamicCacheStartSelfTest.java | 56 ++
...teStartCacheInTransactionAtomicSelfTest.java | 32 +
.../IgniteStartCacheInTransactionSelfTest.java | 254 ++++++
.../IgniteTxMultiThreadedAbstractTest.java | 4 +-
...cheAtomicReferenceMultiNodeAbstractTest.java | 11 -
...GridCacheQueueMultiNodeAbstractSelfTest.java | 6 +-
...dCacheQueueMultiNodeConsistencySelfTest.java | 5 +
...CacheQueueRotativeMultiNodeAbstractTest.java | 10 -
.../GridCacheSetAbstractSelfTest.java | 31 +-
.../IgniteDataStructureWithJobTest.java | 111 +++
...omicOffheapQueueCreateMultiNodeSelfTest.java | 5 +
...ionedAtomicQueueCreateMultiNodeSelfTest.java | 5 +
...rtitionedDataStructuresFailoverSelfTest.java | 5 +
...edOffheapDataStructuresFailoverSelfTest.java | 5 +
...PartitionedQueueCreateMultiNodeSelfTest.java | 5 +
...dCachePartitionedQueueEntryMoveSelfTest.java | 5 +
...nedQueueFailoverDataConsistencySelfTest.java | 5 +
...eplicatedDataStructuresFailoverSelfTest.java | 5 +
...CacheLoadingConcurrentGridStartSelfTest.java | 5 +
.../GridCacheAbstractJobExecutionTest.java | 3 -
...ridCachePartitionNotLoadedEventSelfTest.java | 82 ++
.../GridCachePreloadLifecycleAbstractTest.java | 2 -
.../distributed/IgniteCache150ClientsTest.java | 189 +++++
.../IgniteCacheClientNodeConcurrentStart.java | 14 +-
...teCacheClientNodePartitionsExchangeTest.java | 1 +
.../distributed/IgniteCacheManyClientsTest.java | 318 ++++++++
.../IgniteCacheMessageRecoveryAbstractTest.java | 1 +
.../IgniteCacheTxMessageRecoveryTest.java | 5 +
...heAbstractTransformWriteThroughSelfTest.java | 3 -
.../GridCacheColocatedTxExceptionSelfTest.java | 5 +
...ePartitionedNearDisabledMetricsSelfTest.java | 4 +-
...dCachePartitionedTopologyChangeSelfTest.java | 5 +
.../near/GridCacheNearEvictionSelfTest.java | 3 -
...idCacheNearOnlyMultiNodeFullApiSelfTest.java | 5 -
.../near/GridCacheNearTxExceptionSelfTest.java | 5 +
...PartitionedFullApiMultithreadedSelfTest.java | 5 +
...idCachePartitionedHitsAndMissesSelfTest.java | 3 -
...achePartitionedMultiNodeFullApiSelfTest.java | 49 +-
.../GridCachePartitionedNodeRestartTest.java | 5 +
...ePartitionedOptimisticTxNodeRestartTest.java | 5 +
...achePartitionedPreloadLifecycleSelfTest.java | 2 +-
...CachePartitionedTxMultiThreadedSelfTest.java | 5 +
.../GridCachePartitionedTxSalvageSelfTest.java | 37 +-
.../GridCacheReplicatedFailoverSelfTest.java | 5 +
...eReplicatedFullApiMultithreadedSelfTest.java | 5 +
.../GridCacheReplicatedInvalidateSelfTest.java | 4 +-
...ridCacheReplicatedMultiNodeLockSelfTest.java | 5 +
.../GridCacheReplicatedMultiNodeSelfTest.java | 5 +
.../GridCacheReplicatedNodeRestartSelfTest.java | 5 +
.../GridCacheReplicatedTxExceptionSelfTest.java | 5 +
.../replicated/GridReplicatedTxPreloadTest.java | 2 +
...acheAtomicReplicatedNodeRestartSelfTest.java | 5 +
...CacheReplicatedPreloadLifecycleSelfTest.java | 6 +-
.../GridCacheEvictionFilterSelfTest.java | 4 +-
...cheSynchronousEvictionsFailoverSelfTest.java | 5 +
.../IgniteCacheExpiryPolicyAbstractTest.java | 10 +-
...eCacheExpiryPolicyWithStoreAbstractTest.java | 4 +-
.../IgniteCacheTxStoreSessionTest.java | 4 +
...dCacheLocalFullApiMultithreadedSelfTest.java | 5 +
.../GridCacheLocalTxExceptionSelfTest.java | 5 +
.../GridCacheSwapScanQueryAbstractSelfTest.java | 115 ++-
...ridCacheContinuousQueryAbstractSelfTest.java | 2 -
...CacheClientWriteBehindStoreAbstractTest.java | 104 +++
...teCacheClientWriteBehindStoreAtomicTest.java | 38 +
.../IgnteCacheClientWriteBehindStoreTxTest.java | 32 +
.../closure/GridClosureProcessorSelfTest.java | 29 +-
.../continuous/GridEventConsumeSelfTest.java | 2 -
.../DataStreamProcessorSelfTest.java | 47 +-
.../DataStreamerMultiThreadedSelfTest.java | 3 +
.../processors/igfs/IgfsModesSelfTest.java | 4 +-
.../internal/util/IgniteUtilsSelfTest.java | 22 +
.../ipc/shmem/IgfsSharedMemoryTestServer.java | 2 +
.../IpcSharedMemoryCrashDetectionSelfTest.java | 2 +-
.../ipc/shmem/IpcSharedMemorySpaceSelfTest.java | 2 +-
.../ipc/shmem/IpcSharedMemoryUtilsSelfTest.java | 2 +-
.../LoadWithCorruptedLibFileTestRunner.java | 2 +-
.../IpcSharedMemoryBenchmarkReader.java | 2 +-
.../IpcSharedMemoryBenchmarkWriter.java | 2 +-
.../internal/util/nio/GridNioSelfTest.java | 13 +-
.../internal/util/nio/GridNioSslSelfTest.java | 2 +
.../unsafe/GridUnsafeMemorySelfTest.java | 4 +-
.../tostring/GridToStringBuilderSelfTest.java | 4 +-
.../communication/GridIoManagerBenchmark0.java | 1 +
.../marshaller/MarshallerContextTestImpl.java | 29 +-
.../ignite/messaging/GridMessagingSelfTest.java | 3 -
.../GridP2PContinuousDeploymentSelfTest.java | 2 +
.../p2p/GridP2PLocalDeploymentSelfTest.java | 6 +-
.../p2p/GridP2PRemoteClassLoadersSelfTest.java | 31 +-
.../spi/GridTcpSpiForwardingSelfTest.java | 4 +-
.../GridTcpCommunicationSpiAbstractTest.java | 13 +
...mmunicationSpiConcurrentConnectSelfTest.java | 4 +-
.../GridTcpCommunicationSpiConfigSelfTest.java | 1 -
...cpCommunicationSpiMultithreadedSelfTest.java | 21 +-
...pCommunicationSpiMultithreadedShmemTest.java | 28 +
...dTcpCommunicationSpiRecoveryAckSelfTest.java | 1 +
...GridTcpCommunicationSpiRecoverySelfTest.java | 1 +
.../GridTcpCommunicationSpiShmemSelfTest.java | 38 +
.../tcp/GridTcpCommunicationSpiTcpSelfTest.java | 7 +
.../tcp/TcpClientDiscoverySpiSelfTest.java | 183 ++++-
.../spi/discovery/tcp/TcpDiscoverySelfTest.java | 44 +-
.../testframework/GridSpiTestContext.java | 7 +-
.../ignite/testframework/GridTestUtils.java | 14 +
.../config/GridTestProperties.java | 14 +-
.../testframework/junits/GridAbstractTest.java | 2 +-
.../junits/GridTestKernalContext.java | 3 +-
.../junits/IgniteTestResources.java | 16 +-
.../junits/common/GridCommonAbstractTest.java | 15 +-
.../ignite/testsuites/IgniteBasicTestSuite.java | 32 +-
.../IgniteCacheDataStructuresSelfTestSuite.java | 25 +-
.../IgniteCacheEvictionSelfTestSuite.java | 3 +-
.../IgniteCacheFailoverTestSuite.java | 24 +-
.../IgniteCacheFailoverTestSuite2.java | 47 ++
.../IgniteCacheFullApiSelfTestSuite.java | 8 +-
...niteCacheP2pUnmarshallingErrorTestSuite.java | 20 +-
.../testsuites/IgniteCacheRestartTestSuite.java | 10 +-
.../ignite/testsuites/IgniteCacheTestSuite.java | 48 +-
.../testsuites/IgniteCacheTestSuite2.java | 4 +-
.../testsuites/IgniteCacheTestSuite3.java | 15 +-
.../testsuites/IgniteCacheTestSuite4.java | 29 +-
.../IgniteCacheWriteBehindTestSuite.java | 2 +
.../testsuites/IgniteClientTestSuite.java | 38 +
.../testsuites/IgniteComputeGridTestSuite.java | 1 +
.../testsuites/IgniteKernalSelfTestSuite.java | 16 +-
.../IgniteMarshallerSelfTestSuite.java | 28 +-
.../IgniteSpiCommunicationSelfTestSuite.java | 2 +
.../testsuites/IgniteUtilSelfTestSuite.java | 18 +-
.../apache/ignite/util/GridRandomSelfTest.java | 4 +-
.../ignite/util/TestTcpCommunicationSpi.java | 21 +
modules/extdata/p2p/pom.xml | 2 +-
.../p2p/GridP2PContinuousDeploymentTask1.java | 2 +-
.../tests/p2p/P2PTestTaskExternalPath1.java | 10 +-
.../tests/p2p/P2PTestTaskExternalPath2.java | 8 +-
modules/extdata/uri/pom.xml | 2 +-
modules/gce/pom.xml | 2 +-
modules/geospatial/pom.xml | 2 +-
modules/hadoop/pom.xml | 81 +-
.../fs/IgniteHadoopFileSystemCounterWriter.java | 9 +-
.../processors/hadoop/HadoopClassLoader.java | 29 +
.../processors/hadoop/HadoopDefaultJobInfo.java | 27 +-
.../internal/processors/hadoop/HadoopUtils.java | 237 ------
.../hadoop/SecondaryFileSystemProvider.java | 3 +-
.../hadoop/fs/HadoopFileSystemCacheUtils.java | 241 ++++++
.../hadoop/fs/HadoopFileSystemsUtils.java | 11 +
.../hadoop/fs/HadoopLazyConcurrentMap.java | 5 +
.../hadoop/jobtracker/HadoopJobTracker.java | 25 +-
.../child/HadoopChildProcessRunner.java | 3 +-
.../processors/hadoop/v2/HadoopV2Job.java | 84 +-
.../hadoop/v2/HadoopV2JobResourceManager.java | 22 +-
.../hadoop/v2/HadoopV2TaskContext.java | 37 +-
.../HadoopIgfs20FileSystemAbstractSelfTest.java | 17 +-
...oopSecondaryFileSystemConfigurationTest.java | 14 +
.../apache/ignite/igfs/IgfsEventsTestSuite.java | 5 +-
.../IgniteHadoopFileSystemAbstractSelfTest.java | 2 +-
...IgniteHadoopFileSystemHandshakeSelfTest.java | 7 +
.../IgniteHadoopFileSystemIpcCacheSelfTest.java | 7 +
.../hadoop/HadoopAbstractSelfTest.java | 7 +
.../processors/hadoop/HadoopMapReduceTest.java | 23 +-
.../processors/hadoop/HadoopTasksV1Test.java | 7 +-
.../processors/hadoop/HadoopTasksV2Test.java | 7 +-
.../processors/hadoop/HadoopV2JobSelfTest.java | 6 +-
.../collections/HadoopAbstractMapTest.java | 3 +-
.../collections/HadoopHashMapSelfTest.java | 4 +-
.../HadoopExternalTaskExecutionSelfTest.java | 2 +
.../HadoopExternalCommunicationSelfTest.java | 5 +
.../testsuites/IgniteHadoopTestSuite.java | 9 +-
.../IgniteIgfsLinuxAndMacOSTestSuite.java | 3 +-
modules/hibernate/pom.xml | 16 +-
.../hibernate/CacheHibernateBlobStore.java | 87 +-
.../CacheHibernateBlobStoreFactory.java | 235 ++++++
.../hibernate/src/test/config/factory-cache.xml | 59 ++
.../src/test/config/factory-cache1.xml | 61 ++
.../config/factory-incorrect-store-cache.xml | 56 ++
.../hibernate/HibernateL2CacheSelfTest.java | 5 +
.../HibernateL2CacheTransactionalSelfTest.java | 5 +
.../CacheHibernateStoreFactorySelfTest.java | 273 +++++++
.../testsuites/IgniteHibernateTestSuite.java | 6 +-
modules/indexing/pom.xml | 18 +-
.../processors/query/h2/IgniteH2Indexing.java | 125 ++-
.../query/h2/sql/GridSqlQuerySplitter.java | 49 +-
.../query/h2/twostep/GridMapQueryExecutor.java | 321 ++++++--
.../query/h2/twostep/GridMergeIndex.java | 17 +-
.../h2/twostep/GridMergeIndexUnsorted.java | 7 +-
.../h2/twostep/GridReduceQueryExecutor.java | 656 +++++++++++++---
.../query/h2/twostep/GridResultPage.java | 21 +-
.../CacheAbstractQueryMetricsSelfTest.java | 205 +++++
.../CachePartitionedQueryMetricsSelfTest.java | 32 +
.../CacheReplicatedQueryMetricsSelfTest.java | 32 +
...CacheScanPartitionQueryFallbackSelfTest.java | 408 ++++++++++
.../cache/GridCacheCrossCacheQuerySelfTest.java | 25 +-
.../cache/GridCacheOffHeapSelfTest.java | 1 -
.../cache/GridCacheQueryMetricsSelfTest.java | 206 -----
...idCacheReduceQueryMultithreadedSelfTest.java | 10 -
.../processors/cache/GridCacheSwapSelfTest.java | 3 -
.../IgniteCacheAbstractFieldsQuerySelfTest.java | 13 +-
.../cache/IgniteCacheAbstractQuerySelfTest.java | 79 +-
.../cache/IgniteCacheOffheapEvictQueryTest.java | 196 +++++
...hePartitionedQueryMultiThreadedSelfTest.java | 40 +-
.../IgniteCacheQueryMultiThreadedSelfTest.java | 2 -
...QueryOffheapEvictsMultiThreadedSelfTest.java | 5 +
...lientQueryReplicatedNodeRestartSelfTest.java | 419 ++++++++++
.../IgniteCacheQueryNodeRestartSelfTest.java | 41 +-
.../IgniteCacheQueryNodeRestartSelfTest2.java | 383 +++++++++
...dCacheAbstractReduceFieldsQuerySelfTest.java | 1 -
.../h2/GridIndexingSpiAbstractSelfTest.java | 4 +-
.../query/h2/sql/BaseH2CompareQueryTest.java | 6 +-
.../query/h2/sql/GridQueryParsingTest.java | 5 +-
.../IgniteCacheQuerySelfTestSuite.java | 13 +-
modules/jcl/pom.xml | 2 +-
modules/jta/pom.xml | 2 +-
.../cache/jta/GridCacheXAResource.java | 18 +-
.../processors/cache/GridCacheJtaSelfTest.java | 2 +-
modules/log4j/pom.xml | 2 +-
modules/mesos/README.txt | 2 +-
modules/mesos/pom.xml | 2 +-
.../apache/ignite/mesos/ClusterProperties.java | 15 +
.../apache/ignite/mesos/IgniteScheduler.java | 10 +-
modules/rest-http/pom.xml | 2 +-
modules/scalar-2.10/README.txt | 4 +
modules/scalar-2.10/licenses/apache-2.0.txt | 202 +++++
.../scalar-2.10/licenses/scala-bsd-license.txt | 18 +
modules/scalar-2.10/pom.xml | 197 +++++
modules/scalar/pom.xml | 2 +-
modules/schedule/pom.xml | 2 +-
modules/schema-import/pom.xml | 2 +-
modules/slf4j/pom.xml | 2 +-
modules/spark-2.10/README.txt | 4 +
modules/spark-2.10/licenses/apache-2.0.txt | 202 +++++
.../spark-2.10/licenses/scala-bsd-license.txt | 18 +
modules/spark-2.10/pom.xml | 120 +++
modules/spark/README.txt | 8 +
modules/spark/licenses/apache-2.0.txt | 202 +++++
modules/spark/licenses/scala-bsd-license.txt | 18 +
modules/spark/pom.xml | 110 +++
.../org/apache/ignite/spark/IgniteContext.scala | 119 +++
.../org/apache/ignite/spark/IgniteRDD.scala | 244 ++++++
.../apache/ignite/spark/JavaIgniteContext.scala | 63 ++
.../org/apache/ignite/spark/JavaIgniteRDD.scala | 99 +++
.../ignite/spark/impl/IgniteAbstractRDD.scala | 39 +
.../ignite/spark/impl/IgnitePartition.scala | 24 +
.../ignite/spark/impl/IgniteQueryIterator.scala | 27 +
.../apache/ignite/spark/impl/IgniteSqlRDD.scala | 41 +
.../spark/impl/JavaIgniteAbstractRDD.scala | 34 +
.../ignite/spark/JavaIgniteRDDSelfTest.java | 298 +++++++
.../scala/org/apache/ignite/spark/Entity.scala | 28 +
.../org/apache/ignite/spark/IgniteRddSpec.scala | 231 ++++++
modules/spring/pom.xml | 9 +-
.../GridResourceSpringBeanInjector.java | 2 +-
.../util/spring/IgniteSpringHelperImpl.java | 17 +
.../src/test/config/incorrect-store-cache.xml | 57 ++
modules/spring/src/test/config/node.xml | 43 +
modules/spring/src/test/config/node1.xml | 45 ++
.../test/config/pojo-incorrect-store-cache.xml | 56 ++
modules/spring/src/test/config/store-cache.xml | 59 ++
modules/spring/src/test/config/store-cache1.xml | 62 ++
.../jdbc/CacheJdbcBlobStoreFactorySelfTest.java | 172 ++++
.../jdbc/CacheJdbcPojoStoreFactorySelfTest.java | 193 +++++
.../testsuites/IgniteSpringTestSuite.java | 5 +
modules/ssh/pom.xml | 2 +-
modules/tools/pom.xml | 2 +-
.../ignite/tools/classgen/ClassesGenerator.java | 30 +-
modules/urideploy/pom.xml | 2 +-
modules/visor-console-2.10/README.txt | 4 +
modules/visor-console-2.10/pom.xml | 174 ++++
modules/visor-console/pom.xml | 2 +-
.../commands/cache/VisorCacheCommand.scala | 7 +-
modules/visor-plugins/pom.xml | 2 +-
modules/web/pom.xml | 2 +-
.../IgniteWebSessionSelfTestSuite.java | 2 +-
modules/yardstick/pom.xml | 2 +-
parent/pom.xml | 5 +
pom.xml | 22 +-
scripts/git-patch-prop.sh | 2 +-
495 files changed, 18532 insertions(+), 4895 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a8232be6/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a8232be6/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a8232be6/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a8232be6/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
index 0000000,ceb9bef..37a6e72
mode 000000,100644..100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
@@@ -1,0 -1,251 +1,273 @@@
+ /*
+ * 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 org.apache.ignite.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.marshaller.*;
+ import org.apache.ignite.testframework.junits.common.*;
+
+ import java.util.*;
+
+ /**
+ * Test for {@link ClusterGroup}.
+ */
+ @GridCommonTest(group = "Kernal Self")
+ public class ClusterGroupSelfTest extends ClusterGroupAbstractTest {
+ /** Nodes count. */
+ private static final int NODES_CNT = 4;
+
+ /** Projection node IDs. */
+ private static Collection<UUID> ids;
+
+ /** */
+ private static Ignite ignite;
+
+ /** {@inheritDoc} */
+ @SuppressWarnings({"ConstantConditions"})
+ @Override protected void beforeTestsStarted() throws Exception {
+ assert NODES_CNT > 2;
+
+ ids = new LinkedList<>();
+
+ try {
+ for (int i = 0; i < NODES_CNT; i++) {
+ Ignition.setClientMode(i > 1);
+
+ Ignite g = startGrid(i);
+
+ ids.add(g.cluster().localNode().id());
+
+ if (i == 0)
+ ignite = g;
+ }
+ }
+ finally {
+ Ignition.setClientMode(false);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ for (int i = 0; i < NODES_CNT; i++)
+ stopGrid(i);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected ClusterGroup projection() {
+ return grid(0).cluster().forPredicate(F.nodeForNodeIds(ids));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected UUID localNodeId() {
+ return grid(0).localNode().id();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testRandom() throws Exception {
+ assertTrue(ignite.cluster().nodes().contains(ignite.cluster().forRandom().node()));
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testOldest() throws Exception {
+ ClusterGroup oldest = ignite.cluster().forOldest();
+
+ ClusterNode node = null;
+
+ long minOrder = Long.MAX_VALUE;
+
+ for (ClusterNode n : ignite.cluster().nodes()) {
+ if (n.order() < minOrder) {
+ node = n;
+
+ minOrder = n.order();
+ }
+ }
+
+ assertEquals(oldest.node(), ignite.cluster().forNode(node).node());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testYoungest() throws Exception {
+ ClusterGroup youngest = ignite.cluster().forYoungest();
+
+ ClusterNode node = null;
+
+ long maxOrder = Long.MIN_VALUE;
+
+ for (ClusterNode n : ignite.cluster().nodes()) {
+ if (n.order() > maxOrder) {
+ node = n;
+
+ maxOrder = n.order();
+ }
+ }
+
+ assertEquals(youngest.node(), ignite.cluster().forNode(node).node());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNewNodes() throws Exception {
+ ClusterGroup youngest = ignite.cluster().forYoungest();
+ ClusterGroup oldest = ignite.cluster().forOldest();
+
+ ClusterNode old = oldest.node();
+ ClusterNode last = youngest.node();
+
+ assertNotNull(last);
+
+ try (Ignite g = startGrid(NODES_CNT)) {
+ ClusterNode n = g.cluster().localNode();
+
+ ClusterNode latest = youngest.node();
+
+ assertNotNull(latest);
+ assertEquals(latest.id(), n.id());
+ assertEquals(oldest.node(), old);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testForPredicate() throws Exception {
+ IgnitePredicate<ClusterNode> evenP = new IgnitePredicate<ClusterNode>() {
+ @Override public boolean apply(ClusterNode node) {
+ return node.order() % 2 == 0;
+ }
+ };
+
+ IgnitePredicate<ClusterNode> oddP = new IgnitePredicate<ClusterNode>() {
+ @Override public boolean apply(ClusterNode node) {
+ return node.order() % 2 == 1;
+ }
+ };
+
+ ClusterGroup remotes = ignite.cluster().forRemotes();
+
+ ClusterGroup evenYoungest = remotes.forPredicate(evenP).forYoungest();
+ ClusterGroup evenOldest = remotes.forPredicate(evenP).forOldest();
+
+ ClusterGroup oddYoungest = remotes.forPredicate(oddP).forYoungest();
+ ClusterGroup oddOldest = remotes.forPredicate(oddP).forOldest();
+
+ int clusterSize = ignite.cluster().nodes().size();
+
+ assertEquals(grid(gridMaxOrder(clusterSize, true)).localNode().id(), evenYoungest.node().id());
+ assertEquals(grid(1).localNode().id(), evenOldest.node().id());
+
+ assertEquals(grid(gridMaxOrder(clusterSize, false)).localNode().id(), oddYoungest.node().id());
+ assertEquals(grid(2).localNode().id(), oddOldest.node().id());
+
+ try (Ignite g4 = startGrid(NODES_CNT);
+ Ignite g5 = startGrid(NODES_CNT + 1))
+ {
+ clusterSize = g4.cluster().nodes().size();
+
+ assertEquals(grid(gridMaxOrder(clusterSize, true)).localNode().id(), evenYoungest.node().id());
+ assertEquals(grid(1).localNode().id(), evenOldest.node().id());
+
+ assertEquals(grid(gridMaxOrder(clusterSize, false)).localNode().id(), oddYoungest.node().id());
+ assertEquals(grid(2).localNode().id(), oddOldest.node().id());
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAgeClusterGroupSerialization() throws Exception {
+ Marshaller marshaller = getConfiguration().getMarshaller();
+
+ ClusterGroup grp = ignite.cluster().forYoungest();
+ ClusterNode node = grp.node();
+
+ byte[] arr = marshaller.marshal(grp);
+
+ ClusterGroup obj = marshaller.unmarshal(arr, null);
+
+ assertEquals(node.id(), obj.node().id());
+
+ try (Ignite ignore = startGrid()) {
+ obj = marshaller.unmarshal(arr, null);
+
+ assertEquals(grp.node().id(), obj.node().id());
+ assertFalse(node.id().equals(obj.node().id()));
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientServer() throws Exception {
+ ClusterGroup srv = ignite.cluster().forServers();
+
+ assertEquals(2, srv.nodes().size());
+ assertTrue(srv.nodes().contains(ignite(0).cluster().localNode()));
+ assertTrue(srv.nodes().contains(ignite(1).cluster().localNode()));
+
+ ClusterGroup cli = ignite.cluster().forClients();
+
+ assertEquals(2, srv.nodes().size());
+ assertTrue(cli.nodes().contains(ignite(2).cluster().localNode()));
+ assertTrue(cli.nodes().contains(ignite(3).cluster().localNode()));
+ }
+
+ /**
+ * @param cnt Count.
+ * @param even Even.
+ */
+ private static int gridMaxOrder(int cnt, boolean even) {
+ assert cnt > 2;
+
+ cnt = cnt - (cnt % 2);
+
+ return even ? cnt - 1 : cnt - 2;
+ }
++
++ /**
++ * @throws Exception If failed.
++ */
++ public void testHostNames() throws Exception {
++ Collection<String> inputHostNames = ignite.cluster().hostNames();
++ Collection<String> localNodeHostNames = ignite.cluster().localNode().hostNames();
++ Collection<String> randomNodeHostNames = ignite.cluster().forRandom().node().hostNames();
++ Collection<ClusterNode> allNodes = ignite.cluster().nodes();
++ Collection<String> checkHostNames = new HashSet<String> ();
++
++ for (ClusterNode currentNode : allNodes)
++ Collections.addAll(checkHostNames, currentNode.hostNames().toArray(new String[0]));
++
++ assert(checkHostNames.equals(inputHostNames));
++
++ if (!(localNodeHostNames.isEmpty()) && !(inputHostNames.isEmpty()))
++ assert((inputHostNames.containsAll(localNodeHostNames)) == true);
++
++ if (!(randomNodeHostNames.isEmpty()) && !(inputHostNames.isEmpty()))
++ assert((inputHostNames.containsAll(randomNodeHostNames)) == true);
++ }
+ }
[08/20] incubator-ignite git commit: # ignite-sprint-7 enabled test
Posted by sb...@apache.org.
# ignite-sprint-7 enabled test
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/01d842a5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/01d842a5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/01d842a5
Branch: refs/heads/ignite-428
Commit: 01d842a5110587cf6df4a17217a0674e9bed391f
Parents: 3e8ddb4
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jun 26 12:29:17 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jun 26 12:29:17 2015 +0300
----------------------------------------------------------------------
.../ignite/internal/GridDiscoveryEventSelfTest.java | 12 +++++-------
1 file changed, 5 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/01d842a5/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
index 3981b7f..d820dba 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
@@ -98,7 +98,7 @@ public class GridDiscoveryEventSelfTest extends GridCommonAbstractTest {
private AtomicInteger cnt = new AtomicInteger();
@Override public boolean apply(Event evt) {
- assert evt.type() == EVT_NODE_JOINED;
+ assert evt.type() == EVT_NODE_JOINED : evt;
evts.put(cnt.getAndIncrement(), ((DiscoveryEvent)evt).topologyNodes());
@@ -147,8 +147,6 @@ public class GridDiscoveryEventSelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testLeaveSequenceEvents() throws Exception {
- fail("https://issues.apache.org/jira/browse/IGNITE-932");
-
try {
Ignite g0 = startGrid(0);
@@ -165,6 +163,8 @@ public class GridDiscoveryEventSelfTest extends GridCommonAbstractTest {
private AtomicInteger cnt = new AtomicInteger();
@Override public boolean apply(Event evt) {
+ assert evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED : evt;
+
evts.put(cnt.getAndIncrement(), ((DiscoveryEvent) evt).topologyNodes());
latch.countDown();
@@ -215,8 +215,6 @@ public class GridDiscoveryEventSelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testMixedSequenceEvents() throws Exception {
- fail("https://issues.apache.org/jira/browse/IGNITE-932");
-
try {
Ignite g0 = startGrid(0);
@@ -231,7 +229,7 @@ public class GridDiscoveryEventSelfTest extends GridCommonAbstractTest {
@Override public boolean apply(Event evt) {
assert evt.type() == EVT_NODE_JOINED
- || evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED;
+ || evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED : evt;
evts.put(cnt.getAndIncrement(), ((DiscoveryEvent) evt).topologyNodes());
@@ -347,7 +345,7 @@ public class GridDiscoveryEventSelfTest extends GridCommonAbstractTest {
private AtomicInteger cnt = new AtomicInteger();
@Override public boolean apply(Event evt) {
- assert evt.type() == EVT_NODE_JOINED;
+ assert evt.type() == EVT_NODE_JOINED : evt;
X.println(">>>>>>> Joined " + F.viewReadOnly(((DiscoveryEvent) evt).topologyNodes(),
NODE_2ID));
[15/20] incubator-ignite git commit: Merge remote-tracking branch
'origin/ignite-sprint-7' into ignite-sprint-7
Posted by sb...@apache.org.
Merge remote-tracking branch 'origin/ignite-sprint-7' into ignite-sprint-7
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/70238a62
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/70238a62
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/70238a62
Branch: refs/heads/ignite-428
Commit: 70238a6254388fa3889d5cf31b5bda5dac11a100
Parents: 5fb9f2f 50a4626
Author: ashutak <as...@gridgain.com>
Authored: Fri Jun 26 17:55:31 2015 +0300
Committer: ashutak <as...@gridgain.com>
Committed: Fri Jun 26 17:55:31 2015 +0300
----------------------------------------------------------------------
.../processors/rest/GridRestProcessor.java | 4 ++-
.../handlers/task/GridTaskCommandHandler.java | 12 ++++---
.../processors/task/GridTaskWorker.java | 4 ++-
.../visor/query/VisorQueryCleanupTask.java | 14 +++++++++
.../util/VisorClusterGroupEmptyException.java | 33 ++++++++++++++++++++
5 files changed, 61 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
[16/20] incubator-ignite git commit: # ignite-1006: review
Posted by sb...@apache.org.
# ignite-1006: review
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/8fd909d9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/8fd909d9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/8fd909d9
Branch: refs/heads/ignite-428
Commit: 8fd909d98505fe3964bcf756e8bec42888c1164c
Parents: a8232be
Author: ashutak <as...@gridgain.com>
Authored: Fri Jun 26 18:01:53 2015 +0300
Committer: ashutak <as...@gridgain.com>
Committed: Fri Jun 26 18:01:53 2015 +0300
----------------------------------------------------------------------
.../internal/cluster/ClusterGroupAdapter.java | 12 +-
.../internal/ClusterForHostsSelfTest.java | 113 ---------------
.../internal/ClusterGroupHostsSelfTest.java | 141 +++++++++++++++++++
.../ignite/internal/ClusterGroupSelfTest.java | 22 ---
.../ignite/testsuites/IgniteBasicTestSuite.java | 2 +-
5 files changed, 146 insertions(+), 144 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8fd909d9/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
index b770d98..4b61116 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
@@ -296,16 +296,12 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
/** {@inheritDoc} */
@Override public Collection<String> hostNames() {
- Collection<String> resultHostNames = new HashSet<String> ();
- Collection<ClusterNode> allNodes = nodes();
+ Set<String> res = new HashSet<>();
- if (!(allNodes.isEmpty()))
- {
- for (ClusterNode currentNode : allNodes)
- Collections.addAll(resultHostNames, currentNode.hostNames().toArray(new String[0]));
- }
+ for (ClusterNode node : nodes())
+ res.addAll(node.hostNames());
- return resultHostNames;
+ return Collections.unmodifiableSet(res);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8fd909d9/modules/core/src/test/java/org/apache/ignite/internal/ClusterForHostsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/ClusterForHostsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/ClusterForHostsSelfTest.java
deleted file mode 100644
index 59c3db9..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/ClusterForHostsSelfTest.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cluster.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.spi.discovery.tcp.*;
-import org.apache.ignite.testframework.junits.common.*;
-
-import java.lang.reflect.*;
-import java.util.*;
-
-/**
- * Test for {@link ClusterGroup#forHost(String, String...)}.
- *
- * @see GridProjectionSelfTest
- */
-@GridCommonTest(group = "Kernal Self")
-public class ClusterForHostsSelfTest extends GridCommonAbstractTest {
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
- Collection<String> hostNames = null;
-
- if ("forHostTest".equals(gridName))
- hostNames = Arrays.asList("h_1", "h_2", "h_3");
-
- IgniteConfiguration cfg = super.getConfiguration(gridName);
-
- if (hostNames != null) {
- TcpDiscoverySpi disco = (TcpDiscoverySpi)cfg.getDiscoverySpi();
-
- cfg.setDiscoverySpi(new CustomHostsTcpDiscoverySpi(hostNames).setIpFinder(disco.getIpFinder()));
- }
-
- return cfg;
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testForHosts() throws Exception {
- Ignite ignite = startGrid("forHostTest");
-
- assertEquals(1, ignite.cluster().forHost("h_1").nodes().size());
- assertEquals(1, ignite.cluster().forHost("h_1", "h_3").nodes().size());
- assertEquals(1, ignite.cluster().forHost("unknown_host", "h_2").nodes().size());
- assertEquals(1, ignite.cluster().forHost("h_1", "h_3", "unknown_host", "h_2").nodes().size());
-
- assertEquals(0, ignite.cluster().forHost("unknown_host").nodes().size());
-
- boolean gotNpe = false;
-
- try {
- assertEquals(0, ignite.cluster().forHost(null, null, null).nodes().size());
- }
- catch (NullPointerException e) {
- gotNpe = true;
- }
-
- assertTrue(gotNpe);
- }
-
- /**
- * Tcp discovery spi that allow to customise hostNames of created local node.
- */
- private static class CustomHostsTcpDiscoverySpi extends TcpDiscoverySpi {
- /** Hosts. */
- private final Collection<String> hosts;
-
- /**
- * @param hosts Host names which will be retuned by {@link ClusterNode#hostNames()} of created local node.
- */
- CustomHostsTcpDiscoverySpi(Collection<String> hosts) {
- this.hosts = hosts;
- }
-
- /**
- * @param srvPort Server port.
- */
- @Override protected void initLocalNode(int srvPort, boolean addExtAddrAttr) {
- super.initLocalNode(srvPort, addExtAddrAttr);
-
- try {
- Field hostNamesField = locNode.getClass().getDeclaredField("hostNames");
-
- hostNamesField.setAccessible(true);
-
- hostNamesField.set(locNode, hosts);
- }
- catch (IllegalAccessException | NoSuchFieldException e) {
- U.error(log, "Looks like implementation of " + locNode.getClass()
- + " class was changed. Need to update test.", e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8fd909d9/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupHostsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupHostsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupHostsSelfTest.java
new file mode 100644
index 0000000..297a590
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupHostsSelfTest.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.internal;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import java.lang.reflect.*;
+import java.util.*;
+
+/**
+ * Test for {@link ClusterGroup#forHost(String, String...)}.
+ *
+ * @see ClusterGroupSelfTest
+ */
+@GridCommonTest(group = "Kernal Self")
+public class ClusterGroupHostsSelfTest extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGrid();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ Collection<String> hostNames = Arrays.asList("h_1", "h_2", "h_3");
+
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TcpDiscoverySpi disco = (TcpDiscoverySpi)cfg.getDiscoverySpi();
+
+ cfg.setDiscoverySpi(new CustomHostsTcpDiscoverySpi(hostNames).setIpFinder(disco.getIpFinder()));
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testForHosts() throws Exception {
+ Ignite ignite = grid();
+
+ assertEquals(1, ignite.cluster().forHost("h_1").nodes().size());
+ assertEquals(1, ignite.cluster().forHost("h_1", "h_3").nodes().size());
+ assertEquals(1, ignite.cluster().forHost("unknown_host", "h_2").nodes().size());
+ assertEquals(1, ignite.cluster().forHost("h_1", "h_3", "unknown_host", "h_2").nodes().size());
+
+ assertEquals(0, ignite.cluster().forHost("unknown_host").nodes().size());
+
+ boolean gotNpe = false;
+
+ try {
+ assertEquals(0, ignite.cluster().forHost(null, null, null).nodes().size());
+ }
+ catch (NullPointerException e) {
+ gotNpe = true;
+ }
+ finally {
+ assertTrue(gotNpe);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testHostNames() throws Exception {
+ Ignite ignite = grid();
+
+ Collection<String> locNodeHosts = ignite.cluster().localNode().hostNames();
+ Collection<String> clusterHosts = ignite.cluster().hostNames();
+
+ assertTrue(F.eqNotOrdered(locNodeHosts, clusterHosts));
+
+ boolean gotNpe = false;
+
+ try {
+ clusterHosts.add("valueShouldNotToBeAdded");
+ }
+ catch (UnsupportedOperationException e) {
+ gotNpe = true;
+ }
+ finally {
+ assertTrue(gotNpe);
+ }
+ }
+
+ /**
+ * Tcp discovery spi that allow to customise hostNames of created local node.
+ */
+ private static class CustomHostsTcpDiscoverySpi extends TcpDiscoverySpi {
+ /** Hosts. */
+ private final Collection<String> hosts;
+
+ /**
+ * @param hosts Host names which will be retuned by {@link ClusterNode#hostNames()} of created local node.
+ */
+ CustomHostsTcpDiscoverySpi(Collection<String> hosts) {
+ this.hosts = hosts;
+ }
+
+ /**
+ * @param srvPort Server port.
+ */
+ @Override protected void initLocalNode(int srvPort, boolean addExtAddrAttr) {
+ super.initLocalNode(srvPort, addExtAddrAttr);
+
+ try {
+ Field hostNamesField = locNode.getClass().getDeclaredField("hostNames");
+
+ hostNamesField.setAccessible(true);
+
+ hostNamesField.set(locNode, hosts);
+ }
+ catch (IllegalAccessException | NoSuchFieldException e) {
+ U.error(log, "Looks like implementation of " + locNode.getClass()
+ + " class was changed. Need to update test.", e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8fd909d9/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
index 37a6e72..ceb9bef 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/ClusterGroupSelfTest.java
@@ -248,26 +248,4 @@ public class ClusterGroupSelfTest extends ClusterGroupAbstractTest {
return even ? cnt - 1 : cnt - 2;
}
-
- /**
- * @throws Exception If failed.
- */
- public void testHostNames() throws Exception {
- Collection<String> inputHostNames = ignite.cluster().hostNames();
- Collection<String> localNodeHostNames = ignite.cluster().localNode().hostNames();
- Collection<String> randomNodeHostNames = ignite.cluster().forRandom().node().hostNames();
- Collection<ClusterNode> allNodes = ignite.cluster().nodes();
- Collection<String> checkHostNames = new HashSet<String> ();
-
- for (ClusterNode currentNode : allNodes)
- Collections.addAll(checkHostNames, currentNode.hostNames().toArray(new String[0]));
-
- assert(checkHostNames.equals(inputHostNames));
-
- if (!(localNodeHostNames.isEmpty()) && !(inputHostNames.isEmpty()))
- assert((inputHostNames.containsAll(localNodeHostNames)) == true);
-
- if (!(randomNodeHostNames.isEmpty()) && !(inputHostNames.isEmpty()))
- assert((inputHostNames.containsAll(randomNodeHostNames)) == true);
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8fd909d9/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
index 6ff83e2..19c1932 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicTestSuite.java
@@ -65,7 +65,7 @@ public class IgniteBasicTestSuite extends TestSuite {
suite.addTest(IgniteStreamSelfTestSuite.suite());
suite.addTest(new TestSuite(GridSelfTest.class));
- suite.addTest(new TestSuite(ClusterForHostsSelfTest.class));
+ suite.addTest(new TestSuite(ClusterGroupHostsSelfTest.class));
suite.addTest(new TestSuite(IgniteMessagingWithClientTest.class));
GridTestUtils.addTestIfNeeded(suite, ClusterGroupSelfTest.class, ignoredTests);
[11/20] incubator-ignite git commit: Merge remote-tracking branch
'remotes/origin/ignite-883' into ignite-sprint-7
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/ignite-883' into ignite-sprint-7
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/6e23608f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/6e23608f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/6e23608f
Branch: refs/heads/ignite-428
Commit: 6e23608fd0845fb30c470fbb2f0143c7d5d67f82
Parents: 847ddf5 efa92c5
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jun 26 16:21:22 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jun 26 16:21:22 2015 +0300
----------------------------------------------------------------------
.../GridCachePartitionExchangeManager.java | 6 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 151 ++++++-----
.../ignite/spi/discovery/tcp/ServerImpl.java | 103 +++++--
.../spi/discovery/tcp/TcpDiscoverySpi.java | 3 +-
.../tcp/TcpClientDiscoverySpiSelfTest.java | 265 ++++++++++++++++++-
5 files changed, 448 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
[05/20] incubator-ignite git commit: # ignite-1006
Posted by sb...@apache.org.
# ignite-1006
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/c6f66c64
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/c6f66c64
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/c6f66c64
Branch: refs/heads/ignite-428
Commit: c6f66c64becfda669e381aade63ae59d946962c2
Parents: addc91b
Author: Atri <at...@gmail.com>
Authored: Thu Jun 25 14:14:23 2015 +0530
Committer: ashutak <as...@gridgain.com>
Committed: Thu Jun 25 21:11:55 2015 +0300
----------------------------------------------------------------------
.../org/apache/ignite/cluster/ClusterGroup.java | 7 +++++++
.../internal/cluster/ClusterGroupAdapter.java | 14 +++++++++++++
.../cluster/IgniteClusterAsyncImpl.java | 5 +++++
.../ignite/internal/GridProjectionSelfTest.java | 22 ++++++++++++++++++++
4 files changed, 48 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6f66c64/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java b/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
index 06854d4..e2831a7 100644
--- a/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
+++ b/modules/core/src/main/java/org/apache/ignite/cluster/ClusterGroup.java
@@ -261,6 +261,13 @@ public interface ClusterGroup {
public ClusterNode node();
/**
+ * Gets the read-only collection of hostnames in this cluster group.
+ *
+ * @return All hostnames in this cluster group.
+ */
+ public Collection<String> hostNames();
+
+ /**
* Gets predicate that defines a subset of nodes for this cluster group.
*
* @return Predicate that defines a subset of nodes for this cluster group.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6f66c64/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
index bb82c3b..8c2fece 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java
@@ -295,6 +295,20 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
}
/** {@inheritDoc} */
+ @Override public Collection<String> hostNames() {
+ Collection<String> resultHostNames = new HashSet<String> ();
+ Collection<ClusterNode> allNodes = nodes();
+
+ if (!(allNodes.isEmpty()))
+ {
+ for (ClusterNode currentNode : allNodes)
+ Collections.addAll(resultHostNames, currentNode.hostNames().toArray(new String[0]));
+ }
+
+ return resultHostNames;
+ }
+
+ /** {@inheritDoc} */
@Override public final ClusterNode node(UUID id) {
A.notNull(id, "id");
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6f66c64/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
index 7f67b4f..8c6f4e0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterAsyncImpl.java
@@ -262,6 +262,11 @@ public class IgniteClusterAsyncImpl extends AsyncSupportAdapter<IgniteCluster>
}
/** {@inheritDoc} */
+ @Override public Collection<String> hostNames() {
+ return cluster.hostNames();
+ }
+
+ /** {@inheritDoc} */
@Nullable @Override public ClusterNode node() {
return cluster.node();
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6f66c64/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionSelfTest.java
index 9fbad80..fc2c64d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionSelfTest.java
@@ -248,4 +248,26 @@ public class GridProjectionSelfTest extends GridProjectionAbstractTest {
return even ? cnt - 1 : cnt - 2;
}
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testHostNames() throws Exception {
+ Collection<String> inputHostNames = ignite.cluster().hostNames();
+ Collection<String> localNodeHostNames = ignite.cluster().localNode().hostNames();
+ Collection<String> randomNodeHostNames = ignite.cluster().forRandom().node().hostNames();
+ Collection<ClusterNode> allNodes = ignite.cluster().nodes();
+ Collection<String> checkHostNames = new HashSet<String> ();
+
+ for (ClusterNode currentNode : allNodes)
+ Collections.addAll(checkHostNames, currentNode.hostNames().toArray(new String[0]));
+
+ assert(checkHostNames.equals(inputHostNames));
+
+ if (!(localNodeHostNames.isEmpty()) && !(inputHostNames.isEmpty()))
+ assert((inputHostNames.containsAll(localNodeHostNames)) == true);
+
+ if (!(randomNodeHostNames.isEmpty()) && !(inputHostNames.isEmpty()))
+ assert((inputHostNames.containsAll(randomNodeHostNames)) == true);
+ }
}