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/16 03:04:18 UTC
[1/8] storm git commit: STORM-1079. Batch Puts to HBase.
Repository: storm
Updated Branches:
refs/heads/master ebca03a90 -> 8f8c3e548
STORM-1079. Batch Puts to HBase.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/e3b5c270
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/e3b5c270
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/e3b5c270
Branch: refs/heads/master
Commit: e3b5c2701f1ce912dde052af3e81b9c4ce517f3f
Parents: ce93d5f
Author: Sriharsha Chintalapani <ma...@harsha.io>
Authored: Wed Sep 30 14:44:28 2015 -0700
Committer: Sriharsha Chintalapani <ma...@harsha.io>
Committed: Wed Sep 30 14:44:28 2015 -0700
----------------------------------------------------------------------
.../storm/hbase/bolt/AbstractHBaseBolt.java | 10 +++
.../org/apache/storm/hbase/bolt/HBaseBolt.java | 65 +++++++++++++++++---
2 files changed, 67 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/e3b5c270/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
index d814117..aad3d88 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
@@ -42,6 +42,8 @@ public abstract class AbstractHBaseBolt extends BaseRichBolt {
protected String tableName;
protected HBaseMapper mapper;
protected String configKey;
+ protected int batchSize = 15000;
+ protected int flushIntervalSecs = 0;
public AbstractHBaseBolt(String tableName, HBaseMapper mapper) {
Validate.notEmpty(tableName, "Table name can not be blank or null");
@@ -72,5 +74,13 @@ public abstract class AbstractHBaseBolt extends BaseRichBolt {
Map<String, Object> hbaseConfMap = new HashMap<String, Object>(conf);
hbaseConfMap.put(Config.TOPOLOGY_AUTO_CREDENTIALS, map.get(Config.TOPOLOGY_AUTO_CREDENTIALS));
this.hBaseClient = new HBaseClient(hbaseConfMap, hbConfig, tableName);
+
+ // If interval is non-zero then it has already been explicitly set and we should not default it
+ if (conf.containsKey("topology.message.timeout.secs") && flushIntervalSecs == 0)
+ {
+ Integer topologyTimeout = Integer.parseInt(conf.get("topology.message.timeout.secs").toString());
+ flushIntervalSecs = (int)(Math.floor(topologyTimeout / 2));
+ LOG.info("Setting tick tuple interval to [" + flushIntervalSecs + "] based on topology timeout");
+ }
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/e3b5c270/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
index cf29aa5..7eeca77 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
@@ -19,6 +19,8 @@ package org.apache.storm.hbase.bolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
+import backtype.storm.utils.TupleUtils;
+import backtype.storm.Config;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
@@ -26,7 +28,9 @@ import org.apache.storm.hbase.common.ColumnList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Map;
import java.util.List;
+import java.util.LinkedList;
/**
* Basic bolt for writing to HBase.
@@ -38,9 +42,13 @@ public class HBaseBolt extends AbstractHBaseBolt {
private static final Logger LOG = LoggerFactory.getLogger(HBaseBolt.class);
boolean writeToWAL = true;
+ List<Mutation> batchMutations;
+ List<Tuple> tupleBatch;
public HBaseBolt(String tableName, HBaseMapper mapper) {
super(tableName, mapper);
+ this.batchMutations = new LinkedList<>();
+ this.tupleBatch = new LinkedList<>();
}
public HBaseBolt writeToWAL(boolean writeToWAL) {
@@ -53,21 +61,62 @@ public class HBaseBolt extends AbstractHBaseBolt {
return this;
}
+ public HBaseBolt withBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ return this;
+ }
+
+ public HBaseBolt withFlushIntervalSecs(int flushIntervalSecs) {
+ this.flushIntervalSecs = flushIntervalSecs;
+ return this;
+ }
+
+ @Override
+ public Map<String, Object> getComponentConfiguration() {
+ Map<String, Object> conf = super.getComponentConfiguration();
+ if (conf == null)
+ conf = new Config();
+
+ if (flushIntervalSecs > 0) {
+ LOG.info("Enabling tick tuple with interval [" + flushIntervalSecs + "]");
+ conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, flushIntervalSecs);
+ }
+
+ return conf;
+ }
+
+
@Override
public void execute(Tuple tuple) {
- byte[] rowKey = this.mapper.rowKey(tuple);
- ColumnList cols = this.mapper.columns(tuple);
- List<Mutation> mutations = hBaseClient.constructMutationReq(rowKey, cols, writeToWAL? Durability.SYNC_WAL : Durability.SKIP_WAL);
+ boolean flush = false;
+ if (TupleUtils.isTick(tuple)) {
+ LOG.debug("TICK received! current batch status [" + tupleBatch.size() + "/" + batchSize + "]");
+ flush = true;
+ } else {
+ byte[] rowKey = this.mapper.rowKey(tuple);
+ ColumnList cols = this.mapper.columns(tuple);
+ List<Mutation> mutations = hBaseClient.constructMutationReq(rowKey, cols, writeToWAL? Durability.SYNC_WAL : Durability.SKIP_WAL);
+ batchMutations.addAll(mutations);
+ tupleBatch.add(tuple);
+ if (tupleBatch.size() >= batchSize)
+ flush = true;
+ }
try {
- this.hBaseClient.batchMutate(mutations);
+ if (flush && !tupleBatch.isEmpty()) {
+ this.hBaseClient.batchMutate(batchMutations);
+ LOG.debug("acknowledging tuples after batchMutate");
+ for(Tuple t : tupleBatch)
+ collector.ack(t);
+ }
} catch(Exception e){
this.collector.reportError(e);
- this.collector.fail(tuple);
- return;
+ for (Tuple t : tupleBatch)
+ collector.fail(t);
+ } finally {
+ tupleBatch.clear();
+ batchMutations.clear();
}
-
- this.collector.ack(tuple);
}
@Override
[8/8] storm git commit: add STORM-1079 to CHANGELOG.md
Posted by ka...@apache.org.
add STORM-1079 to CHANGELOG.md
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/8f8c3e54
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/8f8c3e54
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/8f8c3e54
Branch: refs/heads/master
Commit: 8f8c3e548596422d768ec4d0b151d2375794172b
Parents: a69e3ae
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Oct 16 10:04:01 2015 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Oct 16 10:04:01 2015 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/8f8c3e54/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 44587d0..b618e2b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 0.11.0
+ * STORM-1079: Batch Puts to HBase
* STORM-1084: Improve Storm config validation process to use java annotations instead of *_SCHEMA format
* STORM-1106: Netty should not limit attempts to reconnect
* STORM-1103: Changes log message to DEBUG from INFO
[2/8] storm git commit: STORM-1079. Batch Puts to HBase.
Posted by ka...@apache.org.
STORM-1079. Batch Puts to HBase.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/7710eef9
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/7710eef9
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/7710eef9
Branch: refs/heads/master
Commit: 7710eef9249249de2f8bbd5670a50cf6901f7b3e
Parents: e3b5c27
Author: Sriharsha Chintalapani <ma...@harsha.io>
Authored: Wed Sep 30 16:10:36 2015 -0700
Committer: Sriharsha Chintalapani <ma...@harsha.io>
Committed: Wed Sep 30 16:10:36 2015 -0700
----------------------------------------------------------------------
external/storm-hbase/README.md | 10 ++++++++++
.../org/apache/storm/hbase/bolt/AbstractHBaseBolt.java | 7 -------
2 files changed, 10 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/7710eef9/external/storm-hbase/README.md
----------------------------------------------------------------------
diff --git a/external/storm-hbase/README.md b/external/storm-hbase/README.md
index 81d351a..fd4d0ad 100644
--- a/external/storm-hbase/README.md
+++ b/external/storm-hbase/README.md
@@ -107,6 +107,16 @@ To use the `HBaseBolt`, construct it with the name of the table to write to, an
```java
HBaseBolt hbase = new HBaseBolt("WordCount", mapper);
```
+
+ HBaseBolt params
+
+|Arg |Description | Type | Default |
+|--- |--- |---
+|writeToWAL | To turn Durability SYNC_WAL or SKIP_WAL | Boolean (Optional) | True |
+|configKey | Any Hbase related configs | Map (Optional) | |
+|batchSize | Max no.of Tuples batched together to write to HBase | Int (Optional) | 15000 |
+|flushIntervalSecs| (In seconds) If > 0 HBase Bolt will periodically flush transaction batches. Enabling this is recommended to avoid tuple timeouts while waiting for a batch to fill up. | Int (Optional) | 0 |
+
The `HBaseBolt` will delegate to the `mapper` instance to figure out how to persist tuple data to HBase.
http://git-wip-us.apache.org/repos/asf/storm/blob/7710eef9/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
index aad3d88..3372192 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
@@ -75,12 +75,5 @@ public abstract class AbstractHBaseBolt extends BaseRichBolt {
hbaseConfMap.put(Config.TOPOLOGY_AUTO_CREDENTIALS, map.get(Config.TOPOLOGY_AUTO_CREDENTIALS));
this.hBaseClient = new HBaseClient(hbaseConfMap, hbConfig, tableName);
- // If interval is non-zero then it has already been explicitly set and we should not default it
- if (conf.containsKey("topology.message.timeout.secs") && flushIntervalSecs == 0)
- {
- Integer topologyTimeout = Integer.parseInt(conf.get("topology.message.timeout.secs").toString());
- flushIntervalSecs = (int)(Math.floor(topologyTimeout / 2));
- LOG.info("Setting tick tuple interval to [" + flushIntervalSecs + "] based on topology timeout");
- }
}
}
[3/8] storm git commit: STORM-1079. Batch Puts to HBase.
Posted by ka...@apache.org.
STORM-1079. Batch Puts to HBase.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/b53b9eb3
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/b53b9eb3
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/b53b9eb3
Branch: refs/heads/master
Commit: b53b9eb3c12f8caa7180c880ba52c550535ef192
Parents: 7710eef
Author: Sriharsha Chintalapani <ma...@harsha.io>
Authored: Sat Oct 3 08:06:57 2015 -0700
Committer: Sriharsha Chintalapani <ma...@harsha.io>
Committed: Sat Oct 3 08:06:57 2015 -0700
----------------------------------------------------------------------
.../org/apache/storm/hbase/bolt/AbstractHBaseBolt.java | 1 -
.../java/org/apache/storm/hbase/bolt/HBaseBolt.java | 12 ++++++++----
2 files changed, 8 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/b53b9eb3/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
index 3372192..404aa7a 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
@@ -74,6 +74,5 @@ public abstract class AbstractHBaseBolt extends BaseRichBolt {
Map<String, Object> hbaseConfMap = new HashMap<String, Object>(conf);
hbaseConfMap.put(Config.TOPOLOGY_AUTO_CREDENTIALS, map.get(Config.TOPOLOGY_AUTO_CREDENTIALS));
this.hBaseClient = new HBaseClient(hbaseConfMap, hbConfig, tableName);
-
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/b53b9eb3/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
index 7eeca77..4cdf388 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
@@ -74,8 +74,9 @@ public class HBaseBolt extends AbstractHBaseBolt {
@Override
public Map<String, Object> getComponentConfiguration() {
Map<String, Object> conf = super.getComponentConfiguration();
- if (conf == null)
+ if (conf == null) {
conf = new Config();
+ }
if (flushIntervalSecs > 0) {
LOG.info("Enabling tick tuple with interval [" + flushIntervalSecs + "]");
@@ -98,21 +99,24 @@ public class HBaseBolt extends AbstractHBaseBolt {
List<Mutation> mutations = hBaseClient.constructMutationReq(rowKey, cols, writeToWAL? Durability.SYNC_WAL : Durability.SKIP_WAL);
batchMutations.addAll(mutations);
tupleBatch.add(tuple);
- if (tupleBatch.size() >= batchSize)
+ if (tupleBatch.size() >= batchSize) {
flush = true;
+ }
}
try {
if (flush && !tupleBatch.isEmpty()) {
this.hBaseClient.batchMutate(batchMutations);
LOG.debug("acknowledging tuples after batchMutate");
- for(Tuple t : tupleBatch)
+ for(Tuple t : tupleBatch) {
collector.ack(t);
+ }
}
} catch(Exception e){
this.collector.reportError(e);
- for (Tuple t : tupleBatch)
+ for (Tuple t : tupleBatch) {
collector.fail(t);
+ }
} finally {
tupleBatch.clear();
batchMutations.clear();
[6/8] storm git commit: STORM-1079. Fixed reset of tupleBatch.
Posted by ka...@apache.org.
STORM-1079. Fixed reset of tupleBatch.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/99da29b4
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/99da29b4
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/99da29b4
Branch: refs/heads/master
Commit: 99da29b443c89580be34512b67d780c811d6d38a
Parents: fc03910
Author: Sriharsha Chintalapani <ma...@harsha.io>
Authored: Wed Oct 14 14:38:13 2015 -0700
Committer: Sriharsha Chintalapani <ma...@harsha.io>
Committed: Wed Oct 14 14:38:13 2015 -0700
----------------------------------------------------------------------
.../org/apache/storm/hbase/bolt/HBaseBolt.java | 27 ++++++++++----------
1 file changed, 14 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/99da29b4/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
index d470892..ea17b2c 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
@@ -93,34 +93,35 @@ public class HBaseBolt extends AbstractHBaseBolt {
@Override
public void execute(Tuple tuple) {
boolean flush = false;
- if (TupleUtils.isTick(tuple)) {
- LOG.debug("TICK received! current batch status [" + tupleBatch.size() + "/" + batchSize + "]");
- flush = true;
- } else {
- byte[] rowKey = this.mapper.rowKey(tuple);
- ColumnList cols = this.mapper.columns(tuple);
- List<Mutation> mutations = hBaseClient.constructMutationReq(rowKey, cols, writeToWAL? Durability.SYNC_WAL : Durability.SKIP_WAL);
- batchMutations.addAll(mutations);
- tupleBatch.add(tuple);
- if (tupleBatch.size() >= batchSize) {
+ try {
+ if (TupleUtils.isTick(tuple)) {
+ LOG.debug("TICK received! current batch status [" + tupleBatch.size() + "/" + batchSize + "]");
flush = true;
+ } else {
+ byte[] rowKey = this.mapper.rowKey(tuple);
+ ColumnList cols = this.mapper.columns(tuple);
+ List<Mutation> mutations = hBaseClient.constructMutationReq(rowKey, cols, writeToWAL? Durability.SYNC_WAL : Durability.SKIP_WAL);
+ batchMutations.addAll(mutations);
+ tupleBatch.add(tuple);
+ if (tupleBatch.size() >= batchSize) {
+ flush = true;
+ }
}
- }
- try {
if (flush && !tupleBatch.isEmpty()) {
this.hBaseClient.batchMutate(batchMutations);
LOG.debug("acknowledging tuples after batchMutate");
for(Tuple t : tupleBatch) {
collector.ack(t);
}
+ tupleBatch.clear();
+ batchMutations.clear();
}
} catch(Exception e){
this.collector.reportError(e);
for (Tuple t : tupleBatch) {
collector.fail(t);
}
- } finally {
tupleBatch.clear();
batchMutations.clear();
}
[5/8] storm git commit: STORM-1079. Fixed compilation erro and
Logging.
Posted by ka...@apache.org.
STORM-1079. Fixed compilation erro and Logging.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/fc039103
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/fc039103
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/fc039103
Branch: refs/heads/master
Commit: fc0391038c7d839487a2cafce55519d246974b13
Parents: 9a25ab7
Author: Sriharsha Chintalapani <ma...@harsha.io>
Authored: Mon Oct 12 07:08:09 2015 -0700
Committer: Sriharsha Chintalapani <ma...@harsha.io>
Committed: Mon Oct 12 07:10:07 2015 -0700
----------------------------------------------------------------------
.../src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/fc039103/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
index 34e2eba..d470892 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
@@ -78,13 +78,13 @@ public class HBaseBolt extends AbstractHBaseBolt {
conf = new Config();
}
- if (conf.containsKey("topology.message.timeout.secs") && tickTupleInterval == 0) {
+ if (conf.containsKey("topology.message.timeout.secs") && flushIntervalSecs == 0) {
Integer topologyTimeout = Integer.parseInt(conf.get("topology.message.timeout.secs").toString());
flushIntervalSecs = (int)(Math.floor(topologyTimeout / 2));
- LOG.debug("Setting flush interval to [" + flushIntervalSecs + "] based on topology.message.timeout.secs");
+ LOG.debug("Setting flush interval to [{}] based on topology.message.timeout.secs", flushIntervalSecs);
}
- LOG.info("Enabling tick tuple with interval [" + flushIntervalSecs + "]");
+ LOG.info("Enabling tick tuple with interval [{}]", flushIntervalSecs);
conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, flushIntervalSecs);
return conf;
}
[4/8] storm git commit: STORM-1079. Batch Puts to HBase. Add default
flushIntervalSecs.
Posted by ka...@apache.org.
STORM-1079. Batch Puts to HBase. Add default flushIntervalSecs.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9a25ab79
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9a25ab79
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9a25ab79
Branch: refs/heads/master
Commit: 9a25ab79b4457c4ed98b64c96218e0462efe72c4
Parents: b53b9eb
Author: Sriharsha Chintalapani <ma...@harsha.io>
Authored: Thu Oct 8 09:13:54 2015 -0700
Committer: Sriharsha Chintalapani <ma...@harsha.io>
Committed: Thu Oct 8 09:13:54 2015 -0700
----------------------------------------------------------------------
.../main/java/org/apache/storm/hbase/bolt/HBaseBolt.java | 9 ++++++---
1 file changed, 6 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/9a25ab79/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
index 4cdf388..34e2eba 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseBolt.java
@@ -78,11 +78,14 @@ public class HBaseBolt extends AbstractHBaseBolt {
conf = new Config();
}
- if (flushIntervalSecs > 0) {
- LOG.info("Enabling tick tuple with interval [" + flushIntervalSecs + "]");
- conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, flushIntervalSecs);
+ if (conf.containsKey("topology.message.timeout.secs") && tickTupleInterval == 0) {
+ Integer topologyTimeout = Integer.parseInt(conf.get("topology.message.timeout.secs").toString());
+ flushIntervalSecs = (int)(Math.floor(topologyTimeout / 2));
+ LOG.debug("Setting flush interval to [" + flushIntervalSecs + "] based on topology.message.timeout.secs");
}
+ LOG.info("Enabling tick tuple with interval [" + flushIntervalSecs + "]");
+ conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, flushIntervalSecs);
return conf;
}
[7/8] storm git commit: Merge branch 'STORM-1079' of
https://github.com/harshach/incubator-storm into STORM-1079
Posted by ka...@apache.org.
Merge branch 'STORM-1079' of https://github.com/harshach/incubator-storm into STORM-1079
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/a69e3ae2
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/a69e3ae2
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/a69e3ae2
Branch: refs/heads/master
Commit: a69e3ae268adc226146b91bec5fcd889f70d055c
Parents: ebca03a 99da29b
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Oct 16 09:19:01 2015 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Oct 16 09:19:01 2015 +0900
----------------------------------------------------------------------
external/storm-hbase/README.md | 10 +++
.../storm/hbase/bolt/AbstractHBaseBolt.java | 2 +
.../org/apache/storm/hbase/bolt/HBaseBolt.java | 75 +++++++++++++++++---
3 files changed, 78 insertions(+), 9 deletions(-)
----------------------------------------------------------------------