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 2014/08/08 21:48:41 UTC
[30/50] [abbrv] git commit: STORM-405: Add kafka trident state so
messages can be sent to kafka topics. Added a mapper interface so users can
specify there own implementation that converts tuples into kafka keys and
messages. Also added a topic selector
STORM-405: Add kafka trident state so messages can be sent to kafka topics. Added a mapper interface so users can specify there own implementation that converts tuples into kafka keys and messages.
Also added a topic selector interface so users can provide custom implmentation that can chose what topics a tuple should be forwarded to.
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/827e0ba9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/827e0ba9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/827e0ba9
Branch: refs/heads/master
Commit: 827e0ba9e88f5a719532f2cf55a3963ef0bfb6db
Parents: 425b756
Author: Parth Brahmbhatt <br...@gmail.com>
Authored: Wed Jul 16 10:58:29 2014 -0700
Committer: Parth Brahmbhatt <br...@gmail.com>
Committed: Mon Jul 28 09:34:59 2014 -0700
----------------------------------------------------------------------
external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java | 7 +++++--
.../jvm/storm/kafka/bolt/selector/DefaultTopicSelector.java | 7 ++-----
.../jvm/storm/kafka/bolt/selector/KafkaTopicSelector.java | 4 +---
.../src/jvm/storm/kafka/trident/TridentKafkaState.java | 8 ++++++--
.../storm/kafka/trident/selector/DefaultTopicSelector.java | 7 ++-----
.../jvm/storm/kafka/trident/selector/KafkaTopicSelector.java | 5 +----
6 files changed, 17 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/827e0ba9/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java b/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
index 6326e9d..6c9a7d1 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
@@ -91,14 +91,17 @@ public class KafkaBolt<K, V> extends BaseRichBolt {
public void execute(Tuple input) {
K key = null;
V message = null;
+ String topic = null;
try {
key = mapper.getKeyFromTuple(input);
message = mapper.getMessageFromTuple(input);
- for(String topic : topicSelector.getTopics(input)) {
+ topic = topicSelector.getTopic(input);
+ if(topic != null ) {
producer.send(new KeyedMessage<K, V>(topic, key, message));
}
} catch (Exception ex) {
- LOG.error("Could not send message with key '" + key + "' and value '" + message + "'", ex);
+ LOG.error("Could not send message with key = " + key
+ + " and value = " + message + " to topic = " + topic, ex);
} finally {
collector.ack(input);
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/827e0ba9/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/DefaultTopicSelector.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/DefaultTopicSelector.java b/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/DefaultTopicSelector.java
index 8737dae..9c87658 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/DefaultTopicSelector.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/DefaultTopicSelector.java
@@ -18,9 +18,6 @@
package storm.kafka.bolt.selector;
import backtype.storm.tuple.Tuple;
-import com.google.common.collect.Lists;
-
-import java.util.List;
public class DefaultTopicSelector implements KafkaTopicSelector {
@@ -31,7 +28,7 @@ public class DefaultTopicSelector implements KafkaTopicSelector {
}
@Override
- public List<String> getTopics(Tuple tuple) {
- return Lists.newArrayList(topicName);
+ public String getTopic(Tuple tuple) {
+ return topicName;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/827e0ba9/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/KafkaTopicSelector.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/KafkaTopicSelector.java b/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/KafkaTopicSelector.java
index 8f2de90..baf6c4a 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/KafkaTopicSelector.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/KafkaTopicSelector.java
@@ -19,8 +19,6 @@ package storm.kafka.bolt.selector;
import backtype.storm.tuple.Tuple;
-import java.util.List;
-
public interface KafkaTopicSelector {
- List<String> getTopics(Tuple tuple);
+ String getTopic(Tuple tuple);
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/827e0ba9/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java
index 7e1ef8d..c4ea38d 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java
@@ -79,14 +79,18 @@ public class TridentKafkaState implements State {
}
public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
+ String topic = null;
for (TridentTuple tuple : tuples) {
try {
- for(String topic : topicSelector.getTopics(tuple)) {
+ topic = topicSelector.getTopic(tuple);
+
+ if(topic != null) {
producer.send(new KeyedMessage(topic, mapper.getKeyFromTuple(tuple),
mapper.getMessageFromTuple(tuple)));
}
} catch (Exception ex) {
- String errorMsg = "Could not send message with key '" + mapper.getKeyFromTuple(tuple);
+ String errorMsg = "Could not send message with key = " + mapper.getKeyFromTuple(tuple)
+ + " to topic = " + topic;
LOG.warn(errorMsg, ex);
throw new FailedException(errorMsg, ex);
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/827e0ba9/external/storm-kafka/src/jvm/storm/kafka/trident/selector/DefaultTopicSelector.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/trident/selector/DefaultTopicSelector.java b/external/storm-kafka/src/jvm/storm/kafka/trident/selector/DefaultTopicSelector.java
index 08a42be..473a38d 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/selector/DefaultTopicSelector.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/selector/DefaultTopicSelector.java
@@ -17,11 +17,8 @@
*/
package storm.kafka.trident.selector;
-import com.google.common.collect.Lists;
import storm.trident.tuple.TridentTuple;
-import java.util.List;
-
public class DefaultTopicSelector implements KafkaTopicSelector {
private final String topicName;
@@ -31,7 +28,7 @@ public class DefaultTopicSelector implements KafkaTopicSelector {
}
@Override
- public List<String> getTopics(TridentTuple tuple) {
- return Lists.newArrayList(topicName);
+ public String getTopic(TridentTuple tuple) {
+ return topicName;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/827e0ba9/external/storm-kafka/src/jvm/storm/kafka/trident/selector/KafkaTopicSelector.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/trident/selector/KafkaTopicSelector.java b/external/storm-kafka/src/jvm/storm/kafka/trident/selector/KafkaTopicSelector.java
index ae0befe..837a40d 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/selector/KafkaTopicSelector.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/selector/KafkaTopicSelector.java
@@ -17,11 +17,8 @@
*/
package storm.kafka.trident.selector;
-import backtype.storm.tuple.Tuple;
import storm.trident.tuple.TridentTuple;
-import java.util.List;
-
public interface KafkaTopicSelector {
- List<String> getTopics(TridentTuple tuple);
+ String getTopic(TridentTuple tuple);
}