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/10/28 15:13:40 UTC
[01/14] storm git commit: STORM-697: Added tupleMetaData flag
Repository: storm
Updated Branches:
refs/heads/master 437c4b149 -> ee9e4fffb
STORM-697: Added tupleMetaData flag
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5b4c28a0
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5b4c28a0
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5b4c28a0
Branch: refs/heads/master
Commit: 5b4c28a088ffc62ebcc28e8c28a25d096aa1eb78
Parents: bc54e8e
Author: matt.tieman <ma...@inin.com>
Authored: Tue Mar 3 11:46:30 2015 -0500
Committer: matt.tieman <ma...@inin.com>
Committed: Tue Mar 3 11:46:30 2015 -0500
----------------------------------------------------------------------
external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5b4c28a0/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java b/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
index 5c85983..9db0f5d 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
@@ -38,6 +38,7 @@ public class KafkaConfig implements Serializable {
public long maxOffsetBehind = Long.MAX_VALUE;
public boolean useStartOffsetTimeIfOffsetOutOfRange = true;
public int metricsTimeBucketSizeInSecs = 60;
+ public boolean tupleMetaData = false; // True to generate tuples from MessageAndRealOffset, false to generate only from the message
public KafkaConfig(BrokerHosts hosts, String topic) {
this(hosts, topic, kafka.api.OffsetRequest.DefaultClientId());
[07/14] storm git commit: STORM-697: Changed constructor to accept
instance of MessageMetadataScheme. Assures a type-safe cast in
deserializeMessageWithMetadata
Posted by ka...@apache.org.
STORM-697: Changed constructor to accept instance of MessageMetadataScheme. Assures a type-safe cast in deserializeMessageWithMetadata
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9bfd834a
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9bfd834a
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9bfd834a
Branch: refs/heads/master
Commit: 9bfd834aaf344a63c1f010a40a7f1bce5a315d4f
Parents: e6d319d
Author: Matt Tieman <ma...@inin.com>
Authored: Sat Apr 11 17:36:17 2015 -0400
Committer: Matt Tieman <ma...@inin.com>
Committed: Sat Apr 11 17:36:17 2015 -0400
----------------------------------------------------------------------
.../src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/9bfd834a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
index 5eb20b5..dcdbf8b 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
@@ -3,13 +3,12 @@ package storm.kafka;
import java.util.Arrays;
import java.util.List;
-import backtype.storm.spout.Scheme;
import backtype.storm.spout.SchemeAsMultiScheme;
public class MessageMetadataSchemeAsMultiScheme extends SchemeAsMultiScheme {
private static final long serialVersionUID = -7172403703813625116L;
- public MessageMetadataSchemeAsMultiScheme(Scheme scheme) {
+ public MessageMetadataSchemeAsMultiScheme(MessageMetadataScheme scheme) {
super(scheme);
}
[08/14] storm git commit: STORM-697: Merged in master,
resolved conflicts
Posted by ka...@apache.org.
STORM-697: Merged in master, resolved conflicts
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/472567ec
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/472567ec
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/472567ec
Branch: refs/heads/master
Commit: 472567ec775b054d454617ca640adbf9c3d24383
Parents: 9bfd834 7cf4d25
Author: Matt Tieman <ma...@inin.com>
Authored: Sat Oct 3 14:46:56 2015 -0400
Committer: Matt Tieman <ma...@inin.com>
Committed: Sat Oct 3 14:46:56 2015 -0400
----------------------------------------------------------------------
.gitignore | 9 +-
.travis.yml | 21 +
BYLAWS.md | 96 -
CHANGELOG.md | 303 +-
DEVELOPER.md | 58 +-
LICENSE | 146 +-
NOTICE | 4 +-
README.markdown | 61 +-
SECURITY.md | 273 +-
STORM-UI-REST-API.md | 131 +-
TODO | 178 -
bin/storm | 585 +-
bin/storm-config.cmd | 30 +-
bin/storm.cmd | 40 +-
bin/storm.py | 618 +
conf/defaults.yaml | 26 +-
conf/jaas_kerberos.conf | 17 +
conf/log4j2.xml | 32 +
conf/logback.xml | 28 -
conf/storm-env.sh | 24 +
dev-tools/github/__init__.py | 196 +-
dev-tools/jira-github-join.py | 77 +-
dev-tools/jira/__init__.py | 435 +-
dev-tools/report/__init__.py | 14 +
dev-tools/report/formatter.py | 68 +
dev-tools/report/report.py | 252 +
dev-tools/report/report_builder.py | 86 +
dev-tools/storm-merge.py | 31 +
dev-tools/test-ns.py | 30 +
.../travis/print-errors-from-test-reports.py | 76 +
dev-tools/travis/save-logs.py | 54 +
dev-tools/travis/travis-install.sh | 39 +
dev-tools/travis/travis-script.sh | 44 +
doap_Storm.rdf | 6 +-
docs/README.md | 9 +
docs/_includes/header.html | 13 +-
docs/_posts/2015-03-25-storm094-released.md | 24 +
docs/_posts/2015-06-04-storm095-released.md | 23 +
.../2015-06-15-storm0100-beta-released.md | 294 +
docs/about/multi-language.md | 2 +-
docs/css/main.scss | 3 +-
docs/documentation/BYLAWS.md | 98 +
docs/documentation/Clojure-DSL.md | 4 +-
docs/documentation/Command-line-client.md | 2 +-
docs/documentation/Common-patterns.md | 20 +-
docs/documentation/Concepts.md | 57 +-
docs/documentation/Configuration.md | 4 +-
docs/documentation/Contributing-to-Storm.md | 2 +-
docs/documentation/Distributed-RPC.md | 2 +-
docs/documentation/Documentation.md | 4 +-
docs/documentation/FAQ.md | 7 +-
.../Guaranteeing-message-processing.md | 8 +-
docs/documentation/Home.md | 2 +-
docs/documentation/Hooks.md | 6 +-
docs/documentation/Local-mode.md | 4 +-
docs/documentation/Maven.md | 4 +-
docs/documentation/Metrics.md | 2 +-
docs/documentation/Multilang-protocol.md | 68 +-
docs/documentation/Powered-By.md | 112 +-
...unning-topologies-on-a-production-cluster.md | 10 +-
.../Serialization-(prior-to-0.6.0).md | 4 +-
docs/documentation/Serialization.md | 2 +-
.../documentation/Setting-up-a-Storm-cluster.md | 17 +-
docs/documentation/Structure-of-the-codebase.md | 10 +-
docs/documentation/Transactional-topologies.md | 18 +-
docs/documentation/Trident-API-Overview.md | 4 +-
docs/documentation/Troubleshooting.md | 37 +
docs/documentation/Tutorial.md | 24 +-
...nding-the-parallelism-of-a-Storm-topology.md | 21 +-
.../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 +
docs/downloads.html | 128 +-
docs/images/architecture.png | Bin 0 -> 69825 bytes
docs/images/architecture.svg | 1458 ++
docs/images/bolt.png | Bin 0 -> 24796 bytes
docs/images/bolt.svg | 743 +
docs/images/security.png | Bin 0 -> 72415 bytes
docs/images/security.svg | 1779 ++
docs/images/spout.png | Bin 0 -> 22911 bytes
docs/images/spout.svg | 833 +
docs/images/storm.svg | 1326 ++
docs/images/topology.png | Bin 59837 -> 64740 bytes
docs/images/topology.svg | 1044 ++
docs/images/topology_dark.png | Bin 0 -> 49692 bytes
docs/images/topology_dark.svg | 1101 ++
examples/storm-starter/README.markdown | 16 +-
.../storm-starter/multilang/resources/storm.js | 373 -
.../storm-starter/multilang/resources/storm.py | 260 -
.../storm-starter/multilang/resources/storm.rb | 236 -
examples/storm-starter/pom.xml | 97 +-
.../jvm/storm/starter/BasicDRPCTopology.java | 8 +-
.../src/jvm/storm/starter/ReachTopology.java | 4 +-
.../storm/starter/SkewedRollingTopWords.java | 134 +
.../storm/starter/TransactionalGlobalCount.java | 5 +-
.../storm/starter/bolt/AbstractRankerBolt.java | 4 +-
.../storm/starter/bolt/RollingCountAggBolt.java | 77 +
.../storm/starter/bolt/RollingCountBolt.java | 4 +-
.../starter/trident/TridentKafkaWordCount.java | 230 +
.../jvm/storm/starter/util/TupleHelpers.java | 33 -
external/README.md | 18 +
external/flux/.gitignore | 15 +
external/flux/README.md | 834 +
external/flux/flux-core/pom.xml | 95 +
.../main/java/org/apache/storm/flux/Flux.java | 263 +
.../java/org/apache/storm/flux/FluxBuilder.java | 595 +
.../apache/storm/flux/api/TopologySource.java | 39 +
.../org/apache/storm/flux/model/BeanDef.java | 39 +
.../apache/storm/flux/model/BeanReference.java | 39 +
.../org/apache/storm/flux/model/BoltDef.java | 24 +
.../storm/flux/model/ConfigMethodDef.java | 62 +
.../storm/flux/model/ExecutionContext.java | 77 +
.../apache/storm/flux/model/GroupingDef.java | 77 +
.../org/apache/storm/flux/model/IncludeDef.java | 54 +
.../org/apache/storm/flux/model/ObjectDef.java | 90 +
.../apache/storm/flux/model/PropertyDef.java | 58 +
.../org/apache/storm/flux/model/SpoutDef.java | 24 +
.../org/apache/storm/flux/model/StreamDef.java | 64 +
.../apache/storm/flux/model/TopologyDef.java | 216 +
.../storm/flux/model/TopologySourceDef.java | 36 +
.../org/apache/storm/flux/model/VertexDef.java | 36 +
.../apache/storm/flux/parser/FluxParser.java | 208 +
.../flux-core/src/main/resources/splash.txt | 9 +
.../org/apache/storm/flux/FluxBuilderTest.java | 31 +
.../org/apache/storm/flux/IntegrationTest.java | 39 +
.../java/org/apache/storm/flux/TCKTest.java | 245 +
.../multilang/MultilangEnvirontmentTest.java | 89 +
.../apache/storm/flux/test/SimpleTopology.java | 59 +
.../storm/flux/test/SimpleTopologySource.java | 52 +
.../test/SimpleTopologyWithConfigParam.java | 55 +
.../org/apache/storm/flux/test/TestBolt.java | 80 +
.../storm/flux/test/TridentTopologySource.java | 71 +
.../src/test/resources/configs/bad_hbase.yaml | 98 +
.../resources/configs/config-methods-test.yaml | 70 +
.../resources/configs/diamond-topology.yaml | 87 +
.../existing-topology-method-override.yaml | 25 +
.../existing-topology-reflection-config.yaml | 24 +
.../configs/existing-topology-reflection.yaml | 24 +
.../configs/existing-topology-trident.yaml | 24 +
.../resources/configs/existing-topology.yaml | 23 +
.../src/test/resources/configs/hdfs_test.yaml | 97 +
.../test/resources/configs/include_test.yaml | 25 +
.../configs/invalid-existing-topology.yaml | 33 +
.../src/test/resources/configs/kafka_test.yaml | 126 +
.../src/test/resources/configs/shell_test.yaml | 104 +
.../test/resources/configs/simple_hbase.yaml | 120 +
.../resources/configs/substitution-test.yaml | 106 +
.../src/test/resources/configs/tck.yaml | 95 +
.../src/test/resources/configs/test.properties | 18 +
.../flux-core/src/test/resources/log4j2.xml | 34 +
external/flux/flux-examples/README.md | 66 +
external/flux/flux-examples/pom.xml | 140 +
.../storm/flux/examples/WordCountClient.java | 74 +
.../apache/storm/flux/examples/WordCounter.java | 71 +
.../src/main/resources/hbase_bolt.properties | 18 +
.../src/main/resources/hdfs_bolt.properties | 26 +
.../src/main/resources/kafka_spout.yaml | 136 +
.../src/main/resources/multilang.yaml | 89 +
.../src/main/resources/simple_hbase.yaml | 92 +
.../src/main/resources/simple_hdfs.yaml | 105 +
.../src/main/resources/simple_wordcount.yaml | 68 +
external/flux/flux-ui/README.md | 3 +
external/flux/flux-wrappers/pom.xml | 51 +
.../flux/wrappers/bolts/FluxShellBolt.java | 56 +
.../storm/flux/wrappers/bolts/LogInfoBolt.java | 44 +
.../flux/wrappers/spouts/FluxShellSpout.java | 55 +
.../main/resources/resources/randomsentence.js | 93 +
.../main/resources/resources/splitsentence.py | 24 +
external/flux/pom.xml | 119 +
external/storm-elasticsearch/README.md | 99 +
external/storm-elasticsearch/pom.xml | 108 +
.../elasticsearch/ElasticsearchGetRequest.java | 36 +
.../elasticsearch/EsLookupResultOutput.java | 43 +
.../elasticsearch/bolt/AbstractEsBolt.java | 81 +
.../storm/elasticsearch/bolt/EsIndexBolt.java | 75 +
.../storm/elasticsearch/bolt/EsLookupBolt.java | 81 +
.../elasticsearch/bolt/EsPercolateBolt.java | 88 +
.../common/DefaultEsTupleMapper.java | 42 +
.../storm/elasticsearch/common/EsConfig.java | 82 +
.../elasticsearch/common/EsTupleMapper.java | 55 +
.../common/StormElasticSearchClient.java | 48 +
.../common/TransportAddresses.java | 72 +
.../storm/elasticsearch/trident/EsState.java | 113 +
.../elasticsearch/trident/EsStateFactory.java | 55 +
.../storm/elasticsearch/trident/EsUpdater.java | 35 +
.../bolt/AbstractEsBoltIntegrationTest.java | 91 +
.../elasticsearch/bolt/AbstractEsBoltTest.java | 65 +
.../elasticsearch/bolt/EsIndexBoltTest.java | 69 +
.../elasticsearch/bolt/EsIndexTopology.java | 120 +
.../bolt/EsLookupBoltIntegrationTest.java | 137 +
.../elasticsearch/bolt/EsLookupBoltTest.java | 125 +
.../elasticsearch/bolt/EsPercolateBoltTest.java | 62 +
.../elasticsearch/common/EsConfigTest.java | 71 +
.../storm/elasticsearch/common/EsConstants.java | 22 +
.../storm/elasticsearch/common/EsTestUtil.java | 75 +
.../common/TransportAddressesTest.java | 81 +
.../trident/EsStateFactoryTest.java | 32 +
.../trident/TridentEsTopology.java | 135 +
external/storm-eventhubs/README.md | 41 +
external/storm-eventhubs/pom.xml | 113 +
.../eventhubs/bolt/DefaultEventDataFormat.java | 47 +
.../storm/eventhubs/bolt/EventHubBolt.java | 101 +
.../eventhubs/bolt/EventHubBoltConfig.java | 109 +
.../storm/eventhubs/bolt/IEventDataFormat.java | 28 +
.../eventhubs/samples/AtMostOnceEventCount.java | 54 +
.../storm/eventhubs/samples/EventCount.java | 158 +
.../storm/eventhubs/samples/EventHubLoop.java | 52 +
.../samples/OpaqueTridentEventCount.java | 53 +
.../samples/TransactionalTridentEventCount.java | 81 +
.../eventhubs/samples/bolt/GlobalCountBolt.java | 83 +
.../samples/bolt/PartialCountBolt.java | 63 +
.../apache/storm/eventhubs/spout/EventData.java | 48 +
.../storm/eventhubs/spout/EventDataScheme.java | 55 +
.../eventhubs/spout/EventHubReceiverImpl.java | 149 +
.../storm/eventhubs/spout/EventHubSpout.java | 263 +
.../eventhubs/spout/EventHubSpoutConfig.java | 179 +
.../eventhubs/spout/EventHubSpoutException.java | 37 +
.../storm/eventhubs/spout/FieldConstants.java | 25 +
.../storm/eventhubs/spout/IEventDataScheme.java | 30 +
.../eventhubs/spout/IEventHubReceiver.java | 36 +
.../spout/IEventHubReceiverFactory.java | 30 +
.../eventhubs/spout/IPartitionCoordinator.java | 27 +
.../eventhubs/spout/IPartitionManager.java | 37 +
.../spout/IPartitionManagerFactory.java | 33 +
.../storm/eventhubs/spout/IStateStore.java | 31 +
.../apache/storm/eventhubs/spout/MessageId.java | 56 +
.../storm/eventhubs/spout/PartitionManager.java | 101 +
.../eventhubs/spout/SimplePartitionManager.java | 139 +
.../spout/StaticPartitionCoordinator.java | 85 +
.../eventhubs/spout/ZookeeperStateStore.java | 95 +
.../storm/eventhubs/trident/Coordinator.java | 60 +
.../trident/ITridentPartitionManager.java | 35 +
.../ITridentPartitionManagerFactory.java | 26 +
.../trident/OpaqueTridentEventHubEmitter.java | 69 +
.../trident/OpaqueTridentEventHubSpout.java | 64 +
.../storm/eventhubs/trident/Partition.java | 39 +
.../storm/eventhubs/trident/Partitions.java | 41 +
.../TransactionalTridentEventHubEmitter.java | 167 +
.../TransactionalTridentEventHubSpout.java | 66 +
.../trident/TridentPartitionManager.java | 93 +
.../src/main/resources/config.properties | 45 +
.../eventhubs/spout/EventHubReceiverMock.java | 97 +
.../spout/EventHubSpoutCallerMock.java | 96 +
.../spout/PartitionManagerCallerMock.java | 105 +
.../spout/SpoutOutputCollectorMock.java | 66 +
.../storm/eventhubs/spout/StateStoreMock.java | 54 +
.../storm/eventhubs/spout/TestEventData.java | 47 +
.../eventhubs/spout/TestEventHubSpout.java | 72 +
.../eventhubs/spout/TestPartitionManager.java | 117 +
.../TestTransactionalTridentEmitter.java | 93 +
.../eventhubs/trident/TridentCollectorMock.java | 52 +
external/storm-hbase/LICENSE | 202 -
external/storm-hbase/pom.xml | 4 +-
.../org/apache/storm/hbase/bolt/HBaseBolt.java | 2 +-
.../storm/hbase/bolt/HBaseLookupBolt.java | 4 +-
.../org/apache/storm/hbase/common/Utils.java | 10 +-
.../mapper/SimpleTridentHBaseMapMapper.java | 50 +
.../trident/mapper/TridentHBaseMapMapper.java | 40 +
.../hbase/trident/state/HBaseMapState.java | 47 +-
.../storm/hbase/trident/state/HBaseState.java | 2 +-
external/storm-hdfs/README.md | 9 +
external/storm-hdfs/pom.xml | 67 +-
.../org/apache/storm/hdfs/bolt/HdfsBolt.java | 143 +-
.../storm/hdfs/bolt/SequenceFileBolt.java | 2 +-
.../hdfs/common/security/HdfsSecurityUtil.java | 5 +-
.../ha/codedistributor/HDFSCodeDistributor.java | 101 +
.../apache/storm/hdfs/trident/HdfsState.java | 392 +-
.../trident/rotation/FileRotationPolicy.java | 14 +
.../rotation/FileSizeRotationPolicy.java | 13 +
.../hdfs/trident/rotation/NoRotationPolicy.java | 10 +
.../trident/rotation/TimedRotationPolicy.java | 31 +-
.../storm/hdfs/bolt/HdfsFileTopology.java | 6 +-
.../storm/hdfs/bolt/SequenceFileTopology.java | 4 +-
.../apache/storm/hdfs/bolt/TestHdfsBolt.java | 242 +
.../storm/hdfs/trident/FixedBatchSpout.java | 2 +-
.../storm/hdfs/trident/HdfsStateTest.java | 206 +
.../storm/hdfs/trident/TridentFileTopology.java | 2 +-
.../hdfs/trident/TridentSequenceTopology.java | 6 +-
external/storm-hive/README.md | 115 +
external/storm-hive/pom.xml | 188 +
.../org/apache/storm/hive/bolt/HiveBolt.java | 328 +
.../bolt/mapper/DelimitedRecordHiveMapper.java | 143 +
.../storm/hive/bolt/mapper/HiveMapper.java | 81 +
.../hive/bolt/mapper/JsonRecordHiveMapper.java | 132 +
.../apache/storm/hive/common/HiveOptions.java | 157 +
.../org/apache/storm/hive/common/HiveUtils.java | 76 +
.../apache/storm/hive/common/HiveWriter.java | 419 +
.../apache/storm/hive/trident/HiveState.java | 306 +
.../storm/hive/trident/HiveStateFactory.java | 48 +
.../apache/storm/hive/trident/HiveUpdater.java | 31 +
.../apache/storm/hive/bolt/HiveSetupUtil.java | 220 +
.../apache/storm/hive/bolt/HiveTopology.java | 150 +
.../hive/bolt/HiveTopologyPartitioned.java | 153 +
.../apache/storm/hive/bolt/TestHiveBolt.java | 404 +
.../storm/hive/common/TestHiveWriter.java | 193 +
.../storm/hive/trident/TridentHiveTopology.java | 190 +
external/storm-jdbc/README.md | 302 +
external/storm-jdbc/pom.xml | 129 +
.../storm/jdbc/bolt/AbstractJdbcBolt.java | 62 +
.../apache/storm/jdbc/bolt/JdbcInsertBolt.java | 96 +
.../apache/storm/jdbc/bolt/JdbcLookupBolt.java | 77 +
.../org/apache/storm/jdbc/common/Column.java | 111 +
.../storm/jdbc/common/ConnectionProvider.java | 43 +
.../jdbc/common/HikariCPConnectionProvider.java | 63 +
.../apache/storm/jdbc/common/JdbcClient.java | 235 +
.../java/org/apache/storm/jdbc/common/Util.java | 75 +
.../storm/jdbc/mapper/JdbcLookupMapper.java | 43 +
.../apache/storm/jdbc/mapper/JdbcMapper.java | 33 +
.../jdbc/mapper/SimpleJdbcLookupMapper.java | 63 +
.../storm/jdbc/mapper/SimpleJdbcMapper.java | 94 +
.../storm/jdbc/trident/state/JdbcQuery.java | 40 +
.../storm/jdbc/trident/state/JdbcState.java | 160 +
.../jdbc/trident/state/JdbcStateFactory.java | 40 +
.../storm/jdbc/trident/state/JdbcUpdater.java | 32 +
.../storm/jdbc/common/JdbcClientTest.java | 94 +
.../org/apache/storm/jdbc/common/UtilTest.java | 69 +
.../org/apache/storm/jdbc/spout/UserSpout.java | 90 +
.../jdbc/topology/AbstractUserTopology.java | 115 +
.../jdbc/topology/UserPersistanceTopology.java | 62 +
.../UserPersistanceTridentTopology.java | 61 +
external/storm-jdbc/src/test/sql/test.sql | 18 +
external/storm-kafka/CHANGELOG.md | 13 -
external/storm-kafka/README.md | 117 +-
external/storm-kafka/pom.xml | 25 +-
.../jvm/storm/kafka/DynamicBrokersReader.java | 26 +
.../ExponentialBackoffMsgRetryManager.java | 27 +-
.../jvm/storm/kafka/FailedMsgRetryManager.java | 3 +
.../src/jvm/storm/kafka/KafkaConfig.java | 2 +-
.../src/jvm/storm/kafka/KafkaSpout.java | 30 +-
.../src/jvm/storm/kafka/KafkaUtils.java | 21 +-
.../src/jvm/storm/kafka/PartitionManager.java | 32 +-
.../src/jvm/storm/kafka/SpoutConfig.java | 3 +
.../src/jvm/storm/kafka/ZkCoordinator.java | 2 +-
.../src/jvm/storm/kafka/ZkState.java | 2 +-
.../src/jvm/storm/kafka/bolt/KafkaBolt.java | 101 +-
.../kafka/trident/OpaqueTridentKafkaSpout.java | 4 +-
.../trident/TransactionalTridentKafkaSpout.java | 4 +-
.../kafka/trident/TridentKafkaEmitter.java | 4 +-
.../storm/kafka/DynamicBrokersReaderTest.java | 13 +
.../ExponentialBackoffMsgRetryManagerTest.java | 43 +-
.../src/test/storm/kafka/KafkaUtilsTest.java | 37 +-
.../src/test/storm/kafka/ZkCoordinatorTest.java | 1 +
.../test/storm/kafka/bolt/KafkaBoltTest.java | 136 +-
external/storm-redis/README.md | 254 +
external/storm-redis/pom.xml | 65 +
.../storm/redis/bolt/AbstractRedisBolt.java | 109 +
.../storm/redis/bolt/RedisLookupBolt.java | 133 +
.../apache/storm/redis/bolt/RedisStoreBolt.java | 121 +
.../redis/common/config/JedisClusterConfig.java | 131 +
.../redis/common/config/JedisPoolConfig.java | 163 +
.../common/container/JedisClusterContainer.java | 65 +
.../JedisCommandsContainerBuilder.java | 53 +
.../JedisCommandsInstanceContainer.java | 37 +
.../redis/common/container/JedisContainer.java | 76 +
.../common/mapper/RedisDataTypeDescription.java | 70 +
.../redis/common/mapper/RedisLookupMapper.java | 43 +
.../storm/redis/common/mapper/RedisMapper.java | 29 +
.../redis/common/mapper/RedisStoreMapper.java | 24 +
.../storm/redis/common/mapper/TupleMapper.java | 41 +
.../trident/state/AbstractRedisMapState.java | 137 +
.../state/AbstractRedisStateQuerier.java | 95 +
.../state/AbstractRedisStateUpdater.java | 96 +
.../storm/redis/trident/state/KeyFactory.java | 57 +
.../storm/redis/trident/state/Options.java | 40 +
.../trident/state/RedisClusterMapState.java | 350 +
.../redis/trident/state/RedisClusterState.java | 112 +
.../trident/state/RedisClusterStateQuerier.java | 71 +
.../trident/state/RedisClusterStateUpdater.java | 93 +
.../redis/trident/state/RedisMapState.java | 381 +
.../storm/redis/trident/state/RedisState.java | 115 +
.../redis/trident/state/RedisStateQuerier.java | 69 +
.../redis/trident/state/RedisStateUpdater.java | 98 +
.../storm/redis/topology/LookupWordCount.java | 166 +
.../redis/topology/PersistentWordCount.java | 116 +
.../storm/redis/topology/WordCounter.java | 67 +
.../apache/storm/redis/topology/WordSpout.java | 88 +
.../storm/redis/trident/PrintFunction.java | 40 +
.../redis/trident/WordCountLookupMapper.java | 57 +
.../redis/trident/WordCountStoreMapper.java | 39 +
.../redis/trident/WordCountTridentRedis.java | 98 +
.../trident/WordCountTridentRedisCluster.java | 106 +
.../WordCountTridentRedisClusterMap.java | 101 +
.../redis/trident/WordCountTridentRedisMap.java | 94 +
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 +
log4j2/cluster.xml | 76 +
log4j2/worker.xml | 77 +
logback/cluster.xml | 85 -
logback/worker.xml | 41 -
pom.xml | 231 +-
.../maven-shade-clojure-transformer/pom.xml | 2 +-
storm-buildtools/storm-maven-plugins/pom.xml | 81 +
.../storm/maven/plugin/util/CommandExec.java | 89 +
.../plugin/versioninfo/VersionInfoMojo.java | 304 +
storm-core/pom.xml | 492 +-
storm-core/src/clj/backtype/storm/bootstrap.clj | 64 -
storm-core/src/clj/backtype/storm/clojure.clj | 2 +-
storm-core/src/clj/backtype/storm/cluster.clj | 243 +-
.../clj/backtype/storm/command/get_errors.clj | 52 +
.../clj/backtype/storm/command/kill_workers.clj | 33 +
.../backtype/storm/command/shell_submission.clj | 9 +-
storm-core/src/clj/backtype/storm/config.clj | 38 +-
storm-core/src/clj/backtype/storm/converter.clj | 238 +
.../backtype/storm/daemon/builtin_metrics.clj | 25 +-
.../src/clj/backtype/storm/daemon/common.clj | 60 +-
.../src/clj/backtype/storm/daemon/drpc.clj | 27 +-
.../src/clj/backtype/storm/daemon/executor.clj | 218 +-
.../src/clj/backtype/storm/daemon/logviewer.clj | 117 +-
.../src/clj/backtype/storm/daemon/nimbus.clj | 474 +-
.../clj/backtype/storm/daemon/supervisor.clj | 312 +-
.../src/clj/backtype/storm/daemon/task.clj | 26 +-
.../src/clj/backtype/storm/daemon/worker.clj | 225 +-
storm-core/src/clj/backtype/storm/disruptor.clj | 23 +-
.../src/clj/backtype/storm/local_state.clj | 99 +
storm-core/src/clj/backtype/storm/log.clj | 2 +-
.../src/clj/backtype/storm/messaging/loader.clj | 13 +-
.../src/clj/backtype/storm/messaging/local.clj | 2 +-
storm-core/src/clj/backtype/storm/stats.clj | 78 +-
storm-core/src/clj/backtype/storm/testing.clj | 25 +-
storm-core/src/clj/backtype/storm/thrift.clj | 43 +-
storm-core/src/clj/backtype/storm/timer.clj | 7 +-
storm-core/src/clj/backtype/storm/tuple.clj | 8 +-
storm-core/src/clj/backtype/storm/ui/core.clj | 467 +-
.../src/clj/backtype/storm/ui/helpers.clj | 88 +-
storm-core/src/clj/backtype/storm/util.clj | 78 +-
storm-core/src/clj/backtype/storm/zookeeper.clj | 134 +-
.../src/dev/drpc-simple-acl-test-scenario.yaml | 17 +
storm-core/src/dev/resources/storm.js | 373 -
storm-core/src/dev/resources/storm.py | 260 -
storm-core/src/dev/resources/storm.rb | 236 -
storm-core/src/genthrift.sh | 13 +-
storm-core/src/java_license_header.txt | 17 +
storm-core/src/jvm/backtype/storm/Config.java | 369 +-
.../jvm/backtype/storm/ConfigValidation.java | 54 +-
.../src/jvm/backtype/storm/LogWriter.java | 83 +
.../src/jvm/backtype/storm/StormSubmitter.java | 88 +-
.../storm/codedistributor/ICodeDistributor.java | 56 +
.../LocalFileSystemCodeDistributor.java | 106 +
.../storm/coordination/BatchBoltExecutor.java | 4 +-
.../coordination/BatchSubtopologyBuilder.java | 17 +-
.../storm/drpc/DRPCInvocationsClient.java | 6 +
.../src/jvm/backtype/storm/drpc/DRPCSpout.java | 28 +-
.../storm/drpc/LinearDRPCInputDeclarer.java | 5 +-
.../storm/drpc/LinearDRPCTopologyBuilder.java | 22 +-
.../jvm/backtype/storm/drpc/ReturnResults.java | 8 +-
.../storm/generated/AlreadyAliveException.java | 149 +-
.../backtype/storm/generated/Assignment.java | 983 ++
.../storm/generated/AuthorizationException.java | 149 +-
.../src/jvm/backtype/storm/generated/Bolt.java | 194 +-
.../jvm/backtype/storm/generated/BoltStats.java | 908 +-
.../storm/generated/ClusterSummary.java | 533 +-
.../storm/generated/ClusterWorkerHeartbeat.java | 768 +
.../storm/generated/ComponentCommon.java | 426 +-
.../storm/generated/ComponentObject.java | 86 +-
.../backtype/storm/generated/Credentials.java | 220 +-
.../storm/generated/DRPCExecutionException.java | 149 +-
.../backtype/storm/generated/DRPCRequest.java | 185 +-
.../backtype/storm/generated/DebugOptions.java | 506 +
.../storm/generated/DistributedRPC.java | 529 +-
.../generated/DistributedRPCInvocations.java | 1199 +-
.../jvm/backtype/storm/generated/ErrorInfo.java | 300 +-
.../backtype/storm/generated/ExecutorInfo.java | 198 +-
.../storm/generated/ExecutorSpecificStats.java | 72 +-
.../backtype/storm/generated/ExecutorStats.java | 589 +-
.../storm/generated/ExecutorSummary.java | 371 +-
.../storm/generated/GetInfoOptions.java | 422 +
.../storm/generated/GlobalStreamId.java | 185 +-
.../jvm/backtype/storm/generated/Grouping.java | 163 +-
.../generated/InvalidTopologyException.java | 149 +-
.../backtype/storm/generated/JavaObject.java | 239 +-
.../backtype/storm/generated/JavaObjectArg.java | 108 +-
.../backtype/storm/generated/KillOptions.java | 176 +-
.../storm/generated/LSApprovedWorkers.java | 458 +
.../generated/LSSupervisorAssignments.java | 471 +
.../storm/generated/LSSupervisorId.java | 406 +
.../storm/generated/LSWorkerHeartbeat.java | 755 +
.../storm/generated/LocalAssignment.java | 561 +
.../storm/generated/LocalStateData.java | 471 +
.../jvm/backtype/storm/generated/Nimbus.java | 14822 ++++++++++++-----
.../backtype/storm/generated/NimbusSummary.java | 796 +
.../jvm/backtype/storm/generated/NodeInfo.java | 556 +
.../storm/generated/NotAliveException.java | 149 +-
.../backtype/storm/generated/NullStruct.java | 112 +-
.../storm/generated/NumErrorsChoice.java | 65 +
.../storm/generated/RebalanceOptions.java | 348 +-
.../storm/generated/ShellComponent.java | 202 +-
.../jvm/backtype/storm/generated/SpoutSpec.java | 194 +-
.../backtype/storm/generated/SpoutStats.java | 614 +-
.../storm/generated/StateSpoutSpec.java | 194 +-
.../jvm/backtype/storm/generated/StormBase.java | 1382 ++
.../backtype/storm/generated/StormTopology.java | 410 +-
.../backtype/storm/generated/StreamInfo.java | 249 +-
.../backtype/storm/generated/SubmitOptions.java | 208 +-
.../storm/generated/SupervisorInfo.java | 1288 ++
.../storm/generated/SupervisorSummary.java | 424 +-
.../storm/generated/ThriftSerializedObject.java | 516 +
.../storm/generated/TopologyActionOptions.java | 387 +
.../backtype/storm/generated/TopologyInfo.java | 881 +-
.../storm/generated/TopologyInitialStatus.java | 3 +-
.../storm/generated/TopologyStatus.java | 68 +
.../storm/generated/TopologySummary.java | 587 +-
.../storm/grouping/PartialKeyGrouping.java | 107 +
.../storm/messaging/ConnectionWithStatus.java | 49 +
.../backtype/storm/messaging/netty/Client.java | 694 +-
.../backtype/storm/messaging/netty/Context.java | 53 +-
.../storm/messaging/netty/MessageBatch.java | 24 +-
.../storm/messaging/netty/MessageBuffer.java | 58 +
.../messaging/netty/SaslStormClientHandler.java | 5 +-
.../backtype/storm/messaging/netty/Server.java | 182 +-
.../messaging/netty/StormClientHandler.java | 46 +
.../netty/StormClientPipelineFactory.java | 5 +-
.../backtype/storm/metric/EventLoggerBolt.java | 58 +
.../storm/metric/FileBasedEventLogger.java | 105 +
.../jvm/backtype/storm/metric/IEventLogger.java | 59 +
.../storm/metric/MetricsConsumerBolt.java | 2 +-
.../jvm/backtype/storm/metric/SystemBolt.java | 15 +
.../backtype/storm/nimbus/ILeaderElector.java | 60 +
.../jvm/backtype/storm/nimbus/NimbusInfo.java | 93 +
.../auth/DefaultHttpCredentialsPlugin.java | 19 +-
.../storm/security/auth/ITransportPlugin.java | 4 +-
.../storm/security/auth/ReqContext.java | 28 +-
.../security/auth/SaslTransportPlugin.java | 26 +-
.../security/auth/ShellBasedGroupsMapping.java | 2 +-
.../security/auth/SimpleTransportPlugin.java | 10 +-
.../storm/security/auth/TBackoffConnect.java | 4 +-
.../storm/security/auth/ThriftClient.java | 12 +-
.../auth/authorizer/DRPCAuthorizerBase.java | 17 +
.../authorizer/DRPCSimpleACLAuthorizer.java | 20 +-
.../authorizer/ImpersonationAuthorizer.java | 171 +
.../auth/authorizer/SimpleACLAuthorizer.java | 55 +-
.../auth/digest/DigestSaslTransportPlugin.java | 6 +-
.../auth/digest/ServerCallbackHandler.java | 21 +-
.../kerberos/KerberosSaslTransportPlugin.java | 14 +-
.../security/auth/kerberos/NoOpTTrasport.java | 40 +
.../auth/kerberos/ServerCallbackHandler.java | 38 +-
.../auth/kerberos/jaas_kerberos_cluster.conf | 20 +-
.../auth/kerberos/jaas_kerberos_launcher.conf | 19 +
.../serialization/BlowfishTupleSerializer.java | 5 +-
.../DefaultSerializationDelegate.java | 11 +-
.../GzipBridgeSerializationDelegate.java | 7 +-
.../GzipBridgeThriftSerializationDelegate.java | 64 +
.../GzipSerializationDelegate.java | 10 +-
.../GzipThriftSerializationDelegate.java | 57 +
.../serialization/SerializationDelegate.java | 2 +-
.../ThriftSerializationDelegate.java | 52 +
.../storm/spout/ISpoutOutputCollector.java | 6 +-
.../jvm/backtype/storm/spout/ShellSpout.java | 16 +
.../storm/spout/SpoutOutputCollector.java | 5 +
.../src/jvm/backtype/storm/task/IBolt.java | 4 +-
.../src/jvm/backtype/storm/task/ShellBolt.java | 22 +-
.../backtype/storm/task/TopologyContext.java | 149 +-
.../testing/OpaqueMemoryTransactionalSpout.java | 2 +-
.../backtype/storm/testing/SpoutTracker.java | 6 +
.../storm/testing/TestWordBytesCounter.java | 27 +
.../backtype/storm/testing/TestWordCounter.java | 6 +-
.../topology/BaseConfigurationDeclarer.java | 2 +-
.../backtype/storm/topology/BoltDeclarer.java | 4 +
.../ComponentConfigurationDeclarer.java | 2 +-
.../storm/topology/IBasicOutputCollector.java | 5 +-
.../backtype/storm/topology/InputDeclarer.java | 131 +
.../storm/topology/TopologyBuilder.java | 51 +-
.../TransactionalTopologyBuilder.java | 25 +-
.../src/jvm/backtype/storm/tuple/Fields.java | 21 +-
.../src/jvm/backtype/storm/tuple/ITuple.java | 126 +-
.../src/jvm/backtype/storm/tuple/Tuple.java | 3 +-
.../jvm/backtype/storm/utils/DRPCClient.java | 2 +-
.../utils/DisruptorBackpressureCallback.java | 27 +
.../backtype/storm/utils/DisruptorQueue.java | 206 +-
.../storm/utils/ExtendedThreadPoolExecutor.java | 67 +
.../jvm/backtype/storm/utils/LocalState.java | 163 +-
.../src/jvm/backtype/storm/utils/Monitor.java | 8 +-
.../jvm/backtype/storm/utils/NimbusClient.java | 82 +-
.../jvm/backtype/storm/utils/RateTracker.java | 166 +
.../jvm/backtype/storm/utils/RotatingMap.java | 12 +-
.../jvm/backtype/storm/utils/ShellProcess.java | 29 +-
.../StormBoundedExponentialBackoffRetry.java | 4 +-
.../jvm/backtype/storm/utils/TimeCacheMap.java | 60 +-
.../backtype/storm/utils/TransferDrainer.java | 62 +-
.../jvm/backtype/storm/utils/TupleUtils.java | 35 +
.../src/jvm/backtype/storm/utils/Utils.java | 317 +-
.../jvm/backtype/storm/utils/VersionInfo.java | 131 +
.../backtype/storm/utils/VersionedStore.java | 9 +-
.../storm/utils/WorkerBackpressureCallback.java | 26 +
.../storm/utils/WorkerBackpressureThread.java | 59 +
storm-core/src/jvm/storm/trident/Stream.java | 2 +-
.../src/jvm/storm/trident/TridentTopology.java | 19 +-
.../trident/drpc/ReturnResultsReducer.java | 4 +-
.../storm/trident/planner/SubtopologyBolt.java | 2 +-
.../storm/trident/spout/BatchSpoutExecutor.java | 2 +-
.../jvm/storm/trident/spout/IBatchSpout.java | 2 +-
.../spout/IOpaquePartitionedTridentSpout.java | 2 +-
.../trident/spout/IPartitionedTridentSpout.java | 2 +-
.../jvm/storm/trident/spout/ITridentSpout.java | 6 +-
.../OpaquePartitionedTridentSpoutExecutor.java | 20 +-
.../spout/PartitionedTridentSpoutExecutor.java | 26 +-
.../trident/spout/RichSpoutBatchExecutor.java | 16 +-
.../trident/spout/RichSpoutBatchTriggerer.java | 18 +-
.../trident/spout/TridentSpoutCoordinator.java | 6 +-
.../trident/spout/TridentSpoutExecutor.java | 10 +-
.../storm/trident/testing/FeederBatchSpout.java | 4 +-
.../testing/FeederCommitterBatchSpout.java | 2 +-
.../storm/trident/testing/FixedBatchSpout.java | 2 +-
.../trident/topology/TridentBoltExecutor.java | 20 +-
.../topology/TridentTopologyBuilder.java | 35 +-
storm-core/src/multilang/js/storm.js | 366 -
storm-core/src/multilang/py/storm.py | 260 -
storm-core/src/multilang/rb/storm.rb | 236 -
.../worker-launcher/.deps/worker-launcher.Po | 16 +
storm-core/src/py/__init__.py | 2 +
storm-core/src/py/storm/DistributedRPC-remote | 35 +-
storm-core/src/py/storm/DistributedRPC.py | 40 +-
.../py/storm/DistributedRPCInvocations-remote | 43 +-
.../src/py/storm/DistributedRPCInvocations.py | 97 +-
storm-core/src/py/storm/Nimbus-remote | 121 +-
storm-core/src/py/storm/Nimbus.py | 1183 +-
storm-core/src/py/storm/__init__.py | 2 +
storm-core/src/py/storm/constants.py | 8 +-
storm-core/src/py/storm/ttypes.py | 2720 ++-
storm-core/src/py_license_header.txt | 18 +
.../storm-core-version-info.properties | 24 +
storm-core/src/storm.thrift | 131 +-
storm-core/src/ui/public/component.html | 175 +-
.../src/ui/public/css/bootstrap-1.4.0.css | 2469 ---
.../src/ui/public/css/bootstrap-3.3.1.min.css | 5 +
.../src/ui/public/css/dataTables.bootstrap.css | 319 +
.../public/css/jquery.dataTables.1.10.4.min.css | 1 +
.../src/ui/public/css/jsonFormatter.min.css | 1 +
storm-core/src/ui/public/css/style.css | 129 +-
.../src/ui/public/images/back_disabled.png | Bin 0 -> 1361 bytes
.../src/ui/public/images/back_enabled.png | Bin 0 -> 1379 bytes
.../src/ui/public/images/back_enabled_hover.png | Bin 0 -> 1375 bytes
.../src/ui/public/images/forward_disabled.png | Bin 0 -> 1363 bytes
.../src/ui/public/images/forward_enabled.png | Bin 0 -> 1380 bytes
.../ui/public/images/forward_enabled_hover.png | Bin 0 -> 1379 bytes
storm-core/src/ui/public/images/sort_asc.png | Bin 0 -> 1118 bytes
.../src/ui/public/images/sort_asc_disabled.png | Bin 0 -> 2916 bytes
storm-core/src/ui/public/images/sort_both.png | Bin 0 -> 1136 bytes
storm-core/src/ui/public/images/sort_desc.png | Bin 0 -> 1127 bytes
.../src/ui/public/images/sort_desc_disabled.png | Bin 0 -> 1045 bytes
storm-core/src/ui/public/index.html | 124 +-
.../src/ui/public/js/bootstrap-3.3.1.min.js | 7 +
storm-core/src/ui/public/js/bootstrap-twipsy.js | 310 -
.../ui/public/js/dataTables.bootstrap.min.js | 8 +
.../src/ui/public/js/jquery-1.11.1.min.js | 4 +
storm-core/src/ui/public/js/jquery-1.6.2.min.js | 18 -
.../public/js/jquery.dataTables.1.10.4.min.js | 157 +
.../src/ui/public/js/jquery.tablesorter.min.js | 5 -
.../src/ui/public/js/jsonFormatter.min.js | 2 +
storm-core/src/ui/public/js/script.js | 171 +-
.../public/templates/anti-forgery-template.html | 19 -
.../templates/component-page-template.html | 147 +-
.../public/templates/index-page-template.html | 115 +-
.../templates/topology-page-template.html | 110 +-
.../src/ui/public/templates/user-template.html | 2 +-
storm-core/src/ui/public/topology.html | 175 +-
.../test/clj/backtype/storm/clojure_test.clj | 9 +-
.../test/clj/backtype/storm/cluster_test.clj | 48 +-
.../test/clj/backtype/storm/config_test.clj | 65 +-
.../test/clj/backtype/storm/drpc_test.clj | 8 +-
.../test/clj/backtype/storm/grouping_test.clj | 50 +-
.../clj/backtype/storm/integration_test.clj | 20 +-
.../clj/backtype/storm/local_state_test.clj | 40 +-
.../test/clj/backtype/storm/logviewer_test.clj | 8 +-
.../storm/messaging/netty_integration_test.clj | 5 +-
.../storm/messaging/netty_unit_test.clj | 75 +-
.../test/clj/backtype/storm/messaging_test.clj | 6 +-
.../test/clj/backtype/storm/metrics_test.clj | 8 +-
.../test/clj/backtype/storm/multilang_test.clj | 6 +-
.../test/clj/backtype/storm/nimbus_test.clj | 332 +-
.../scheduler/multitenant_scheduler_test.clj | 4 +-
.../test/clj/backtype/storm/scheduler_test.clj | 4 +-
.../storm/security/auth/AuthUtils_test.clj | 6 +-
.../auth/DefaultHttpCredentialsPlugin_test.clj | 31 +-
.../backtype/storm/security/auth/auth_test.clj | 163 +-
.../authorizer/DRPCSimpleACLAuthorizer_test.clj | 15 +
.../storm/security/auth/drpc-auth-alice.jaas | 17 +
.../storm/security/auth/drpc-auth-bob.jaas | 17 +
.../storm/security/auth/drpc-auth-charlie.jaas | 17 +
.../storm/security/auth/drpc-auth-server.jaas | 17 +
.../storm/security/auth/drpc_auth_test.clj | 13 +-
.../storm/security/auth/nimbus_auth_test.clj | 245 +-
.../clj/backtype/storm/subtopology_test.clj | 9 +-
.../test/clj/backtype/storm/supervisor_test.clj | 53 +-
.../test/clj/backtype/storm/tick_tuple_test.clj | 7 +-
.../clj/backtype/storm/transactional_test.clj | 20 +-
.../test/clj/backtype/storm/utils_test.clj | 12 -
.../test/clj/backtype/storm/worker_test.clj | 37 +
.../storm/grouping/PartialKeyGroupingTest.java | 66 +
.../GzipBridgeSerializationDelegateTest.java | 6 +-
...ipBridgeThriftSerializationDelegateTest.java | 71 +
.../ThriftBridgeSerializationDelegateTest.java | 60 +
.../storm/topology/TopologyBuilderTest.java | 48 +
.../utils/DisruptorQueueBackpressureTest.java | 115 +
.../storm/utils/DisruptorQueueTest.java | 40 +-
.../backtype/storm/utils/RateTrackerTest.java | 94 +
storm-core/test/resources/log4j2-test.xml | 32 +
storm-core/test/resources/test_runner.clj | 114 +
storm-dist/binary/LICENSE | 224 +-
storm-dist/binary/NOTICE | 4 +-
storm-dist/binary/pom.xml | 2 +-
storm-dist/binary/src/main/assembly/binary.xml | 182 +-
storm-dist/source/pom.xml | 2 +-
storm-multilang/javascript/pom.xml | 32 +
.../src/main/resources/resources/storm.js | 373 +
storm-multilang/python/pom.xml | 32 +
.../src/main/resources/resources/storm.py | 260 +
storm-multilang/ruby/pom.xml | 32 +
.../ruby/src/main/resources/resources/storm.rb | 236 +
734 files changed, 83995 insertions(+), 19288 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/472567ec/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/472567ec/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/472567ec/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
----------------------------------------------------------------------
diff --cc external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
index e1186e3,052d525..ed36fbe
--- a/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
@@@ -134,17 -135,16 +135,23 @@@ public class PartitionManager
if (toEmit == null) {
return EmitState.NO_EMITTED;
}
- Iterable<List<Object>> tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg);
+
+ Iterable<List<Object>> tups;
+ if (_spoutConfig.tupleMetaData) {
+ tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg, _partition, toEmit.offset);
+ } else {
+ tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg);
+ }
+
if (tups != null) {
- for (List<Object> tup : tups) {
- collector.emit(tup, new KafkaMessageId(_partition, toEmit.offset));
+ if(_spoutConfig.topicAsStreamId) {
+ for (List<Object> tup : tups) {
+ collector.emit(_spoutConfig.topic, tup, new KafkaMessageId(_partition, toEmit.offset));
+ }
+ } else {
+ for (List<Object> tup : tups) {
+ collector.emit(tup, new KafkaMessageId(_partition, toEmit.offset));
+ }
}
break;
} else {
http://git-wip-us.apache.org/repos/asf/storm/blob/472567ec/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/472567ec/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
----------------------------------------------------------------------
diff --cc external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
index 362d721,e25d904..8f4343e
--- a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
+++ b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
@@@ -28,24 -23,28 +28,26 @@@ import java.util.Properties
import kafka.api.OffsetRequest;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
- import kafka.javaapi.producer.Producer;
import kafka.message.MessageAndOffset;
- import kafka.producer.KeyedMessage;
- import kafka.producer.ProducerConfig;
+
+ import org.apache.kafka.clients.producer.KafkaProducer;
+ import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.After;
++import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
-import org.junit.Assert;
-import storm.kafka.trident.GlobalPartitionInformation;
-
-import java.util.List;
-import java.util.Properties;
-
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import org.mockito.Mockito;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+import storm.kafka.trident.GlobalPartitionInformation;
+import backtype.storm.spout.SchemeAsMultiScheme;
+import backtype.storm.utils.Utils;
+
+import com.google.common.collect.ImmutableMap;
-
public class KafkaUtilsTest {
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaUtilsTest.class);
private KafkaTestBroker broker;
private SimpleConsumer simpleConsumer;
private KafkaConfig config;
[06/14] storm git commit: STORM-697: Updated TridentKafkaEmitter.emit
to accept message partition and offset. If config.tupleMetaData is true,
generate tuples with metadata
Posted by ka...@apache.org.
STORM-697: Updated TridentKafkaEmitter.emit to accept message partition and offset. If config.tupleMetaData is true, generate tuples with metadata
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/e6d319d7
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/e6d319d7
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/e6d319d7
Branch: refs/heads/master
Commit: e6d319d79e193bd64b5dba7011200e9876cdafa9
Parents: 1119465
Author: Matt Tieman <ma...@inin.com>
Authored: Sat Apr 11 17:34:37 2015 -0400
Committer: Matt Tieman <ma...@inin.com>
Committed: Sat Apr 11 17:34:37 2015 -0400
----------------------------------------------------------------------
.../jvm/storm/kafka/trident/TridentKafkaEmitter.java | 14 ++++++++++----
1 file changed, 10 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/e6d319d7/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
index 1a9be43..e2e0f2a 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
@@ -121,7 +121,7 @@ public class TridentKafkaEmitter {
long endoffset = offset;
for (MessageAndOffset msg : msgs) {
- emit(collector, msg.message());
+ emit(collector, msg.message(), partition, msg.offset());
endoffset = msg.nextOffset();
}
Map newMeta = new HashMap();
@@ -172,15 +172,21 @@ public class TridentKafkaEmitter {
if (offset > nextOffset) {
throw new RuntimeException("Error when re-emitting batch. overshot the end offset");
}
- emit(collector, msg.message());
+ emit(collector, msg.message(), partition, msg.offset());
offset = msg.nextOffset();
}
}
}
}
- private void emit(TridentCollector collector, Message msg) {
- Iterable<List<Object>> values = KafkaUtils.generateTuples(_config, msg);
+ private void emit(TridentCollector collector, Message msg, Partition partition, long offset) {
+ Iterable<List<Object>> values;
+ if (_config.tupleMetaData) {
+ values = KafkaUtils.generateTuples(_config, msg, partition, offset);
+ } else {
+ values = KafkaUtils.generateTuples(_config, msg);
+ }
+
if (values != null) {
for (List<Object> value : values) {
collector.emit(value);
[02/14] storm git commit: STORM-697: Overload of generateTuples to
accept the Partition and offset
Posted by ka...@apache.org.
STORM-697: Overload of generateTuples to accept the Partition and offset
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/6e4fde20
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/6e4fde20
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/6e4fde20
Branch: refs/heads/master
Commit: 6e4fde20af8d285cdf4829e4c2c4aef4cd45d89d
Parents: 5b4c28a
Author: matt.tieman <ma...@inin.com>
Authored: Tue Mar 3 11:47:38 2015 -0500
Committer: matt.tieman <ma...@inin.com>
Committed: Tue Mar 3 11:47:38 2015 -0500
----------------------------------------------------------------------
.../storm-kafka/src/jvm/storm/kafka/KafkaUtils.java | 15 +++++++++++++++
1 file changed, 15 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/6e4fde20/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java b/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
index 137dc99..9af49fe 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
@@ -210,6 +210,21 @@ public class KafkaUtils {
}
return tups;
}
+
+ public static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg, Partition partition, int offset) {
+ Iterable<List<Object>> tups;
+ ByteBuffer payload = msg.payload();
+ if (payload == null) {
+ return null;
+ }
+
+ if (kafkaConfig.scheme instanceof MessageMetadataSchemeAsMultiScheme) {
+ tups = ((MessageMetadataSchemeAsMultiScheme) kafkaConfig.scheme).deserializeMessageWithMetadata(Utils.toByteArray(payload), partition, offset);
+ } else {
+ tups = kafkaConfig.scheme.deserialize(Utils.toByteArray(payload));
+ }
+ return tups;
+ }
public static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex) {
[11/14] storm git commit: STORM-697: Updated kafka-storm README to
include MessageMetadataSchemeAsMultiScheme as an implementation of
MultiScheme
Posted by ka...@apache.org.
STORM-697: Updated kafka-storm README to include MessageMetadataSchemeAsMultiScheme as an implementation of MultiScheme
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/25e7bc49
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/25e7bc49
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/25e7bc49
Branch: refs/heads/master
Commit: 25e7bc496a85c663d50b0c114c620557f35ee034
Parents: 4d79d9b
Author: Matt Tieman <ma...@inin.com>
Authored: Mon Oct 26 23:07:01 2015 -0400
Committer: Matt Tieman <ma...@inin.com>
Committed: Mon Oct 26 23:07:01 2015 -0400
----------------------------------------------------------------------
external/storm-kafka/README.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/25e7bc49/external/storm-kafka/README.md
----------------------------------------------------------------------
diff --git a/external/storm-kafka/README.md b/external/storm-kafka/README.md
index ed39dc5..1bf14b7 100644
--- a/external/storm-kafka/README.md
+++ b/external/storm-kafka/README.md
@@ -104,8 +104,8 @@ also controls the naming of your output field.
```
The default `RawMultiScheme` just takes the `byte[]` and returns a tuple with `byte[]` as is. The name of the
-outputField is "bytes". There are alternative implementation like `SchemeAsMultiScheme` and
-`KeyValueSchemeAsMultiScheme` which can convert the `byte[]` to `String`.
+outputField is "bytes". There are alternative implementations like `SchemeAsMultiScheme`,
+`KeyValueSchemeAsMultiScheme`, and `MessageMetadataSchemeAsMultiScheme` which can convert the `byte[]` to `String`.
### Examples
[03/14] storm git commit: STORM-697: Test for
MessageMetadataSchemeAsMultiScheme and generateTuples with metadata using
SchemeAsMultiScheme
Posted by ka...@apache.org.
STORM-697: Test for MessageMetadataSchemeAsMultiScheme and generateTuples with metadata using SchemeAsMultiScheme
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/6e768665
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/6e768665
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/6e768665
Branch: refs/heads/master
Commit: 6e768665320d08815c53f27e706ef2ae1ff5af78
Parents: 6e4fde2
Author: matt.tieman <ma...@inin.com>
Authored: Tue Mar 3 11:48:57 2015 -0500
Committer: matt.tieman <ma...@inin.com>
Committed: Tue Mar 3 11:48:57 2015 -0500
----------------------------------------------------------------------
.../src/test/storm/kafka/KafkaUtilsTest.java | 66 +++++++++++++++++---
1 file changed, 56 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/6e768665/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
index 1f1bbbc..a7c9b2b 100644
--- a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
+++ b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
@@ -17,9 +17,14 @@
*/
package storm.kafka;
-import backtype.storm.spout.SchemeAsMultiScheme;
-import backtype.storm.utils.Utils;
-import com.google.common.collect.ImmutableMap;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+
+import java.util.List;
+import java.util.Properties;
+
import kafka.api.OffsetRequest;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
@@ -27,18 +32,17 @@ import kafka.javaapi.producer.Producer;
import kafka.message.MessageAndOffset;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
+
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import storm.kafka.trident.GlobalPartitionInformation;
+import org.mockito.Mockito;
-import java.util.List;
-import java.util.Properties;
+import storm.kafka.trident.GlobalPartitionInformation;
+import backtype.storm.spout.SchemeAsMultiScheme;
+import backtype.storm.utils.Utils;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import com.google.common.collect.ImmutableMap;
public class KafkaUtilsTest {
@@ -166,6 +170,47 @@ public class KafkaUtilsTest {
assertEquals(value, lists.iterator().next().get(0));
}
}
+
+ @Test
+ public void generateTuplesWithMessageAndMetadataScheme() {
+ String value = "value";
+ Partition mockPartition = Mockito.mock(Partition.class);
+ mockPartition.partition = 0;
+ int offset = 0;
+
+ config.scheme = new MessageMetadataSchemeAsMultiScheme(new StringMessageAndMetadataScheme());
+ config.tupleMetaData = true;
+
+ createTopicAndSendMessage(null, value);
+ ByteBufferMessageSet messageAndOffsets = getLastMessage();
+ for (MessageAndOffset msg : messageAndOffsets) {
+ Iterable<List<Object>> lists = KafkaUtils.generateTuples(config, msg.message(), mockPartition, offset);
+ List<Object> values = lists.iterator().next();
+ assertEquals("Message is incorrect", value, values.get(0));
+ assertEquals("Offset is incorrect", offset, values.get(1));
+ assertEquals("Partition is incorrect", mockPartition.partition, values.get(2));
+ }
+ }
+
+ @Test
+ public void generateTuplesWithValueSchemeAndMessageAndMetadata() {
+ String value = "value";
+ Partition mockPartition = Mockito.mock(Partition.class);
+ mockPartition.partition = 0;
+ int offset = 0;
+
+ config.scheme = new SchemeAsMultiScheme(new StringScheme());
+ config.tupleMetaData = true;
+
+ createTopicAndSendMessage(null, value);
+ ByteBufferMessageSet messageAndOffsets = getLastMessage();
+ for (MessageAndOffset msg : messageAndOffsets) {
+ Iterable<List<Object>> lists = KafkaUtils.generateTuples(config, msg.message(), mockPartition, offset);
+ List<Object> values = lists.iterator().next();
+ assertEquals("Incorrect number of tuple values", 1, values.size());
+ assertEquals("Message is incorrect", value, values.get(0));
+ }
+ }
private ByteBufferMessageSet getLastMessage() {
long offsetOfLastMessage = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, OffsetRequest.LatestTime()) - 1;
@@ -174,6 +219,7 @@ public class KafkaUtilsTest {
private void runGetValueOnlyTuplesTest() {
String value = "value";
+
createTopicAndSendMessage(null, value);
ByteBufferMessageSet messageAndOffsets = getLastMessage();
for (MessageAndOffset msg : messageAndOffsets) {
[14/14] storm git commit: add STORM-697 to CHANGELOG.md
Posted by ka...@apache.org.
add STORM-697 to CHANGELOG.md
* also add Matthew Tieman to the contributors list
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/ee9e4fff
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/ee9e4fff
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/ee9e4fff
Branch: refs/heads/master
Commit: ee9e4fffb684d3ba19c0cce8e97775bea5061bb4
Parents: 81f6355
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Wed Oct 28 23:12:48 2015 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Wed Oct 28 23:12:48 2015 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
README.markdown | 5 +++--
2 files changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/ee9e4fff/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1f50307..f7fd810 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 0.11.0
+ * STORM-697: Support for Emitting Kafka Message Offset and Partition
* STORM-1074: Add Avro HDFS bolt
* STORM-566: Improve documentation including incorrect Kryo ser. framework docs
* STORM-1073: Refactor AbstractHdfsBolt
http://git-wip-us.apache.org/repos/asf/storm/blob/ee9e4fff/README.markdown
----------------------------------------------------------------------
diff --git a/README.markdown b/README.markdown
index 052dcd2..3f66e42 100644
--- a/README.markdown
+++ b/README.markdown
@@ -235,8 +235,9 @@ under the License.
* Dmytro Dragan ([@3Dragan](https://github.com/3Dragan))
* Ningyu An ([@jetan9](https://github.com/jetan9))
* Pete Prokopowicz ([@prokopowicz](https://github.com/prokopowicz))
-* Priyank Shah ([@priyank5485](https://github.com/priyank5485)]
-* Joshua Martell ([@jmartell7](https://github.com/jmartell7)]
+* Priyank Shah ([@priyank5485](https://github.com/priyank5485))
+* Joshua Martell ([@jmartell7](https://github.com/jmartell7))
+* Matthew Tieman ([@mjtieman](https://github.com/mjtieman)])
## Acknowledgements
[10/14] storm git commit: STORM-697: Review feedback: Removed
tupleMetaData from SpoutConfig,
used instanceof check on spout scheme to determine if tuples should be
generated with meta data
Posted by ka...@apache.org.
STORM-697: Review feedback: Removed tupleMetaData from SpoutConfig, used instanceof check on spout scheme to determine if tuples should be generated with meta data
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/4d79d9b3
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/4d79d9b3
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/4d79d9b3
Branch: refs/heads/master
Commit: 4d79d9b3b66bd381d6654e68fb418d2efad8e922
Parents: 47f4a3c
Author: Matt Tieman <ma...@inin.com>
Authored: Fri Oct 23 23:47:13 2015 -0400
Committer: Matt Tieman <ma...@inin.com>
Committed: Fri Oct 23 23:47:13 2015 -0400
----------------------------------------------------------------------
.../src/jvm/storm/kafka/KafkaConfig.java | 4 ++--
.../src/jvm/storm/kafka/KafkaUtils.java | 11 ++-------
.../MessageMetadataSchemeAsMultiScheme.java | 1 -
.../src/jvm/storm/kafka/PartitionManager.java | 8 +++++--
.../kafka/trident/TridentKafkaEmitter.java | 4 ++--
.../src/test/storm/kafka/KafkaUtilsTest.java | 25 ++------------------
6 files changed, 14 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/4d79d9b3/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java b/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
index ea0e421..49c7526 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
@@ -23,7 +23,8 @@ import backtype.storm.spout.RawMultiScheme;
import java.io.Serializable;
public class KafkaConfig implements Serializable {
-
+ private static final long serialVersionUID = 5276718734571623855L;
+
public final BrokerHosts hosts;
public final String topic;
public final String clientId;
@@ -38,7 +39,6 @@ public class KafkaConfig implements Serializable {
public long maxOffsetBehind = Long.MAX_VALUE;
public boolean useStartOffsetTimeIfOffsetOutOfRange = true;
public int metricsTimeBucketSizeInSecs = 60;
- public boolean tupleMetaData = false; // True to generate tuples from MessageAndRealOffset, false to generate only from the message
public KafkaConfig(BrokerHosts hosts, String topic) {
this(hosts, topic, kafka.api.OffsetRequest.DefaultClientId());
http://git-wip-us.apache.org/repos/asf/storm/blob/4d79d9b3/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java b/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
index 340f629..2e047b3 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
@@ -208,19 +208,12 @@ public class KafkaUtils {
return tups;
}
- public static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg, Partition partition, long offset) {
- Iterable<List<Object>> tups;
+ public static Iterable<List<Object>> generateTuples(MessageMetadataSchemeAsMultiScheme scheme, Message msg, Partition partition, long offset) {
ByteBuffer payload = msg.payload();
if (payload == null) {
return null;
}
-
- if (kafkaConfig.scheme instanceof MessageMetadataSchemeAsMultiScheme) {
- tups = ((MessageMetadataSchemeAsMultiScheme) kafkaConfig.scheme).deserializeMessageWithMetadata(Utils.toByteArray(payload), partition, offset);
- } else {
- tups = kafkaConfig.scheme.deserialize(Utils.toByteArray(payload));
- }
- return tups;
+ return scheme.deserializeMessageWithMetadata(Utils.toByteArray(payload), partition, offset);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d79d9b3/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
index dcdbf8b..e89e391 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
@@ -12,7 +12,6 @@ public class MessageMetadataSchemeAsMultiScheme extends SchemeAsMultiScheme {
super(scheme);
}
- @SuppressWarnings("unchecked")
public Iterable<List<Object>> deserializeMessageWithMetadata(byte[] message, Partition partition, long offset) {
List<Object> o = ((MessageMetadataScheme) scheme).deserializeMessageWithMetadata(message, partition, offset);
if (o == null) {
http://git-wip-us.apache.org/repos/asf/storm/blob/4d79d9b3/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java b/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
index 980ed8b..39e42ed 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
@@ -23,12 +23,16 @@ import backtype.storm.metric.api.CountMetric;
import backtype.storm.metric.api.MeanReducer;
import backtype.storm.metric.api.ReducedMetric;
import backtype.storm.spout.SpoutOutputCollector;
+
import com.google.common.collect.ImmutableMap;
+
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
import kafka.message.MessageAndOffset;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
import storm.kafka.KafkaSpout.EmitState;
import storm.kafka.KafkaSpout.MessageAndRealOffset;
import storm.kafka.trident.MaxMetric;
@@ -137,8 +141,8 @@ public class PartitionManager {
}
Iterable<List<Object>> tups;
- if (_spoutConfig.tupleMetaData) {
- tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg, _partition, toEmit.offset);
+ if (_spoutConfig.scheme instanceof MessageMetadataSchemeAsMultiScheme) {
+ tups = KafkaUtils.generateTuples((MessageMetadataSchemeAsMultiScheme) _spoutConfig.scheme, toEmit.msg, _partition, toEmit.offset);
} else {
tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d79d9b3/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
index 5ac5709..39aac1a 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaEmitter.java
@@ -181,8 +181,8 @@ public class TridentKafkaEmitter {
private void emit(TridentCollector collector, Message msg, Partition partition, long offset) {
Iterable<List<Object>> values;
- if (_config.tupleMetaData) {
- values = KafkaUtils.generateTuples(_config, msg, partition, offset);
+ if (_config.scheme instanceof MessageMetadataSchemeAsMultiScheme) {
+ values = KafkaUtils.generateTuples((MessageMetadataSchemeAsMultiScheme) _config.scheme, msg, partition, offset);
} else {
values = KafkaUtils.generateTuples(_config, msg);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/4d79d9b3/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
index 8f4343e..65e8d2b 100644
--- a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
+++ b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
@@ -180,39 +180,18 @@ public class KafkaUtilsTest {
mockPartition.partition = 0;
long offset = 0L;
- config.scheme = new MessageMetadataSchemeAsMultiScheme(new StringMessageAndMetadataScheme());
- config.tupleMetaData = true;
+ MessageMetadataSchemeAsMultiScheme scheme = new MessageMetadataSchemeAsMultiScheme(new StringMessageAndMetadataScheme());
createTopicAndSendMessage(null, value);
ByteBufferMessageSet messageAndOffsets = getLastMessage();
for (MessageAndOffset msg : messageAndOffsets) {
- Iterable<List<Object>> lists = KafkaUtils.generateTuples(config, msg.message(), mockPartition, offset);
+ Iterable<List<Object>> lists = KafkaUtils.generateTuples(scheme, msg.message(), mockPartition, offset);
List<Object> values = lists.iterator().next();
assertEquals("Message is incorrect", value, values.get(0));
assertEquals("Partition is incorrect", mockPartition.partition, values.get(1));
assertEquals("Offset is incorrect", offset, values.get(2));
}
}
-
- @Test
- public void generateTuplesWithValueSchemeAndMessageAndMetadata() {
- String value = "value";
- Partition mockPartition = Mockito.mock(Partition.class);
- mockPartition.partition = 0;
- Long offset = 0L;
-
- config.scheme = new SchemeAsMultiScheme(new StringScheme());
- config.tupleMetaData = true;
-
- createTopicAndSendMessage(null, value);
- ByteBufferMessageSet messageAndOffsets = getLastMessage();
- for (MessageAndOffset msg : messageAndOffsets) {
- Iterable<List<Object>> lists = KafkaUtils.generateTuples(config, msg.message(), mockPartition, offset);
- List<Object> values = lists.iterator().next();
- assertEquals("Incorrect number of tuple values", 1, values.size());
- assertEquals("Message is incorrect", value, values.get(0));
- }
- }
private ByteBufferMessageSet getLastMessage() {
long offsetOfLastMessage = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, OffsetRequest.LatestTime()) - 1;
[04/14] storm git commit: STORM-697: Added scheme to include
Partition and offset when generating tuple. >>>
Posted by ka...@apache.org.
STORM-697: Added scheme to include Partition and offset when generating tuple. >>>
The MessageMetadataScheme interface extends Sheme and defines a deserialization method that accepts the message byte[], Partition, and the offset.
MessageMetadataSchemeAsMultiScheme follows the same pattern as KeyValueSchemeAsMultiScheme, extending SchemeAsMultiScheme and providing a deserialization method named for the method defined by
MessageMetadataScheme.
StringMessageAndMetadataScheme provides an implementation of MessageMetadataScheme, following the same pattern as StringKeyValueScheme.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2f119c6e
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2f119c6e
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2f119c6e
Branch: refs/heads/master
Commit: 2f119c6e2edace030afeb9ee0885010f1de7fc28
Parents: 6e76866
Author: matt.tieman <ma...@inin.com>
Authored: Tue Mar 3 11:50:04 2015 -0500
Committer: matt.tieman <ma...@inin.com>
Committed: Tue Mar 3 11:59:44 2015 -0500
----------------------------------------------------------------------
.../jvm/storm/kafka/MessageMetadataScheme.java | 25 ++++++++++++++++++++
.../MessageMetadataSchemeAsMultiScheme.java | 25 ++++++++++++++++++++
.../kafka/StringMessageAndMetadataScheme.java | 25 ++++++++++++++++++++
3 files changed, 75 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/2f119c6e/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
new file mode 100644
index 0000000..d0dd2be
--- /dev/null
+++ b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
@@ -0,0 +1,25 @@
+package storm.kafka;
+
+import java.util.List;
+import backtype.storm.spout.Scheme;
+
+/**
+ * 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.
+ */
+public interface MessageMetadataScheme extends Scheme {
+ public List<Object> deserializeMessageWithMetadata(byte[] message, Partition partition, int offset);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/2f119c6e/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
new file mode 100644
index 0000000..6226676
--- /dev/null
+++ b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
@@ -0,0 +1,25 @@
+package storm.kafka;
+
+import java.util.Arrays;
+import java.util.List;
+
+import backtype.storm.spout.Scheme;
+import backtype.storm.spout.SchemeAsMultiScheme;
+
+public class MessageMetadataSchemeAsMultiScheme extends SchemeAsMultiScheme {
+ private static final long serialVersionUID = -7172403703813625116L;
+
+ public MessageMetadataSchemeAsMultiScheme(Scheme scheme) {
+ super(scheme);
+ }
+
+ @SuppressWarnings("unchecked")
+ public Iterable<List<Object>> deserializeMessageWithMetadata(byte[] message, Partition partition, int offset) {
+ List<Object> o = ((MessageMetadataScheme) scheme).deserializeMessageWithMetadata(message, partition, offset);
+ if (o == null) {
+ return null;
+ } else {
+ return Arrays.asList(o);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/2f119c6e/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java b/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
new file mode 100644
index 0000000..262a27c
--- /dev/null
+++ b/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
@@ -0,0 +1,25 @@
+package storm.kafka;
+
+import java.util.List;
+
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Values;
+
+public class StringMessageAndMetadataScheme extends StringScheme implements MessageMetadataScheme {
+ private static final long serialVersionUID = -5441841920447947374L;
+
+ public static final String STRING_SCHEME_PARTITION_KEY = "partition";
+ public static final String STRING_SCHEME_OFFSET = "offset";
+
+ @Override
+ public List<Object> deserializeMessageWithMetadata(byte[] message, Partition partition, int offset) {
+ String stringMessage = StringScheme.deserializeString(message);
+ return new Values(stringMessage, partition.partition, offset);
+ }
+
+ @Override
+ public Fields getOutputFields() {
+ return new Fields(STRING_SCHEME_KEY, STRING_SCHEME_PARTITION_KEY, STRING_SCHEME_OFFSET);
+ }
+
+}
[09/14] storm git commit: STORM-697: Merged in master,
resolved conflicts
Posted by ka...@apache.org.
STORM-697: Merged in master, resolved conflicts
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/47f4a3c3
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/47f4a3c3
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/47f4a3c3
Branch: refs/heads/master
Commit: 47f4a3c3d3396b35928578e36b27dbcc49f00752
Parents: 472567e 93043e6
Author: matt.tieman <ma...@inin.com>
Authored: Wed Oct 14 17:31:08 2015 -0400
Committer: matt.tieman <ma...@inin.com>
Committed: Wed Oct 14 17:31:08 2015 -0400
----------------------------------------------------------------------
CHANGELOG.md | 14 +
README.markdown | 1 +
STORM-UI-REST-API.md | 4 +-
bin/storm | 20 +-
bin/storm.py | 35 +-
conf/defaults.yaml | 18 +
dev-tools/storm-merge.py | 2 +-
docs/DYNAMIC_LOG_LEVEL_SETTINGS.md | 41 +
docs/images/dynamic_log_level_settings_1.png | Bin 0 -> 93689 bytes
docs/images/dynamic_log_level_settings_2.png | Bin 0 -> 78785 bytes
examples/storm-starter/pom.xml | 7 +
.../storm/starter/MultipleLoggerTopology.java | 105 +
.../starter/ResourceAwareExampleTopology.java | 87 +
.../bolt/IntermediateRankingsBoltTest.java | 2 +-
.../starter/bolt/RollingCountBoltTest.java | 2 +-
.../starter/bolt/TotalRankingsBoltTest.java | 2 +-
.../storm/starter/tools/MockTupleHelpers.java | 40 -
external/storm-hdfs/pom.xml | 15 +
.../apache/storm/hdfs/bolt/TestHdfsBolt.java | 18 +-
external/storm-hive/pom.xml | 7 +
.../apache/storm/hive/bolt/TestHiveBolt.java | 56 +-
.../src/jvm/storm/kafka/PartitionManager.java | 20 +-
storm-core/pom.xml | 12 +
storm-core/src/clj/backtype/storm/cluster.clj | 28 +-
.../backtype/storm/command/set_log_level.clj | 75 +
storm-core/src/clj/backtype/storm/converter.clj | 4 +-
.../src/clj/backtype/storm/daemon/common.clj | 4 +-
.../src/clj/backtype/storm/daemon/drpc.clj | 26 +-
.../src/clj/backtype/storm/daemon/nimbus.clj | 232 +-
.../clj/backtype/storm/daemon/supervisor.clj | 13 +-
.../src/clj/backtype/storm/daemon/worker.clj | 134 +-
storm-core/src/clj/backtype/storm/log.clj | 12 +-
storm-core/src/clj/backtype/storm/stats.clj | 1282 ++-
storm-core/src/clj/backtype/storm/ui/core.clj | 1050 +--
.../src/clj/backtype/storm/ui/helpers.clj | 30 -
storm-core/src/jvm/backtype/storm/Config.java | 70 +
.../jvm/backtype/storm/ConfigValidation.java | 22 +
.../src/jvm/backtype/storm/StormSubmitter.java | 35 +-
.../storm/coordination/CoordinatedBolt.java | 2 +-
.../backtype/storm/generated/Assignment.java | 194 +-
.../storm/generated/BoltAggregateStats.java | 704 ++
.../jvm/backtype/storm/generated/BoltStats.java | 442 +-
.../storm/generated/ClusterSummary.java | 110 +-
.../storm/generated/ClusterWorkerHeartbeat.java | 54 +-
.../storm/generated/CommonAggregateStats.java | 902 ++
.../generated/ComponentAggregateStats.java | 752 ++
.../storm/generated/ComponentPageInfo.java | 2194 +++++
.../backtype/storm/generated/ComponentType.java | 62 +
.../backtype/storm/generated/Credentials.java | 46 +-
.../storm/generated/ExecutorAggregateStats.java | 526 ++
.../backtype/storm/generated/ExecutorStats.java | 170 +-
.../storm/generated/LSApprovedWorkers.java | 46 +-
.../generated/LSSupervisorAssignments.java | 50 +-
.../storm/generated/LSWorkerHeartbeat.java | 38 +-
.../storm/generated/LocalAssignment.java | 38 +-
.../storm/generated/LocalStateData.java | 50 +-
.../jvm/backtype/storm/generated/LogConfig.java | 475 +
.../jvm/backtype/storm/generated/LogLevel.java | 836 ++
.../storm/generated/LogLevelAction.java | 65 +
.../jvm/backtype/storm/generated/Nimbus.java | 8830 +++++++++++++-----
.../jvm/backtype/storm/generated/NodeInfo.java | 34 +-
.../storm/generated/RebalanceOptions.java | 46 +-
.../storm/generated/SpecificAggregateStats.java | 387 +
.../storm/generated/SpoutAggregateStats.java | 407 +
.../backtype/storm/generated/SpoutStats.java | 254 +-
.../jvm/backtype/storm/generated/StormBase.java | 94 +-
.../storm/generated/SupervisorInfo.java | 276 +-
.../storm/generated/SupervisorSummary.java | 168 +-
.../backtype/storm/generated/TopologyInfo.java | 162 +-
.../storm/generated/TopologyPageInfo.java | 2003 ++++
.../backtype/storm/generated/TopologyStats.java | 1094 +++
.../AbstractDNSToSwitchMapping.java | 95 +
.../networktopography/DNSToSwitchMapping.java | 50 +
.../DefaultRackDNSToSwitchMapping.java | 35 +
.../jvm/backtype/storm/nimbus/NimbusInfo.java | 8 +
.../jvm/backtype/storm/scheduler/Cluster.java | 46 +-
.../storm/scheduler/SupervisorDetails.java | 61 +-
.../backtype/storm/scheduler/Topologies.java | 15 +-
.../storm/scheduler/TopologyDetails.java | 330 +-
.../storm/scheduler/resource/Component.java | 54 +
.../storm/scheduler/resource/RAS_Node.java | 547 ++
.../resource/ResourceAwareScheduler.java | 149 +
.../storm/scheduler/resource/ResourceUtils.java | 133 +
.../resource/strategies/IStrategy.java | 37 +
.../strategies/ResourceAwareStrategy.java | 480 +
.../storm/security/auth/ReqContext.java | 7 +
.../auth/authorizer/SimpleACLAuthorizer.java | 15 +-
.../auth/kerberos/ServerCallbackHandler.java | 2 +
.../AlternateRackDNSToSwitchMapping.java | 48 +
.../topology/BaseConfigurationDeclarer.java | 31 +-
.../ComponentConfigurationDeclarer.java | 3 +
.../src/jvm/backtype/storm/tuple/Tuple.java | 8 +
.../src/jvm/backtype/storm/tuple/TupleImpl.java | 7 +-
.../backtype/storm/utils/DisruptorQueue.java | 11 +-
.../src/jvm/backtype/storm/utils/Utils.java | 21 +-
.../trident/topology/TridentBoltExecutor.java | 4 +-
storm-core/src/py/storm/Nimbus-remote | 28 +
storm-core/src/py/storm/Nimbus.py | 1451 ++-
storm-core/src/py/storm/ttypes.py | 2577 ++++-
storm-core/src/storm.thrift | 127 +
storm-core/src/ui/public/css/style.css | 8 +
storm-core/src/ui/public/js/script.js | 20 +
storm-core/src/ui/public/js/visualization.js | 92 +-
.../templates/component-page-template.html | 2 +-
.../templates/topology-page-template.html | 120 +-
storm-core/src/ui/public/topology.html | 149 +-
.../test/clj/backtype/storm/cluster_test.clj | 4 +-
.../test/clj/backtype/storm/nimbus_test.clj | 56 +-
.../scheduler/multitenant_scheduler_test.clj | 34 +-
.../scheduler/resource_aware_scheduler_test.clj | 669 ++
.../test/clj/backtype/storm/scheduler_test.clj | 3 +-
.../auth/DefaultHttpCredentialsPlugin_test.clj | 40 +-
.../test/clj/backtype/storm/supervisor_test.clj | 14 +
.../test/clj/backtype/storm/worker_test.clj | 179 +-
.../backtype/storm/utils/MockTupleHelpers.java | 40 +
115 files changed, 27909 insertions(+), 4937 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/47f4a3c3/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
----------------------------------------------------------------------
diff --cc external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
index ed36fbe,10405fa..980ed8b
--- a/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
@@@ -135,23 -135,16 +135,23 @@@ public class PartitionManager
if (toEmit == null) {
return EmitState.NO_EMITTED;
}
-
- Iterable<List<Object>> tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg);
++
+ Iterable<List<Object>> tups;
+ if (_spoutConfig.tupleMetaData) {
+ tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg, _partition, toEmit.offset);
+ } else {
+ tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg);
+ }
+
- if (tups != null) {
- if(_spoutConfig.topicAsStreamId) {
- for (List<Object> tup : tups) {
- collector.emit(_spoutConfig.topic, tup, new KafkaMessageId(_partition, toEmit.offset));
- }
- } else {
- for (List<Object> tup : tups) {
- collector.emit(tup, new KafkaMessageId(_partition, toEmit.offset));
- }
+ if ((tups != null) && tups.iterator().hasNext()) {
+ if(_spoutConfig.topicAsStreamId) {
+ for (List<Object> tup : tups) {
+ collector.emit(_spoutConfig.topic, tup, new KafkaMessageId(_partition, toEmit.offset));
+ }
+ } else {
+ for (List<Object> tup : tups) {
+ collector.emit(tup, new KafkaMessageId(_partition, toEmit.offset));
+ }
}
break;
} else {
[13/14] storm git commit: Merge branch 'STORM-697' of
https://github.com/mjtieman/storm into STORM-697
Posted by ka...@apache.org.
Merge branch 'STORM-697' of https://github.com/mjtieman/storm into STORM-697
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/81f6355a
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/81f6355a
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/81f6355a
Branch: refs/heads/master
Commit: 81f6355afed7f258605e8c137ffee332d36b2b71
Parents: 437c4b1 c0c830c
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Wed Oct 28 23:10:33 2015 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Wed Oct 28 23:10:33 2015 +0900
----------------------------------------------------------------------
external/storm-kafka/README.md | 44 ++++++++++-------
.../src/jvm/storm/kafka/KafkaConfig.java | 3 +-
.../src/jvm/storm/kafka/KafkaUtils.java | 8 +++
.../jvm/storm/kafka/MessageMetadataScheme.java | 25 ++++++++++
.../MessageMetadataSchemeAsMultiScheme.java | 40 +++++++++++++++
.../src/jvm/storm/kafka/PartitionManager.java | 13 ++++-
.../kafka/StringMessageAndMetadataScheme.java | 42 ++++++++++++++++
.../kafka/trident/TridentKafkaEmitter.java | 14 ++++--
.../src/test/storm/kafka/KafkaUtilsTest.java | 52 ++++++++++++++------
9 files changed, 203 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
[05/14] storm git commit: STORM-697: Fixed incorrect typing for offset
Posted by ka...@apache.org.
STORM-697: Fixed incorrect typing for offset
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/11194653
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/11194653
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/11194653
Branch: refs/heads/master
Commit: 11194653f43ce0d4f65d0051270cd86f2191cbc5
Parents: 2f119c6
Author: matt.tieman <ma...@inin.com>
Authored: Tue Mar 3 16:42:31 2015 -0500
Committer: matt.tieman <ma...@inin.com>
Committed: Tue Mar 3 16:42:31 2015 -0500
----------------------------------------------------------------------
external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java | 2 +-
.../src/jvm/storm/kafka/MessageMetadataScheme.java | 2 +-
.../jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java | 2 +-
.../storm-kafka/src/jvm/storm/kafka/PartitionManager.java | 9 ++++++++-
.../src/jvm/storm/kafka/StringMessageAndMetadataScheme.java | 2 +-
.../storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java | 8 ++++----
6 files changed, 16 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/11194653/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java b/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
index 9af49fe..17d0fb7 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
@@ -211,7 +211,7 @@ public class KafkaUtils {
return tups;
}
- public static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg, Partition partition, int offset) {
+ public static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg, Partition partition, long offset) {
Iterable<List<Object>> tups;
ByteBuffer payload = msg.payload();
if (payload == null) {
http://git-wip-us.apache.org/repos/asf/storm/blob/11194653/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
index d0dd2be..da7acbf 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
@@ -21,5 +21,5 @@ import backtype.storm.spout.Scheme;
* limitations under the License.
*/
public interface MessageMetadataScheme extends Scheme {
- public List<Object> deserializeMessageWithMetadata(byte[] message, Partition partition, int offset);
+ public List<Object> deserializeMessageWithMetadata(byte[] message, Partition partition, long offset);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/11194653/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
index 6226676..5eb20b5 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
@@ -14,7 +14,7 @@ public class MessageMetadataSchemeAsMultiScheme extends SchemeAsMultiScheme {
}
@SuppressWarnings("unchecked")
- public Iterable<List<Object>> deserializeMessageWithMetadata(byte[] message, Partition partition, int offset) {
+ public Iterable<List<Object>> deserializeMessageWithMetadata(byte[] message, Partition partition, long offset) {
List<Object> o = ((MessageMetadataScheme) scheme).deserializeMessageWithMetadata(message, partition, offset);
if (o == null) {
return null;
http://git-wip-us.apache.org/repos/asf/storm/blob/11194653/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java b/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
index 63e70cf..e1186e3 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java
@@ -134,7 +134,14 @@ public class PartitionManager {
if (toEmit == null) {
return EmitState.NO_EMITTED;
}
- Iterable<List<Object>> tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg);
+
+ Iterable<List<Object>> tups;
+ if (_spoutConfig.tupleMetaData) {
+ tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg, _partition, toEmit.offset);
+ } else {
+ tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg);
+ }
+
if (tups != null) {
for (List<Object> tup : tups) {
collector.emit(tup, new KafkaMessageId(_partition, toEmit.offset));
http://git-wip-us.apache.org/repos/asf/storm/blob/11194653/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java b/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
index 262a27c..2dc4c02 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
@@ -12,7 +12,7 @@ public class StringMessageAndMetadataScheme extends StringScheme implements Mess
public static final String STRING_SCHEME_OFFSET = "offset";
@Override
- public List<Object> deserializeMessageWithMetadata(byte[] message, Partition partition, int offset) {
+ public List<Object> deserializeMessageWithMetadata(byte[] message, Partition partition, long offset) {
String stringMessage = StringScheme.deserializeString(message);
return new Values(stringMessage, partition.partition, offset);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/11194653/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
index a7c9b2b..362d721 100644
--- a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
+++ b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
@@ -176,7 +176,7 @@ public class KafkaUtilsTest {
String value = "value";
Partition mockPartition = Mockito.mock(Partition.class);
mockPartition.partition = 0;
- int offset = 0;
+ long offset = 0L;
config.scheme = new MessageMetadataSchemeAsMultiScheme(new StringMessageAndMetadataScheme());
config.tupleMetaData = true;
@@ -187,8 +187,8 @@ public class KafkaUtilsTest {
Iterable<List<Object>> lists = KafkaUtils.generateTuples(config, msg.message(), mockPartition, offset);
List<Object> values = lists.iterator().next();
assertEquals("Message is incorrect", value, values.get(0));
- assertEquals("Offset is incorrect", offset, values.get(1));
- assertEquals("Partition is incorrect", mockPartition.partition, values.get(2));
+ assertEquals("Partition is incorrect", mockPartition.partition, values.get(1));
+ assertEquals("Offset is incorrect", offset, values.get(2));
}
}
@@ -197,7 +197,7 @@ public class KafkaUtilsTest {
String value = "value";
Partition mockPartition = Mockito.mock(Partition.class);
mockPartition.partition = 0;
- int offset = 0;
+ Long offset = 0L;
config.scheme = new SchemeAsMultiScheme(new StringScheme());
config.tupleMetaData = true;
[12/14] storm git commit: STORM-697: Review feedback: Fixed missing
or misplaced licenses. Added a more verbose explaination of
MessageMetadataSchemeAsMultiScheme in the README
Posted by ka...@apache.org.
STORM-697: Review feedback: Fixed missing or misplaced licenses. Added a more verbose explaination of MessageMetadataSchemeAsMultiScheme in the README
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/c0c830c1
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c0c830c1
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c0c830c1
Branch: refs/heads/master
Commit: c0c830c1e41059a39a61b6563d77ecab5f333186
Parents: 25e7bc4
Author: matt.tieman <ma...@inin.com>
Authored: Wed Oct 28 09:10:53 2015 -0400
Committer: matt.tieman <ma...@inin.com>
Committed: Wed Oct 28 09:10:53 2015 -0400
----------------------------------------------------------------------
external/storm-kafka/README.md | 44 ++++++++++++--------
.../jvm/storm/kafka/MessageMetadataScheme.java | 10 ++---
.../MessageMetadataSchemeAsMultiScheme.java | 17 ++++++++
.../kafka/StringMessageAndMetadataScheme.java | 17 ++++++++
4 files changed, 65 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/c0c830c1/external/storm-kafka/README.md
----------------------------------------------------------------------
diff --git a/external/storm-kafka/README.md b/external/storm-kafka/README.md
index 1bf14b7..04fb96c 100644
--- a/external/storm-kafka/README.md
+++ b/external/storm-kafka/README.md
@@ -6,16 +6,16 @@ Provides core Storm and Trident spout implementations for consuming data from Ap
##Spouts
We support both Trident and core Storm spouts. For both spout implementations, we use a BrokerHost interface that
tracks Kafka broker host to partition mapping and kafkaConfig that controls some Kafka related parameters.
-
+
###BrokerHosts
-In order to initialize your Kafka spout/emitter you need to construct an instance of the marker interface BrokerHosts.
+In order to initialize your Kafka spout/emitter you need to construct an instance of the marker interface BrokerHosts.
Currently, we support the following two implementations:
####ZkHosts
-ZkHosts is what you should use if you want to dynamically track Kafka broker to partition mapping. This class uses
+ZkHosts is what you should use if you want to dynamically track Kafka broker to partition mapping. This class uses
Kafka's ZooKeeper entries to track brokerHost -> partition mapping. You can instantiate an object by calling
```java
- public ZkHosts(String brokerZkStr, String brokerZkPath)
+ public ZkHosts(String brokerZkStr, String brokerZkPath)
public ZkHosts(String brokerZkStr)
```
Where brokerZkStr is just ip:port (e.g. localhost:2181). brokerZkPath is the root directory under which all the topics and
@@ -40,7 +40,7 @@ of this class, you need to first construct an instance of GlobalPartitionInforma
```
###KafkaConfig
-The second thing needed for constructing a kafkaSpout is an instance of KafkaConfig.
+The second thing needed for constructing a kafkaSpout is an instance of KafkaConfig.
```java
public KafkaConfig(BrokerHosts hosts, String topic)
public KafkaConfig(BrokerHosts hosts, String topic, String clientId)
@@ -103,9 +103,17 @@ also controls the naming of your output field.
public Fields getOutputFields();
```
-The default `RawMultiScheme` just takes the `byte[]` and returns a tuple with `byte[]` as is. The name of the
-outputField is "bytes". There are alternative implementations like `SchemeAsMultiScheme`,
-`KeyValueSchemeAsMultiScheme`, and `MessageMetadataSchemeAsMultiScheme` which can convert the `byte[]` to `String`.
+The default `RawMultiScheme` just takes the `byte[]` and returns a tuple with `byte[]` as is. The name of the outputField is "bytes". There are alternative implementations like `SchemeAsMultiScheme` and `KeyValueSchemeAsMultiScheme` which can convert the `byte[]` to `String`.
+
+There is also an extension of `SchemeAsMultiScheme`, `MessageMetadataSchemeAsMultiScheme`,
+which has an additional deserialize method that accepts the message `byte[]` in addition to the `Partition` and `offset` associated with the message.
+
+```java
+public Iterable<List<Object>> deserializeMessageWithMetadata(byte[] message, Partition partition, long offset)
+
+```
+
+This is useful for auditing/replaying messages from arbitrary points on a Kafka topic, saving the partition and offset of each message of a discrete stream instead of persisting the entire message.
### Examples
@@ -184,7 +192,7 @@ use Kafka 0.8.1.1 built against Scala 2.10, you would use the following dependen
Note that the ZooKeeper and log4j dependencies are excluded to prevent version conflicts with Storm's dependencies.
##Writing to Kafka as part of your topology
-You can create an instance of storm.kafka.bolt.KafkaBolt and attach it as a component to your topology or if you
+You can create an instance of storm.kafka.bolt.KafkaBolt and attach it as a component to your topology or if you
are using trident you can use storm.kafka.trident.TridentState, storm.kafka.trident.TridentStateFactory and
storm.kafka.trident.TridentKafkaUpdater.
@@ -199,9 +207,9 @@ These interfaces have 2 methods defined:
```
As the name suggests, these methods are called to map a tuple to Kafka key and Kafka message. If you just want one field
-as key and one field as value, then you can use the provided FieldNameBasedTupleToKafkaMapper.java
-implementation. In the KafkaBolt, the implementation always looks for a field with field name "key" and "message" if you
-use the default constructor to construct FieldNameBasedTupleToKafkaMapper for backward compatibility
+as key and one field as value, then you can use the provided FieldNameBasedTupleToKafkaMapper.java
+implementation. In the KafkaBolt, the implementation always looks for a field with field name "key" and "message" if you
+use the default constructor to construct FieldNameBasedTupleToKafkaMapper for backward compatibility
reasons. Alternatively you could also specify a different key and message field by using the non default constructor.
In the TridentKafkaState you must specify what is the field name for key and message as there is no default constructor.
These should be specified while constructing and instance of FieldNameBasedTupleToKafkaMapper.
@@ -213,12 +221,12 @@ public interface KafkaTopicSelector {
String getTopics(Tuple/TridentTuple tuple);
}
```
-The implementation of this interface should return the topic to which the tuple's key/message mapping needs to be published
-You can return a null and the message will be ignored. If you have one static topic name then you can use
+The implementation of this interface should return the topic to which the tuple's key/message mapping needs to be published
+You can return a null and the message will be ignored. If you have one static topic name then you can use
DefaultTopicSelector.java and set the name of the topic in the constructor.
### Specifying Kafka producer properties
-You can provide all the produce properties , see http://kafka.apache.org/documentation.html#newproducerconfigs
+You can provide all the produce properties , see http://kafka.apache.org/documentation.html#newproducerconfigs
section "Important configuration properties for the producer", in your Storm topology config by setting the properties
map with key kafka.broker.properties.
@@ -227,7 +235,7 @@ map with key kafka.broker.properties.
For the bolt :
```java
TopologyBuilder builder = new TopologyBuilder();
-
+
Fields fields = new Fields("key", "message");
FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
new Values("storm", "1"),
@@ -241,7 +249,7 @@ For the bolt :
.withTopicSelector(new DefaultTopicSelector("test"))
.withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper());
builder.setBolt("forwardToKafka", bolt, 8).shuffleGrouping("spout");
-
+
Config conf = new Config();
//set producer properties.
Properties props = new Properties();
@@ -250,7 +258,7 @@ For the bolt :
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
conf.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props);
-
+
StormSubmitter.submitTopology("kafkaboltTest", conf, builder.createTopology());
```
http://git-wip-us.apache.org/repos/asf/storm/blob/c0c830c1/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
index da7acbf..92a5598 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataScheme.java
@@ -1,8 +1,3 @@
-package storm.kafka;
-
-import java.util.List;
-import backtype.storm.spout.Scheme;
-
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -20,6 +15,11 @@ import backtype.storm.spout.Scheme;
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package storm.kafka;
+
+import java.util.List;
+import backtype.storm.spout.Scheme;
+
public interface MessageMetadataScheme extends Scheme {
public List<Object> deserializeMessageWithMetadata(byte[] message, Partition partition, long offset);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/c0c830c1/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
index e89e391..0567809 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/MessageMetadataSchemeAsMultiScheme.java
@@ -1,3 +1,20 @@
+/**
+ * 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 storm.kafka;
import java.util.Arrays;
http://git-wip-us.apache.org/repos/asf/storm/blob/c0c830c1/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java b/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
index 2dc4c02..031d497 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/StringMessageAndMetadataScheme.java
@@ -1,3 +1,20 @@
+/**
+ * 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 storm.kafka;
import java.util.List;