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:07 UTC
[04/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/905969ca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/905969ca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/905969ca
Branch: refs/heads/security
Commit: 905969ca3a744d86a562d3d85805c25f9f9561cc
Parents: a26329c
Author: Parth Brahmbhatt <br...@gmail.com>
Authored: Wed Jul 16 10:58:29 2014 -0700
Committer: Parth Brahmbhatt <br...@gmail.com>
Committed: Tue Jul 22 14:19:24 2014 -0700
----------------------------------------------------------------------
.../src/jvm/storm/kafka/bolt/KafkaBolt.java | 19 +++++--
.../bolt/selector/DefaultTopicSelector.java | 37 ++++++++++++++
.../kafka/bolt/selector/KafkaTopicSelector.java | 26 ++++++++++
.../storm/kafka/trident/TridentKafkaState.java | 54 +++++++++++++++-----
.../kafka/trident/TridentKafkaStateFactory.java | 38 ++++++++++++--
.../kafka/trident/TridentKafkaUpdater.java | 17 ++++++
.../trident/selector/DefaultTopicSelector.java | 37 ++++++++++++++
.../trident/selector/KafkaTopicSelector.java | 27 ++++++++++
.../src/test/storm/kafka/TridentKafkaTest.java | 36 ++++++++++---
.../test/storm/kafka/TridentKafkaTopology.java | 27 ++++++++--
10 files changed, 287 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/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 a397977..35994e2 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/KafkaBolt.java
@@ -29,6 +29,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaKeyAndMessageMapper;
import storm.kafka.bolt.mapper.TupleToKafkaKeyAndMessageMapper;
+import storm.kafka.bolt.selector.DefaultTopicSelector;
+import storm.kafka.bolt.selector.KafkaTopicSelector;
import java.util.Map;
import java.util.Properties;
@@ -52,14 +54,19 @@ public class KafkaBolt<K, V> extends BaseRichBolt {
private Producer<K, V> producer;
private OutputCollector collector;
- private String topic;
private TupleToKafkaKeyAndMessageMapper<K,V> mapper;
+ private KafkaTopicSelector topicSelector;
public KafkaBolt<K,V> withTupleToKafkaKeyAndMessageMapper(TupleToKafkaKeyAndMessageMapper<K,V> mapper) {
this.mapper = mapper;
return this;
}
+ public KafkaBolt<K,V> withTopicSelector(KafkaTopicSelector selector) {
+ this.topicSelector = topicSelector;
+ return this;
+ }
+
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
//for backward compatibility.
@@ -67,12 +74,16 @@ public class KafkaBolt<K, V> extends BaseRichBolt {
this.mapper = new FieldNameBasedTupleToKafkaKeyAndMessageMapper<K,V>();
}
+ //for backward compatibility.
+ if(topicSelector == null) {
+ this.topicSelector = new DefaultTopicSelector((String) stormConf.get(TOPIC));
+ }
+
Map configMap = (Map) stormConf.get(KAFKA_BROKER_PROPERTIES);
Properties properties = new Properties();
properties.putAll(configMap);
ProducerConfig config = new ProducerConfig(properties);
producer = new Producer<K, V>(config);
- this.topic = (String) stormConf.get(TOPIC);
this.collector = collector;
}
@@ -83,7 +94,9 @@ public class KafkaBolt<K, V> extends BaseRichBolt {
try {
key = mapper.getKeyFromTuple(input);
message = mapper.getMessageFromTuple(input);
- producer.send(new KeyedMessage<K, V>(topic, key, message));
+ for(String topic : topicSelector.getTopics(input)) {
+ 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);
} finally {
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/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
new file mode 100644
index 0000000..8737dae
--- /dev/null
+++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/DefaultTopicSelector.java
@@ -0,0 +1,37 @@
+/**
+ * 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.bolt.selector;
+
+import backtype.storm.tuple.Tuple;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+
+public class DefaultTopicSelector implements KafkaTopicSelector {
+
+ private final String topicName;
+
+ public DefaultTopicSelector(final String topicName) {
+ this.topicName = topicName;
+ }
+
+ @Override
+ public List<String> getTopics(Tuple tuple) {
+ return Lists.newArrayList(topicName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/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
new file mode 100644
index 0000000..8f2de90
--- /dev/null
+++ b/external/storm-kafka/src/jvm/storm/kafka/bolt/selector/KafkaTopicSelector.java
@@ -0,0 +1,26 @@
+/**
+ * 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.bolt.selector;
+
+import backtype.storm.tuple.Tuple;
+
+import java.util.List;
+
+public interface KafkaTopicSelector {
+ List<String> getTopics(Tuple tuple);
+}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/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 d53a17b..7d3122b 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java
@@ -1,6 +1,22 @@
+/**
+ * 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.trident;
-
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.FailedException;
@@ -8,8 +24,12 @@ import com.google.common.collect.Lists;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
+import org.apache.commons.lang.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import storm.kafka.trident.mapper.FieldNameBasedTupleToKafkaKeyAndMessageMapper;
+import storm.kafka.trident.mapper.TridentTupleToKafkaKeyAndMessageMapper;
+import storm.kafka.trident.selector.KafkaTopicSelector;
import storm.trident.operation.TridentCollector;
import storm.trident.state.State;
import storm.trident.tuple.TridentTuple;
@@ -25,14 +45,21 @@ public class TridentKafkaState implements State {
public static final String TOPIC = "topic";
public static final String KAFKA_BROKER_PROPERTIES = "kafka.broker.properties";
- public static final String KEY_FIELD = "keyFieldName";
- public static final String MESSAGE_FIELD_NAME = "messageFieldName";
-
private Producer producer;
private OutputCollector collector;
- private String topic;
- private String keyField;
- private String msgField;
+
+ private TridentTupleToKafkaKeyAndMessageMapper mapper;
+ private KafkaTopicSelector topicSelector;
+
+ public TridentKafkaState withTridentTupleToKafkaKeyAndMessageMapper(TridentTupleToKafkaKeyAndMessageMapper mapper) {
+ this.mapper = mapper;
+ return this;
+ }
+
+ public TridentKafkaState withKafkaTopicSelector(KafkaTopicSelector selector) {
+ this.topicSelector = selector;
+ return this;
+ }
@Override
public void beginCommit(Long txid) {
@@ -45,25 +72,24 @@ public class TridentKafkaState implements State {
}
public void prepare(Map stormConf) {
+ Validate.notNull(mapper, "mapper can not be null");
+ Validate.notNull(topicSelector, "topicSelector can not be null");
Map configMap = (Map) stormConf.get(KAFKA_BROKER_PROPERTIES);
Properties properties = new Properties();
properties.putAll(configMap);
ProducerConfig config = new ProducerConfig(properties);
producer = new Producer(config);
- this.topic = (String) stormConf.get(TOPIC);
- this.keyField = (String) stormConf.get(KEY_FIELD);
- this.msgField = (String) stormConf.get(MESSAGE_FIELD_NAME);
}
public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
for (TridentTuple tuple : tuples) {
try {
- if( tuple.getValueByField(keyField) != null) {
- producer.send(new KeyedMessage(topic, tuple.getValueByField(keyField),
- tuple.getValueByField(msgField)));
+ for(String topic : topicSelector.getTopics(tuple)) {
+ producer.send(new KeyedMessage(topic, mapper.getKeyFromTuple(tuple),
+ mapper.getMessageFromTuple(tuple)));
}
} catch (Exception ex) {
- String errorMsg = "Could not send message with key '" + tuple.getValueByField(keyField);
+ String errorMsg = "Could not send message with key '" + mapper.getKeyFromTuple(tuple);
LOG.warn(errorMsg, ex);
throw new FailedException(errorMsg, ex);
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaStateFactory.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaStateFactory.java b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaStateFactory.java
index 4962b12..97159a2 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaStateFactory.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaStateFactory.java
@@ -1,9 +1,27 @@
+/**
+ * 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.trident;
-
import backtype.storm.task.IMetricsContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import storm.kafka.trident.mapper.TridentTupleToKafkaKeyAndMessageMapper;
+import storm.kafka.trident.selector.KafkaTopicSelector;
import storm.trident.state.State;
import storm.trident.state.StateFactory;
@@ -13,12 +31,26 @@ public class TridentKafkaStateFactory implements StateFactory {
private static final Logger LOG = LoggerFactory.getLogger(TridentKafkaStateFactory.class);
- public TridentKafkaStateFactory(){}
+ private TridentTupleToKafkaKeyAndMessageMapper mapper;
+ private KafkaTopicSelector topicSelector;
+
+
+ public TridentKafkaStateFactory withTridentTupleToKafkaKeyAndMessageMapper(TridentTupleToKafkaKeyAndMessageMapper mapper) {
+ this.mapper = mapper;
+ return this;
+ }
+
+ public TridentKafkaStateFactory withKafkaTopicSelector(KafkaTopicSelector selector) {
+ this.topicSelector = selector;
+ return this;
+ }
@Override
public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
LOG.info("makeState(partitonIndex={}, numpartitions={}", partitionIndex, numPartitions);
- TridentKafkaState state = new TridentKafkaState();
+ TridentKafkaState state = new TridentKafkaState()
+ .withKafkaTopicSelector(this.topicSelector)
+ .withTridentTupleToKafkaKeyAndMessageMapper(this.mapper);
state.prepare(conf);
return state;
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaUpdater.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaUpdater.java b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaUpdater.java
index 7af2fad..6639b36 100644
--- a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaUpdater.java
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaUpdater.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.trident;
import storm.trident.operation.TridentCollector;
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/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
new file mode 100644
index 0000000..08a42be
--- /dev/null
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/selector/DefaultTopicSelector.java
@@ -0,0 +1,37 @@
+/**
+ * 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.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;
+
+ public DefaultTopicSelector(final String topicName) {
+ this.topicName = topicName;
+ }
+
+ @Override
+ public List<String> getTopics(TridentTuple tuple) {
+ return Lists.newArrayList(topicName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/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
new file mode 100644
index 0000000..ae0befe
--- /dev/null
+++ b/external/storm-kafka/src/jvm/storm/kafka/trident/selector/KafkaTopicSelector.java
@@ -0,0 +1,27 @@
+/**
+ * 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.trident.selector;
+
+import backtype.storm.tuple.Tuple;
+import storm.trident.tuple.TridentTuple;
+
+import java.util.List;
+
+public interface KafkaTopicSelector {
+ List<String> getTopics(TridentTuple tuple);
+}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/external/storm-kafka/src/test/storm/kafka/TridentKafkaTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/test/storm/kafka/TridentKafkaTest.java b/external/storm-kafka/src/test/storm/kafka/TridentKafkaTest.java
index ba4e238..e657327 100644
--- a/external/storm-kafka/src/test/storm/kafka/TridentKafkaTest.java
+++ b/external/storm-kafka/src/test/storm/kafka/TridentKafkaTest.java
@@ -1,16 +1,33 @@
+/**
+ * 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 backtype.storm.Config;
-import backtype.storm.tuple.Values;
import com.google.common.collect.Lists;
import kafka.javaapi.consumer.SimpleConsumer;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
import storm.kafka.trident.TridentKafkaState;
-import storm.trident.operation.TridentCollector;
+import storm.kafka.trident.mapper.FieldNameBasedTupleToKafkaKeyAndMessageMapper;
+import storm.kafka.trident.mapper.TridentTupleToKafkaKeyAndMessageMapper;
+import storm.kafka.trident.selector.DefaultTopicSelector;
+import storm.kafka.trident.selector.KafkaTopicSelector;
import storm.trident.testing.MockTridentTuple;
import storm.trident.tuple.TridentTuple;
@@ -23,16 +40,19 @@ public class TridentKafkaTest {
private Config config;
private KafkaConfig kafkaConfig;
private SimpleConsumer simpleConsumer;
-
+ private TridentTupleToKafkaKeyAndMessageMapper mapper;
+ private KafkaTopicSelector topicSelector;
@Before
public void setup() {
broker = new KafkaTestBroker();
simpleConsumer = TestUtils.getKafkaConsumer(broker);
config = TestUtils.getConfig(broker.getBrokerConnectionString());
- config.put(TridentKafkaState.KEY_FIELD, "key");
- config.put(TridentKafkaState.MESSAGE_FIELD_NAME, "message");
- state = new TridentKafkaState();
+ mapper = new FieldNameBasedTupleToKafkaKeyAndMessageMapper("key", "message");
+ topicSelector = new DefaultTopicSelector(TestUtils.TOPIC);
+ state = new TridentKafkaState()
+ .withKafkaTopicSelector(topicSelector)
+ .withTridentTupleToKafkaKeyAndMessageMapper(mapper);
state.prepare(config);
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/905969ca/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 a24a67c..73ee6a7 100644
--- a/external/storm-kafka/src/test/storm/kafka/TridentKafkaTopology.java
+++ b/external/storm-kafka/src/test/storm/kafka/TridentKafkaTopology.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 backtype.storm.Config;
@@ -15,6 +32,8 @@ import storm.kafka.trident.GlobalPartitionInformation;
import storm.kafka.trident.TridentKafkaState;
import storm.kafka.trident.TridentKafkaStateFactory;
import storm.kafka.trident.TridentKafkaUpdater;
+import storm.kafka.trident.mapper.FieldNameBasedTupleToKafkaKeyAndMessageMapper;
+import storm.kafka.trident.selector.DefaultTopicSelector;
import storm.trident.Stream;
import storm.trident.TridentTopology;
import storm.trident.state.StateFactory;
@@ -39,7 +58,11 @@ public class TridentKafkaTopology {
TridentTopology topology = new TridentTopology();
Stream stream = topology.newStream("spout1", spout);
- stream.partitionPersist(new TridentKafkaStateFactory(), fields, new TridentKafkaUpdater(), new Fields());
+
+ TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory()
+ .withKafkaTopicSelector(new DefaultTopicSelector("test"))
+ .withTridentTupleToKafkaKeyAndMessageMapper(new FieldNameBasedTupleToKafkaKeyAndMessageMapper("word", "count"));
+ stream.partitionPersist(stateFactory, fields, new TridentKafkaUpdater(), new Fields());
return topology.build();
}
@@ -80,8 +103,6 @@ public class TridentKafkaTopology {
props.put("serializer.class", "kafka.serializer.StringEncoder");
conf.put(TridentKafkaState.KAFKA_BROKER_PROPERTIES, props);
conf.put(TridentKafkaState.TOPIC, "test");
- conf.put(TridentKafkaState.KEY_FIELD, "word");
- conf.put(TridentKafkaState.MESSAGE_FIELD_NAME, "count");
return conf;
}