일단 환경은 HortonWorks HDP 플랫폼 위입니다.
Storm을 Kafaka의 Consumer로 Spout를 만드는데, 이게 오픈소스라 그런지는 몰라도 잔 버그가 많습니다.
Strom-kafka : ava.lang.RuntimeException:java.lang.RuntimeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/topics/topic_sdka/partitions
이런 당황스러운 오류가 Spout 쪽에서 날수가 있는데,
스택오버플로우에서는 버전을 맞추라느니, 뭐라느니, 그러는데
결론은 Kafka, Storm, Zookeeper Restart 한다음에 다시 Topology 를 올리면 정상적으로 작동됩니다.
예전부터 하둡을 공부하면서 느끼는건 영문 자료보다, 중문자료가 볼만한게 정말 많네요
http://www.aboutyun.com/blog-1328-2341.html (Reference) Thank you!!!
kafka-storm spout拉取数据问题offset out of range(todo)
已有 76 次阅读2015-5-8 14:54
<问题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>