You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2015/09/01 15:32:04 UTC
[02/10] storm git commit: Merge remote-tracking branch
'upstream/master'
Merge remote-tracking branch 'upstream/master'
# Conflicts:
# external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsState.java
# external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java
# external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltTest.java
# external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexTopology.java
# external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/TridentEsTopology.java
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/ae0f814d
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/ae0f814d
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/ae0f814d
Branch: refs/heads/master
Commit: ae0f814d589f718c4f989adffccea9b24bbfd3cf
Parents: 6a54fdb 07e0ff2
Author: Alex Panov <al...@teradata.com>
Authored: Sun Aug 30 11:48:37 2015 +0200
Committer: Alex Panov <al...@teradata.com>
Committed: Sun Aug 30 11:48:37 2015 +0200
----------------------------------------------------------------------
.gitignore | 2 +-
CHANGELOG.md | 11 +-
README.markdown | 4 +
STORM-UI-REST-API.md | 41 +-
conf/defaults.yaml | 7 +-
.../nimbus_ha_leader_election_and_failover.png | Bin 0 -> 154316 bytes
.../images/nimbus_ha_topology_submission.png | Bin 0 -> 134180 bytes
docs/documentation/nimbus-ha-design.md | 217 +++++
external/storm-elasticsearch/README.md | 48 +-
.../storm/elasticsearch/bolt/EsIndexBolt.java | 23 +-
.../elasticsearch/bolt/EsPercolateBolt.java | 24 +-
.../common/DefaultEsTupleMapper.java | 42 +
.../elasticsearch/common/EsTupleMapper.java | 55 ++
.../storm/elasticsearch/trident/EsState.java | 22 +-
.../elasticsearch/trident/EsStateFactory.java | 10 +-
.../storm/elasticsearch/trident/EsUpdater.java | 4 +
.../elasticsearch/bolt/AbstractEsBoltTest.java | 5 +-
.../elasticsearch/bolt/EsIndexBoltTest.java | 7 +-
.../elasticsearch/bolt/EsIndexTopology.java | 4 +-
.../elasticsearch/bolt/EsPercolateBoltTest.java | 4 +-
.../storm/elasticsearch/common/EsTestUtil.java | 5 +
.../trident/TridentEsTopology.java | 4 +-
external/storm-hbase/pom.xml | 2 +-
external/storm-hdfs/pom.xml | 4 +-
.../ha/codedistributor/HDFSCodeDistributor.java | 101 +++
external/storm-kafka/README.md | 3 +
.../src/jvm/storm/kafka/KafkaSpout.java | 6 +-
.../src/jvm/storm/kafka/PartitionManager.java | 10 +-
.../src/jvm/storm/kafka/SpoutConfig.java | 3 +
external/storm-solr/README.md | 201 +++++
external/storm-solr/pom.xml | 98 +++
.../apache/storm/solr/bolt/SolrUpdateBolt.java | 136 ++++
.../storm/solr/config/CountBasedCommit.java | 59 ++
.../storm/solr/config/SolrCommitStrategy.java | 30 +
.../apache/storm/solr/config/SolrConfig.java | 42 +
.../storm/solr/mapper/SolrFieldsMapper.java | 182 +++++
.../storm/solr/mapper/SolrJsonMapper.java | 160 ++++
.../apache/storm/solr/mapper/SolrMapper.java | 32 +
.../storm/solr/mapper/SolrMapperException.java | 24 +
.../org/apache/storm/solr/schema/CopyField.java | 50 ++
.../org/apache/storm/solr/schema/Field.java | 50 ++
.../org/apache/storm/solr/schema/FieldType.java | 63 ++
.../org/apache/storm/solr/schema/Schema.java | 116 +++
.../storm/solr/schema/SolrFieldTypeFinder.java | 182 +++++
.../schema/builder/RestJsonSchemaBuilder.java | 69 ++
.../solr/schema/builder/SchemaBuilder.java | 27 +
.../apache/storm/solr/trident/SolrState.java | 67 ++
.../storm/solr/trident/SolrStateFactory.java | 44 +
.../apache/storm/solr/trident/SolrUpdater.java | 33 +
.../storm/solr/spout/SolrFieldsSpout.java | 76 ++
.../apache/storm/solr/spout/SolrJsonSpout.java | 120 +++
.../storm/solr/topology/SolrFieldsTopology.java | 56 ++
.../storm/solr/topology/SolrJsonTopology.java | 48 ++
.../storm/solr/topology/SolrTopology.java | 82 ++
.../solr/trident/SolrFieldsTridentTopology.java | 45 ++
.../solr/trident/SolrJsonTridentTopology.java | 45 ++
.../org/apache/storm/solr/util/TestUtil.java | 30 +
pom.xml | 19 +-
storm-core/pom.xml | 79 +-
storm-core/src/clj/backtype/storm/cluster.clj | 94 ++-
.../backtype/storm/command/shell_submission.clj | 9 +-
storm-core/src/clj/backtype/storm/config.clj | 15 +-
.../src/clj/backtype/storm/daemon/nimbus.clj | 252 ++++--
.../clj/backtype/storm/daemon/supervisor.clj | 53 +-
storm-core/src/clj/backtype/storm/thrift.clj | 23 +-
storm-core/src/clj/backtype/storm/timer.clj | 7 +-
storm-core/src/clj/backtype/storm/ui/core.clj | 95 ++-
storm-core/src/clj/backtype/storm/zookeeper.clj | 108 ++-
storm-core/src/jvm/backtype/storm/Config.java | 47 +-
.../storm/codedistributor/ICodeDistributor.java | 56 ++
.../LocalFileSystemCodeDistributor.java | 106 +++
.../storm/generated/ClusterSummary.java | 292 ++++---
.../backtype/storm/generated/NimbusSummary.java | 796 +++++++++++++++++++
.../backtype/storm/generated/TopologyInfo.java | 221 +++--
.../storm/generated/TopologySummary.java | 107 ++-
.../backtype/storm/nimbus/ILeaderElector.java | 60 ++
.../jvm/backtype/storm/nimbus/NimbusInfo.java | 93 +++
.../jvm/backtype/storm/utils/NimbusClient.java | 78 +-
.../jvm/backtype/storm/utils/RotatingMap.java | 12 +-
.../jvm/backtype/storm/utils/TimeCacheMap.java | 60 +-
.../src/jvm/backtype/storm/utils/Utils.java | 9 +
storm-core/src/py/storm/ttypes.py | 613 ++++++++------
storm-core/src/storm.thrift | 12 +-
storm-core/src/ui/public/index.html | 21 +
.../public/templates/index-page-template.html | 59 +-
.../templates/topology-page-template.html | 18 +-
storm-core/src/ui/public/topology.html | 35 +-
.../test/clj/backtype/storm/cluster_test.clj | 23 +-
.../storm/messaging/netty_unit_test.clj | 2 +-
.../test/clj/backtype/storm/nimbus_test.clj | 210 +++--
.../backtype/storm/security/auth/auth_test.clj | 4 +-
.../storm/security/auth/nimbus_auth_test.clj | 14 +-
.../test/clj/backtype/storm/supervisor_test.clj | 1 +
.../test/clj/backtype/storm/utils_test.clj | 12 -
storm-dist/binary/src/main/assembly/binary.xml | 14 +
95 files changed, 5721 insertions(+), 843 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/ae0f814d/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsState.java
----------------------------------------------------------------------
diff --cc external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsState.java
index e60c003,e3865e5..1eb0d04
--- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsState.java
+++ b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsState.java
@@@ -17,10 -17,10 +17,11 @@@
*/
package org.apache.storm.elasticsearch.trident;
-import backtype.storm.task.IMetricsContext;
import backtype.storm.topology.FailedException;
+
+import org.apache.storm.elasticsearch.common.StormElasticSearchClient;
import org.apache.storm.elasticsearch.common.EsConfig;
+ import org.apache.storm.elasticsearch.common.EsTupleMapper;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
@@@ -29,12 -33,15 +30,13 @@@ import org.slf4j.LoggerFactory
import storm.trident.state.State;
import storm.trident.tuple.TridentTuple;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
/**
+ * Trident State for storing tuple to ES document.
+ * @since 0.11
*/
-public class EsState implements State {
+class EsState implements State {
private static final Logger LOG = LoggerFactory.getLogger(EsState.class);
private static Client client;
private EsConfig esConfig;
@@@ -83,13 -105,20 +88,20 @@@
}
}
+ /**
+ * Store current state to ElasticSearch.
+ *
+ * @param tuples list of tuples for storing to ES.
+ * Each tuple should have relevant fields (source, index, type, id) for EsState's tupleMapper to extract ES document.
+ * @param collector
+ */
- public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
+ public void updateState(List<TridentTuple> tuples) {
BulkRequestBuilder bulkRequest = client.prepareBulk();
for (TridentTuple tuple : tuples) {
- String source = tuple.getStringByField("source");
- String index = tuple.getStringByField("index");
- String type = tuple.getStringByField("type");
- String id = tuple.getStringByField("id");
+ String source = tupleMapper.getSource(tuple);
+ String index = tupleMapper.getIndex(tuple);
+ String type = tupleMapper.getType(tuple);
+ String id = tupleMapper.getId(tuple);
bulkRequest.add(client.prepareIndex(index, type, id).setSource(source));
}
http://git-wip-us.apache.org/repos/asf/storm/blob/ae0f814d/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java
----------------------------------------------------------------------
diff --cc external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java
index 4f4dd39,9fdf7c6..e5733e7
--- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java
+++ b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsStateFactory.java
@@@ -19,30 -19,37 +19,36 @@@ package org.apache.storm.elasticsearch.
import backtype.storm.task.IMetricsContext;
import org.apache.storm.elasticsearch.common.EsConfig;
+ import org.apache.storm.elasticsearch.common.EsTupleMapper;
+
import storm.trident.state.State;
import storm.trident.state.StateFactory;
import java.util.Map;
/**
+ * StateFactory for providing EsState.
+ * @since 0.11
*/
-public class EsStateFactory implements StateFactory {
+class EsStateFactory implements StateFactory {
+ private final EsConfig esConfig;
+ private EsConfig esConfig;
+ private EsTupleMapper tupleMapper;
- public EsStateFactory(){
-
- }
-
/**
* EsStateFactory constructor
* @param esConfig Elasticsearch configuration containing node addresses and cluster name {@link EsConfig}
+ * @param tupleMapper Tuple to ES document mapper {@link EsTupleMapper}
*/
- EsStateFactory(EsConfig esConfig){
- public EsStateFactory(EsConfig esConfig, EsTupleMapper tupleMapper){
++ EsStateFactory(EsConfig esConfig, EsTupleMapper tupleMapper){
this.esConfig = esConfig;
+ this.tupleMapper = tupleMapper;
}
@Override
public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
- EsState esState = new EsState(esConfig);
+ EsState esState = new EsState(esConfig, tupleMapper);
- esState.prepare(conf, metrics, partitionIndex, numPartitions);
+ esState.prepare();
return esState;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/ae0f814d/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsUpdater.java
----------------------------------------------------------------------
diff --cc external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsUpdater.java
index 685ea8c,935c92e..1fb998b
--- a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsUpdater.java
+++ b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/trident/EsUpdater.java
@@@ -24,8 -24,12 +24,12 @@@ import storm.trident.tuple.TridentTuple
import java.util.List;
public class EsUpdater extends BaseStateUpdater<EsState> {
+ /**
+ * {@inheritDoc}
+ * Each tuple should have relevant fields (source, index, type, id) for EsState's tupleMapper to extract ES document.
+ */
@Override
public void updateState(EsState state, List<TridentTuple> tuples, TridentCollector collector) {
- state.updateState(tuples, collector);
+ state.updateState(tuples);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/ae0f814d/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltTest.java
----------------------------------------------------------------------
diff --cc external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltTest.java
index 8a5fad2,17ccdc9..07b7c43
--- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltTest.java
+++ b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/AbstractEsBoltTest.java
@@@ -17,8 -17,8 +17,10 @@@
*/
package org.apache.storm.elasticsearch.bolt;
+import com.google.common.testing.NullPointerTester;
+
+ import backtype.storm.Config;
+ import backtype.storm.task.OutputCollector;
import org.apache.storm.elasticsearch.common.EsConfig;
import org.junit.After;
import org.junit.Before;
http://git-wip-us.apache.org/repos/asf/storm/blob/ae0f814d/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexBoltTest.java
----------------------------------------------------------------------
diff --cc external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexBoltTest.java
index bcba6d4,5860b3b..ba87616
--- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexBoltTest.java
+++ b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexBoltTest.java
@@@ -60,11 -58,7 +58,12 @@@ public class EsIndexBoltTest extends Ab
@Override
protected EsIndexBolt createBolt(EsConfig esConfig) {
- return new EsIndexBolt(esConfig);
+ EsTupleMapper tupleMapper = EsTestUtil.generateDefaultTupleMapper();
+ return new EsIndexBolt(esConfig, tupleMapper);
}
+
+ @Override
+ protected Class<EsIndexBolt> getBoltClass() {
+ return EsIndexBolt.class;
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/ae0f814d/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexTopology.java
----------------------------------------------------------------------
diff --cc external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexTopology.java
index 3c1e949,1f0118b..70e0738
--- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexTopology.java
+++ b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsIndexTopology.java
@@@ -46,8 -47,11 +47,9 @@@ public class EsIndexTopology
TopologyBuilder builder = new TopologyBuilder();
UserDataSpout spout = new UserDataSpout();
builder.setSpout(SPOUT_ID, spout, 1);
- EsConfig esConfig = new EsConfig();
- esConfig.setClusterName(EsConstants.clusterName);
- esConfig.setNodes(new String[]{"localhost:9300"});
+ EsTupleMapper tupleMapper = EsTestUtil.generateDefaultTupleMapper();
+ EsConfig esConfig = new EsConfig(EsConstants.clusterName, new String[]{"localhost:9300"});
- builder.setBolt(BOLT_ID, new EsIndexBolt(esConfig), 1).shuffleGrouping(SPOUT_ID);
+ builder.setBolt(BOLT_ID, new EsIndexBolt(esConfig, tupleMapper), 1).shuffleGrouping(SPOUT_ID);
EsTestUtil.startEsNode();
EsTestUtil.waitForSeconds(5);
http://git-wip-us.apache.org/repos/asf/storm/blob/ae0f814d/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsPercolateBoltTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/ae0f814d/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/TridentEsTopology.java
----------------------------------------------------------------------
diff --cc external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/TridentEsTopology.java
index 1ae1df7,ee5e607..45d86f8
--- a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/TridentEsTopology.java
+++ b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/trident/TridentEsTopology.java
@@@ -45,9 -46,12 +46,10 @@@ public class TridentEsTopology
TridentTopology topology = new TridentTopology();
Stream stream = topology.newStream("spout", spout);
- EsConfig esConfig = new EsConfig();
- esConfig.setClusterName(EsConstants.clusterName);
- esConfig.setNodes(new String[]{"localhost:9300"});
- Fields esFields = new Fields("index", "type", "source", "id");
+ EsConfig esConfig = new EsConfig(EsConstants.clusterName, new String[]{"localhost:9300"});
+ Fields esFields = new Fields("index", "type", "source");
- StateFactory factory = new EsStateFactory(esConfig);
+ EsTupleMapper tupleMapper = EsTestUtil.generateDefaultTupleMapper();
+ StateFactory factory = new EsStateFactory(esConfig, tupleMapper);
TridentState state = stream.partitionPersist(factory, esFields, new EsUpdater(), new Fields());
EsTestUtil.startEsNode();