You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2014/08/18 21:42:22 UTC
[19/50] 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 interfac
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/c6ef84d1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/c6ef84d1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/c6ef84d1
Branch: refs/heads/security
Commit: c6ef84d13e9e767e6ae21f2772b5ad836e27e43e
Parents: 813ed7f
Author: Parth Brahmbhatt <br...@gmail.com>
Authored: Wed Jul 16 10:58:29 2014 -0700
Committer: Parth Brahmbhatt <br...@gmail.com>
Committed: Wed Jul 30 11:13:03 2014 -0700
----------------------------------------------------------------------
external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java | 2 ++
.../src/jvm/storm/kafka/trident/TridentKafkaState.java | 4 ++--
.../storm-kafka/src/test/storm/kafka/TridentKafkaTopology.java | 3 +--
3 files changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/c6ef84d1/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 6c9a7d1..51cbfd0 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
@@ -98,6 +98,8 @@ public class KafkaBolt<K, V> extends BaseRichBolt {
topic = topicSelector.getTopic(input);
if(topic != null ) {
producer.send(new KeyedMessage<K, V>(topic, key, message));
+ } else {
+ LOG.warn("skipping key = " + key + ", topic selector returned null.");
}
} catch (Exception ex) {
LOG.error("Could not send message with key = " + key
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/c6ef84d1/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 c4ea38d..402ffb1 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java
@@ -38,8 +38,6 @@ import java.util.Properties;
public class TridentKafkaState implements State {
private static final Logger LOG = LoggerFactory.getLogger(TridentKafkaState.class);
-
- public static final String TOPIC = "topic";
public static final String KAFKA_BROKER_PROPERTIES = "kafka.broker.properties";
private Producer producer;
@@ -87,6 +85,8 @@ public class TridentKafkaState implements State {
if(topic != null) {
producer.send(new KeyedMessage(topic, mapper.getKeyFromTuple(tuple),
mapper.getMessageFromTuple(tuple)));
+ } else {
+ LOG.warn("skipping key = " + mapper.getKeyFromTuple(tuple) + ", topic selector returned null.");
}
} catch (Exception ex) {
String errorMsg = "Could not send message with key = " + mapper.getKeyFromTuple(tuple)
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/c6ef84d1/external/storm-kafka/src/test/storm/kafka/TridentKafkaTopology.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/test/storm/kafka/TridentKafkaTopology.java b/external/storm-kafka/src/test/storm/kafka/TridentKafkaTopology.java
index b64e6ca..9cb7bbf 100644
--- a/external/storm-kafka/src/test/storm/kafka/TridentKafkaTopology.java
+++ b/external/storm-kafka/src/test/storm/kafka/TridentKafkaTopology.java
@@ -65,7 +65,7 @@ public class TridentKafkaTopology {
* kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 1 --topic test
*
* run this program and run the kafka consumer:
- * afka-console-consumer.sh --zookeeper localhost:2181 --whitelist test --from-beginning
+ * kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
*
* you should see the messages flowing through.
*
@@ -94,7 +94,6 @@ public class TridentKafkaTopology {
props.put("request.required.acks", "1");
props.put("serializer.class", "kafka.serializer.StringEncoder");
conf.put(TridentKafkaState.KAFKA_BROKER_PROPERTIES, props);
- conf.put(TridentKafkaState.TOPIC, "test");
return conf;
}