<问题1>kafka-spout在拉取数据的时候，offset超出了范围

日志：

[WARN  2014-12-29 20:40:52 KafkaUtils:185 ASpout:3-MultipleThreadSpoutExecutors] Got fetch request with offset out of range: [232]; retrying with default start offset time from configuration. configured start offset time: [-2] offset: [0]

[WARN  2014-12-29 20:40:52 KafkaUtils:185 ASpout:3-MultipleThreadSpoutExecutors] Got fetch request with offset out of range: [239]; retrying with default start offset time from configuration. configured start offset time: [-2] offset: [0]

解决办法：

1.修改SpoutConfig.zkRoot 值

String zkRoot = "";

SpoutConfig spoutConf = new SpoutConfig(hosts, topic, zkRoot, id);

2.将spoutConf.forceFromStart 设为 true;

原因：

storm启动的时候，默认是从头读取-2，当kafkaspout的task出错时候，会重新启动，这时候offset就会从zk中最老的offset，但kafka数据只会保存一段时间，从zk中读取道德offset的数据可能已经不存在，超过现在的范围。

kafkaspout读取的3种选项

  -2: 从最老的开始读

  -1: 从最近的开始读

  0: 从Zk中读

问题2

日志：

java.lang.RuntimeException:java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/topics/topic_sdka/partitions
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:81)


Caused by: java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/topics/topic_sdka/partitions
        at storm.kafka.DynamicBrokersReader.getNumPartitions(DynamicBrokersReader.java:94)
        at storm.kafka.DynamicBrokersReader.getBrokerInfo(DynamicBrokersReader.java:65)
        ... 13 more
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/topics/topic_sdka/partitions
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:111)

问题3：启动kafka server cluster的时候，某个节点启动失败

解决方法：重启所有节点的zookeeper，kafka，问题解决

<详细>

[hadoop@node3 bin]$ ./kafka-topics.sh --create --zookeeper 192.168.11.222:2181,192.168.11.221:2181,192.168.11.220:2181 --replication-factor 1 --partitions 3  --topic topic1
Error while executing topic command org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids
org.I0Itec.zkclient.exception.ZkNoNodeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids
        at org.I0Itec.zkclient.exception.ZkException.create(ZkException.java:47)
        at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:685)
        at org.I0Itec.zkclient.ZkClient.getChildren(ZkClient.java:413)
        at org.I0Itec.zkclient.ZkClient.getChildren(ZkClient.java:409)
        at kafka.utils.ZkUtils$.getChildren(ZkUtils.scala:480)
        at kafka.utils.ZkUtils$.getSortedBrokerList(ZkUtils.scala:81)
        at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:154)
        at kafka.admin.TopicCommand$.createTopic(TopicCommand.scala:86)
        at kafka.admin.TopicCommand$.main(TopicCommand.scala:50)
        at kafka.admin.TopicCommand.main(TopicCommand.scala)
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:102)
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:42)
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1249)
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1277)
        at org.I0Itec.zkclient.ZkConnection.getChildren(ZkConnection.java:99)
        at org.I0Itec.zkclient.ZkClient$2.call(ZkClient.java:416)
        at org.I0Itec.zkclient.ZkClient$2.call(ZkClient.java:413)
        at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
        ... 8 more



