You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2016/05/20 19:46:38 UTC
[1/5] storm git commit: Addresses network flood from KafkaSpout to
kafka server.
Repository: storm
Updated Branches:
refs/heads/1.x-branch 068d5c9c2 -> f069cfb08
Addresses network flood from KafkaSpout to kafka server.
* Allows minBytes in fetch request to be configured
from KafkaConfig.fetchMinBytes.
* Defaults new configuration KafkaConfig.fetchMinBytes to 1.
* Defaults fetchMaxWait to 100ms instead of 10000ms.
Discovered 30 megabits of traffic flowing between a set of KafkaSpouts
and our kafka servers even though no Kafka messages were moving.
Using the wireshark kafka dissector, we were able to see that
each FetchRequest had maxWait set to 10000
and minBytes set to 0. When binBytes is set to 0 the kafka server
responds immediately when there are no messages. In turn the KafkaSpout
polls without any delay causing a constant stream of FetchRequest/
FetchResponse messages. Using a non-KafkaSpout client had a similar
traffic pattern with two key differences
1) minBytes was 1
2) maxWait was 100
With these FetchRequest parameters and no messages flowing,
the kafka server delays the FetchResponse by 100 ms. This reduces
the network traffic from megabits to the low kilobits. It also
reduced the CPU utilization of our kafka server from 140% to 2%.
Hopefully the risk of this change is low because
the old behavior can be restored using the API by setting
KafkaConfig.fetchMaxWait to 10000
KafkaConfig.fetchMinBytes to 0
Conflicts:
external/storm-kafka/src/jvm/storm/kafka/KafkaConfig.java
external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/ff43309e
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/ff43309e
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/ff43309e
Branch: refs/heads/1.x-branch
Commit: ff43309e39fb1db2bf2ae5a3d7d0972440880dca
Parents: fe50a1b
Author: Robert Hastings <rh...@moesol.com>
Authored: Tue May 3 13:40:56 2016 -0700
Committer: Robert Hastings <rh...@moesol.com>
Committed: Tue May 3 13:40:56 2016 -0700
----------------------------------------------------------------------
.../storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java | 3 ++-
.../storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java | 2 +-
2 files changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/ff43309e/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
index e1e1d24..77c7748 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
@@ -31,7 +31,8 @@ public class KafkaConfig implements Serializable {
public int fetchSizeBytes = 1024 * 1024;
public int socketTimeoutMs = 10000;
- public int fetchMaxWait = 10000;
+ public int fetchMaxWait = 100;
+ public int fetchMinBytes = 1;
public int bufferSizeBytes = 1024 * 1024;
public MultiScheme scheme = new RawMultiScheme();
public boolean ignoreZkOffsets = false;
http://git-wip-us.apache.org/repos/asf/storm/blob/ff43309e/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
index a2be825..47afbad 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
@@ -185,7 +185,7 @@ public class KafkaUtils {
int partitionId = partition.partition;
FetchRequestBuilder builder = new FetchRequestBuilder();
FetchRequest fetchRequest = builder.addFetch(topic, partitionId, offset, config.fetchSizeBytes).
- clientId(config.clientId).maxWait(config.fetchMaxWait).build();
+ clientId(config.clientId).maxWait(config.fetchMaxWait).minBytes(config.fetchMinBytes).build();
FetchResponse fetchResponse;
try {
fetchResponse = consumer.fetch(fetchRequest);
[2/5] storm git commit: Merge remote-tracking branch
'apache/1.x-branch' into 1.x-branch
Posted by pt...@apache.org.
Merge remote-tracking branch 'apache/1.x-branch' into 1.x-branch
Conflicts:
external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d936ad78
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d936ad78
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d936ad78
Branch: refs/heads/1.x-branch
Commit: d936ad78169283fcd8e0b8923eefc4063b59f074
Parents: ff43309 0287d9b
Author: Robert Hastings <rh...@moesol.com>
Authored: Tue May 3 14:19:21 2016 -0700
Committer: Robert Hastings <rh...@moesol.com>
Committed: Tue May 3 14:19:21 2016 -0700
----------------------------------------------------------------------
CHANGELOG.md | 34 ++
bin/storm.py | 1 +
conf/defaults.yaml | 2 +-
conf/storm-env.sh | 2 +-
docs/Daemon-Fault-Tolerance.md | 2 +-
docs/Eventlogging.md | 96 ++++
docs/Setting-up-a-Storm-cluster.md | 2 +-
docs/Storm-Scheduler.md | 27 +
docs/_includes/footer.html | 2 +-
docs/flux.md | 6 +-
docs/images/disable-event-logging-topology.png | Bin 0 -> 128737 bytes
docs/images/enable-event-logging-spout.png | Bin 0 -> 136954 bytes
docs/images/enable-event-logging-topology.png | Bin 0 -> 161055 bytes
docs/images/event-logs-view.png | Bin 0 -> 192749 bytes
docs/index.md | 20 +-
examples/storm-starter/pom.xml | 24 +-
external/flux/README.md | 6 +-
external/flux/flux-core/pom.xml | 2 +-
.../java/org/apache/storm/flux/FluxBuilder.java | 29 +-
.../src/test/resources/configs/kafka_test.yaml | 2 +-
external/flux/flux-examples/pom.xml | 15 +-
.../src/main/resources/kafka_spout.yaml | 2 +-
external/flux/flux-wrappers/pom.xml | 2 +-
external/flux/pom.xml | 15 +-
external/sql/pom.xml | 2 +-
external/sql/storm-sql-core/pom.xml | 2 +-
.../storm-sql-core/src/codegen/data/Parser.tdd | 4 +-
.../src/codegen/includes/parserImpls.ftl | 25 +
.../jvm/org/apache/storm/sql/StormSqlImpl.java | 44 +-
.../apache/storm/sql/compiler/ExprCompiler.java | 26 +-
.../storm/sql/parser/SqlCreateFunction.java | 94 ++++
.../test/org/apache/storm/sql/TestStormSql.java | 67 ++-
.../storm/sql/compiler/TestCompilerUtils.java | 15 +-
.../backends/standalone/TestPlanCompiler.java | 16 +
.../apache/storm/sql/parser/TestSqlParser.java | 6 +
external/sql/storm-sql-kafka/pom.xml | 18 +-
external/sql/storm-sql-runtime/pom.xml | 2 +-
.../test/org/apache/storm/sql/TestUtils.java | 6 +
external/storm-cassandra/pom.xml | 2 +-
external/storm-elasticsearch/pom.xml | 5 +-
external/storm-eventhubs/pom.xml | 226 ++++----
external/storm-hbase/pom.xml | 3 +-
.../trident/windowing/HBaseWindowsStore.java | 46 +-
.../windowing/HBaseWindowsStoreFactory.java | 4 +-
external/storm-hdfs/pom.xml | 2 +-
.../hdfs/common/security/HdfsSecurityUtil.java | 24 +-
external/storm-hive/pom.xml | 5 +-
external/storm-jdbc/pom.xml | 4 +-
external/storm-kafka-client/README.md | 109 ++++
external/storm-kafka-client/pom.xml | 85 +++
.../apache/storm/kafka/spout/KafkaSpout.java | 547 +++++++++++++++++++
.../storm/kafka/spout/KafkaSpoutConfig.java | 309 +++++++++++
.../storm/kafka/spout/KafkaSpoutMessageId.java | 101 ++++
.../KafkaSpoutRetryExponentialBackoff.java | 281 ++++++++++
.../kafka/spout/KafkaSpoutRetryService.java | 72 +++
.../storm/kafka/spout/KafkaSpoutStream.java | 70 +++
.../storm/kafka/spout/KafkaSpoutStreams.java | 158 ++++++
.../kafka/spout/KafkaSpoutTupleBuilder.java | 58 ++
.../kafka/spout/KafkaSpoutTuplesBuilder.java | 82 +++
.../kafka/spout/test/KafkaSpoutTestBolt.java | 50 ++
.../spout/test/KafkaSpoutTopologyMain.java | 133 +++++
.../spout/test/TopicTest2TupleBuilder.java | 40 ++
.../test/TopicsTest0Test1TupleBuilder.java | 42 ++
external/storm-kafka/pom.xml | 18 +-
.../jvm/org/apache/storm/kafka/KafkaConfig.java | 8 +-
external/storm-metrics/pom.xml | 21 +-
external/storm-mongodb/pom.xml | 148 ++---
external/storm-mqtt/core/pom.xml | 5 +-
external/storm-mqtt/examples/pom.xml | 5 +-
external/storm-mqtt/pom.xml | 5 +-
external/storm-redis/pom.xml | 2 +-
external/storm-solr/pom.xml | 14 +-
.../apache/storm/solr/bolt/SolrUpdateBolt.java | 23 +-
log4j2/worker.xml | 2 +-
pom.xml | 63 ++-
.../maven-shade-clojure-transformer/pom.xml | 2 +-
storm-buildtools/storm-maven-plugins/pom.xml | 5 +-
storm-core/pom.xml | 6 +-
.../clj/org/apache/storm/daemon/executor.clj | 52 +-
.../clj/org/apache/storm/daemon/logviewer.clj | 84 ++-
.../clj/org/apache/storm/daemon/supervisor.clj | 11 +-
.../src/clj/org/apache/storm/daemon/worker.clj | 19 +-
storm-core/src/clj/org/apache/storm/stats.clj | 39 +-
storm-core/src/clj/org/apache/storm/ui/core.clj | 12 +-
.../src/dev/logviewer-search-context-tests.log | 1 -
.../dev/logviewer-search-context-tests.log.test | 1 +
storm-core/src/dev/small-worker.log | 1 -
storm-core/src/dev/small-worker.log.test | 1 +
storm-core/src/dev/test-3072.log | 3 -
storm-core/src/dev/test-3072.log.test | 3 +
storm-core/src/dev/test-worker.log | 380 -------------
storm-core/src/dev/test-worker.log.test | 380 +++++++++++++
.../apache/storm/daemon/DirectoryCleaner.java | 2 +-
.../storm/metric/internal/RateTracker.java | 26 -
.../DefaultResourceAwareStrategy.java | 9 +-
.../topology/BaseStatefulBoltExecutor.java | 209 +++++++
.../topology/CheckpointTupleForwarder.java | 159 +-----
.../apache/storm/topology/IStatefulBolt.java | 19 +-
.../storm/topology/StatefulBoltExecutor.java | 22 +-
.../apache/storm/topology/TopologyBuilder.java | 48 ++
.../windowing/InMemoryWindowsStoreFactory.java | 6 +-
.../trident/windowing/WindowKryoSerializer.java | 87 +++
.../windowing/WindowTridentProcessor.java | 8 +-
.../trident/windowing/WindowsStateUpdater.java | 2 +-
.../storm/trident/windowing/WindowsStore.java | 1 +
.../trident/windowing/WindowsStoreFactory.java | 9 +-
.../org/apache/storm/utils/DisruptorQueue.java | 24 +-
.../org/apache/storm/utils/VersionedStore.java | 3 +-
storm-core/src/ui/public/component.html | 18 +-
storm-core/src/ui/public/js/script.js | 10 +-
storm-core/src/ui/public/logviewer_search.html | 4 +-
.../templates/component-page-template.html | 8 +-
.../logviewer-search-page-template.html | 5 +-
.../templates/topology-page-template.html | 8 +-
storm-core/src/ui/public/topology.html | 2 +-
.../clj/org/apache/storm/logviewer_test.clj | 97 +++-
.../org/apache/storm/versioned_store_test.clj | 11 +-
.../resource/TestResourceAwareScheduler.java | 61 +++
storm-dist/binary/pom.xml | 2 +-
storm-dist/binary/src/main/assembly/binary.xml | 26 +
storm-dist/source/pom.xml | 2 +-
storm-multilang/javascript/pom.xml | 5 +-
storm-multilang/python/pom.xml | 2 +-
storm-multilang/ruby/pom.xml | 2 +-
storm-rename-hack/pom.xml | 8 +-
125 files changed, 4124 insertions(+), 1171 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d936ad78/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
----------------------------------------------------------------------
diff --cc external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
index 77c7748,c845531..7c55550
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
@@@ -31,8 -33,7 +33,8 @@@ public class KafkaConfig implements Ser
public int fetchSizeBytes = 1024 * 1024;
public int socketTimeoutMs = 10000;
- public int fetchMaxWait = 100;
- public int fetchMinBytes = 1;
- public int fetchMaxWait = 10000;
++ public int fetchMaxWait = FetchRequest.DefaultMaxWait();
++ public int fetchMinBytes = FetchRequest.DefaultMinBytes();
public int bufferSizeBytes = 1024 * 1024;
public MultiScheme scheme = new RawMultiScheme();
public boolean ignoreZkOffsets = false;
[3/5] storm git commit: Merge branch '1.x-branch' of
github.com:MoebiusSolutions/storm into 1.x-branch
Posted by pt...@apache.org.
Merge branch '1.x-branch' of github.com:MoebiusSolutions/storm into 1.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/afe08fc4
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/afe08fc4
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/afe08fc4
Branch: refs/heads/1.x-branch
Commit: afe08fc442139b8eb31761f588309e6c9c75b1dc
Parents: 068d5c9 d936ad7
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Fri May 20 14:32:20 2016 -0400
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Fri May 20 14:32:20 2016 -0400
----------------------------------------------------------------------
.../storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java | 4 ++--
.../storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java | 2 +-
2 files changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
[4/5] storm git commit: add STORM-1674 to changelog
Posted by pt...@apache.org.
add STORM-1674 to changelog
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/dd15981b
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/dd15981b
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/dd15981b
Branch: refs/heads/1.x-branch
Commit: dd15981bfa729bd1cee5709e0a9e994add50bfd6
Parents: afe08fc
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Fri May 20 14:38:32 2016 -0400
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Fri May 20 14:38:32 2016 -0400
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/dd15981b/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a315b1b..e37bf56 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.1.0
+ * STORM-1674: Idle KafkaSpout consumes more bandwidth than needed
* STORM-1842: Forward references in storm.thrift cause tooling issues
* STORM-1730: LocalCluster#shutdown() does not terminate all storm threads/thread pools.
* STORM-1841: Address a few minor issues in windowing and doc
[5/5] storm git commit: STORM-1674: rename KafkaConfig.fetchMinBytes
to KafkaConfig.minFetchByte to retain backward compatibility with 1.0.x
versions
Posted by pt...@apache.org.
STORM-1674: rename KafkaConfig.fetchMinBytes to KafkaConfig.minFetchByte to retain backward compatibility with 1.0.x versions
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/f069cfb0
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/f069cfb0
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/f069cfb0
Branch: refs/heads/1.x-branch
Commit: f069cfb08099a54448bba3f96f20c5b929ea6af3
Parents: dd15981
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Fri May 20 15:24:00 2016 -0400
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Fri May 20 15:45:21 2016 -0400
----------------------------------------------------------------------
.../storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java | 3 ++-
.../storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java | 2 +-
2 files changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/f069cfb0/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
index 7c55550..605e7fd 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
@@ -34,7 +34,6 @@ public class KafkaConfig implements Serializable {
public int fetchSizeBytes = 1024 * 1024;
public int socketTimeoutMs = 10000;
public int fetchMaxWait = FetchRequest.DefaultMaxWait();
- public int fetchMinBytes = FetchRequest.DefaultMinBytes();
public int bufferSizeBytes = 1024 * 1024;
public MultiScheme scheme = new RawMultiScheme();
public boolean ignoreZkOffsets = false;
@@ -42,6 +41,8 @@ public class KafkaConfig implements Serializable {
public long maxOffsetBehind = Long.MAX_VALUE;
public boolean useStartOffsetTimeIfOffsetOutOfRange = true;
public int metricsTimeBucketSizeInSecs = 60;
+ public int minFetchByte = FetchRequest.DefaultMinBytes();
+
public KafkaConfig(BrokerHosts hosts, String topic) {
this(hosts, topic, kafka.api.OffsetRequest.DefaultClientId());
http://git-wip-us.apache.org/repos/asf/storm/blob/f069cfb0/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
index 47afbad..48d3ab7 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
@@ -185,7 +185,7 @@ public class KafkaUtils {
int partitionId = partition.partition;
FetchRequestBuilder builder = new FetchRequestBuilder();
FetchRequest fetchRequest = builder.addFetch(topic, partitionId, offset, config.fetchSizeBytes).
- clientId(config.clientId).maxWait(config.fetchMaxWait).minBytes(config.fetchMinBytes).build();
+ clientId(config.clientId).maxWait(config.fetchMaxWait).minBytes(config.minFetchByte).build();
FetchResponse fetchResponse;
try {
fetchResponse = consumer.fetch(fetchRequest);