Kafka Spout Offset存储在Zookeeper

Kafka Spout Offset存储在Zookeeper，配置中需要单独指定SpoutConfig.zkPort和SpoutConfig.zkServers,否则无法生效。

1import java.util.ArrayList;
2import java.util.List;
3
4import org.apache.storm.Config;
5import org.apache.storm.LocalCluster;
6import org.apache.storm.StormSubmitter;
7import org.apache.storm.kafka.BrokerHosts;
8import org.apache.storm.kafka.KafkaSpout;
9import org.apache.storm.kafka.SpoutConfig;
10import org.apache.storm.kafka.StringScheme;
11import org.apache.storm.kafka.ZkHosts;
12import org.apache.storm.spout.SchemeAsMultiScheme;
13import org.apache.storm.topology.BasicOutputCollector;
14import org.apache.storm.topology.OutputFieldsDeclarer;
15import org.apache.storm.topology.TopologyBuilder;
16import org.apache.storm.topology.base.BaseBasicBolt;
17import org.apache.storm.tuple.Tuple;
18
19/**
20 * ClassName:CKafkaSource <br/>
21 * Function: TODO ADD FUNCTION. <br/>
22 * Reason: TODO ADD REASON. <br/>
23 * Date: 2017年8月10日 下午4:51:24 <br/>
24 */
25public class CKafkaSource {
26
27	@SuppressWarnings({ "serial" })
28	public static void main(String[] args) throws Exception {
29		TopologyBuilder builder = new TopologyBuilder();
30
31		//offset config start
32		BrokerHosts hosts = new ZkHosts("breath:2181");
33		String topicName = "test";
34		SpoutConfig spoutConfig = new SpoutConfig(hosts, "test", "/consumers/" + topicName,
35				"CKafkaSource".toLowerCase());
36		spoutConfig.zkPort = 2181;
37		List<String> servers = new ArrayList<>();
38		servers.add("breath");
39		spoutConfig.zkServers = servers;
40		//offset config start
41		
42		spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
43		KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
44
45		builder.setSpout("spout", kafkaSpout);
46
47		builder.setBolt("bolt", new BaseBasicBolt() {
48			[@Override](https://my.oschina.net/u/1162528)
49			public void execute(Tuple input, BasicOutputCollector collector) {
50				System.err.println(input.getString(0));
51			}
52
53			[@Override](https://my.oschina.net/u/1162528)
54			public void declareOutputFields(OutputFieldsDeclarer declarer) {
55			}
56		}).shuffleGrouping("spout");
57
58		Config conf = new Config();
59		conf.setDebug(false);
60		if (args != null && args.length > 0) {
61			conf.setNumWorkers(3);
62			StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
63		} else {
64			conf.setMaxTaskParallelism(1);
65			LocalCluster cluster = new LocalCluster();
66			cluster.submitTopology("wordcount", conf, builder.createTopology());
67			Thread.sleep(20000);
68			cluster.shutdown();
69		}
70	}
71}