【resovle】
1.restart kafka cluster
<node1>
[2015-01-16 13:58:49,929] INFO Initiating client connection, connectString=192.168.11.222:2181,192.168.11.221:2181,192.168.11.220:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@68098843 (org.apache.zookeeper.ZooKeeper)
[2015-01-16 13:58:49,950] INFO Opening socket connection to server /192.168.11.221:2181 (org.apache.zookeeper.ClientCnxn)
[2015-01-16 13:58:49,956] INFO Socket connection established to node2.ssjs/192.168.11.221:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2015-01-16 13:58:49,982] INFO Session establishment complete on server node2.ssjs/192.168.11.221:2181, sessionid = 0x24ac286ad2c0dc6, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
[2015-01-16 13:58:49,987] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient)
[2015-01-16 13:58:50,032] INFO Registered broker 1 at path /brokers/ids/1 with address 192.168.11.222:9092. (kafka.utils.ZkUtils$)
[2015-01-16 13:58:50,040] INFO [Kafka Server 1], Connecting to ZK: 192.168.11.222:2181,192.168.11.221:2181,192.168.11.220:2181 (kafka.server.KafkaServer)
[2015-01-16 13:58:50,192] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader$)
[2015-01-16 13:58:50,259] INFO conflict in /controller data: 1 stored data: {"version":1,"brokerid":3,"timestamp":"1421387751644"} (kafka.utils.ZkUtils$)
[2015-01-16 13:58:50,265] FATAL Fatal error during KafkaServerStable startup. Prepare to shutdown (kafka.server.KafkaServerStartable)
java.lang.NumberFormatException: For input string: "{"version":1,"brokerid":3,"timestamp":"1421387751644"}"
        at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
        at java.lang.Integer.parseInt(Integer.java:492)
        at java.lang.Integer.parseInt(Integer.java:527)
        at scala.collection.immutable.StringLike$class.toInt(StringLike.scala:229)
        at scala.collection.immutable.StringOps.toInt(StringOps.scala:31)
        at kafka.server.ZookeeperLeaderElector.elect(Unknown Source)
        at kafka.server.ZookeeperLeaderElector.startup(Unknown Source)
        at kafka.controller.KafkaController.startup(Unknown Source)
        at kafka.server.KafkaServer.startup(Unknown Source)
        at kafka.server.KafkaServerStartable.startup(Unknown Source)
        at kafka.Kafka$.main(Unknown Source)
        at kafka.Kafka.main(Unknown Source)
[2015-01-16 13:58:50,271] INFO [Kafka Server 1], Shutting down (kafka.server.KafkaServer)


<good_node2>
[2015-01-16 14:17:57,401] INFO 2 successfully elected as leader (kafka.server.ZookeeperLeaderElector)
[2015-01-16 14:17:57,605] INFO Registered broker 2 at path /brokers/ids/2 with address 192.168.11.221:9092. (kafka.utils.ZkUtils$)
[2015-01-16 14:17:57,625] INFO [Kafka Server 2], started (kafka.server.KafkaServer)
[2015-01-16 14:17:57,779] INFO New leader is 2 (kafka.server.ZookeeperLeaderElector$LeaderChangeListener)




KafkaServer.startup()
kafkaController = new KafkaController(config, zkClient)


restart zk,kafka<node1>
[2015-01-16 15:19:33,273] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient)
[2015-01-16 15:19:33,318] INFO Registered broker 1 at path /brokers/ids/1 with address 192.168.11.222:9092. (kafka.utils.ZkUtils$)
[2015-01-16 15:19:33,319] INFO [Kafka Server 1], Connecting to ZK: 192.168.11.222:2181,192.168.11.221:2181,192.168.11.220:2181 (kafka.server.KafkaServer)
[2015-01-16 15:19:33,474] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader$)
[2015-01-16 15:19:33,543] INFO conflict in /controller data: 1 stored data: {"version":1,"brokerid":2,"timestamp":"1421389077341"} (kafka.utils.ZkUtils$)
[2015-01-16 15:19:33,549] FATAL Fatal error during KafkaServerStable startup. Prepare to shutdown (kafka.server.KafkaServerStartable)
java.lang.NumberFormatException: For input string: "{"version":1,"brokerid":2,"timestamp":"1421389077341"}"
        at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
        at java.lang.Integer.parseInt(Integer.java:492)
        at java.lang.Integer.parseInt(Integer.java:527)
        at scala.collection.immutable.StringLike$class.toInt(StringLike.scala:229)
        at scala.collection.immutable.StringOps.toInt(StringOps.scala:31)
        at kafka.server.ZookeeperLeaderElector.elect(Unknown Source)
        at kafka.server.ZookeeperLeaderElector.startup(Unknown Source)
        at kafka.controller.KafkaController.startup(Unknown Source)
        at kafka.server.KafkaServer.startup(Unknown Source)
        at kafka.server.KafkaServerStartable.startup(Unknown Source)
        at kafka.Kafka$.main(Unknown Source)
        at kafka.Kafka.main(Unknown Source)



restart node1~node3:zk,kafka
resolved!!<session timeout>