使用Storm Toplology使用Kafka队列中的数据

Question

该算法使用KafkaSpout从卡夫卡队列中读取数据。

我正面临以下例外情况：

Exception in thread "main" java.lang.IllegalStateException: Couldn't initialize the topology
  at com.bridgera.iot.kafka.App.main(App.java:63)
Caused by: java.lang.IllegalArgumentException: Storm conf is not valid. Must be json-serializable
  at backtype.storm.StormSubmitter.submitTopology(StormSubmitter.java:104)
  at backtype.storm.StormSubmitter.submitTopology(StormSubmitter.java:70)
  at com.bridgera.iot.kafka.App.main(App.java:60)

我的Java代码：

public static void main( String[] args ) throws AlreadyAliveException, InvalidTopologyException
{
    String nimbusHost = "localhost";
    ZkHosts zkHosts=new ZkHosts("localhost:2181");
    String topic_name="test";
    String consumer_group_id="storm";
    String zookeeper_root="";
    SpoutConfig kafkaConfig=new SpoutConfig(zkHosts, 
            topic_name, zookeeper_root, consumer_group_id);
    kafkaConfig.scheme=new SchemeAsMultiScheme(new StringScheme());
    KafkaSpout kafkaSpout = new KafkaSpout(kafkaConfig);
    TopologyBuilder builder=new TopologyBuilder();
    //builder.setSpout("KafkaSpout", kafkaSpout, 1);
    builder.setSpout("KafkaSpout", kafkaSpout);
    builder.setBolt("PrinterBolt", new PrinterBolt()).globalGrouping("KafkaSpout");
    Map<String, Object> conf = new HashMap<String, Object>();
    conf.put(Config.TRANSACTIONAL_ZOOKEEPER_PORT, 2181);
    conf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Arrays.asList("localhost"));
    conf.put(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT, 20000);
    conf.put(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT, 20000);
    conf.put(Config.STORM_ZOOKEEPER_RETRY_TIMES, 3);
    conf.put(Config.STORM_ZOOKEEPER_RETRY_INTERVAL, 30);
    LocalCluster cluster=new LocalCluster();
    try{
        cluster.submitTopology("KafkaConsumerTopology", conf, builder.createTopology());
        Thread.sleep(120000);
    }catch (Exception e) {
        throw new IllegalStateException("Couldn't initialize the topology", e);
    }
}

让我知道我在配置中做错了什么。仅供参考:我在AWS集群中运行Zookeeper和Storm JVM (本地执行)。

Answer 1

试试这个：

List<String> hosts = Arrays.asList("localhost")
conf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, hosts);

我认为这是从Arrays.asList("localhost")返回的java.util.AbstractList<Object>和对json的解析的问题。