Streaming 程序调用 Producer.close hang 住问题追查复盘

本文作为一个问题追查过程的复盘记录，主要希望找出自己在解决问题中可以优化改进的地方。以后遇到问题，能够快速的进行定位，解决。

问题现象

一个 Spark Streaming 作业从 Kafka 消费数据，写往 ES，在 Spark Streaming 作业中会采集一些 metric 指标发往一个特定的 topic A。每次往 A 发送完数据后会调用 producer.close() 方法，看到的现象为：作业启动一段时间之后 hang 住，类似下图

hang_job.jpg

排查问题的过程

  1. 看到现象后，知道作业 hang 住了，希望能找到为什么 hang 住。找到该作业的 executor 地址（如下图所示）
executor.jpg

然后登录到机器上，通过 lsof 查看对应的进程，再通过 jstack dump 出具体的线程栈信息。由于第一次解决线程 hang 住的问题，得到栈信息后，暂时无从下手，然后 google jvm 线程 hang 住 等关键词，检查死锁 – 发现没有。

发现线程有 RUNNABLE，WAITING，TIMED_WAITING 等状态，然后一个个查看这些状态分别代表啥意思。到这就不知道怎么继续了 – 中间在 Spark Streaming 微信群里请教各路大神，有人说遇到链接关不掉的情况 – 多次重复查看 jstack 出来的信息，发现有一个 WAITING 线程在等待锁，具体如下:

1
2
3
4
5
6
7
8
java.lang.Thread.State: WAITING (on object monitor)
        at java.lang.Object.wait(Native Method)
        - waiting on <0x00000000c52cfba0> (a org.apache.kafka.common.utils.KafkaThread)
        at java.lang.Thread.join(Thread.java:1281)
        - locked <0x00000000c52cfba0> (a org.apache.kafka.common.utils.KafkaThread)
        at java.lang.Thread.join(Thread.java:1355)
        at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:422)
        at org.elasticsearch.hadoop.rest.KafkaProducer.close(DSLKafkaProducer.java:60)

然后对照到代码，在 Producer.close() 中有一句代码如下 ioThread.join()，猜测是 ioThread 一直没有执行完毕导致的。

  1. 注释掉 producer.close() 这一句代码之后，重新上线运行 Spark Streaming 作业，发现没有再次出现问题。大致确定问题出在 producer.close() 这里。但是不确定更深层次的问题是啥。期间猜测是由于 producer 发送数据的时候需要有 leader 确认（配置有关），然后将这个配置修改为无需 leader 确认立即返回，但是依然会导致作业 hang 住。然后阅读源码，发现 producer.close() 方法做了两件事：1）将还未发送出去的数据发送出去，2）等待正在发送的数据完成。暂时没有找到造成 ioThread 线程 hang 住的原因。暂时不知道具体 hang 住的地方在哪，至此暂时告一段落。

  2. 再次跟进该问题，尝试找出造成线程 hang 住的原因，尝试 jdb attach 到具体的线程。得到如下信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
> thread 0x1
kafka-producer-network-thread | producer-12[1] where
	  [1] sun.nio.ch.EPollArrayWrapper.epollWait (native method)
	  [2] sun.nio.ch.EPollArrayWrapper.poll (EPollArrayWrapper.java:269)
	  [3] sun.nio.ch.EPollSelectorImpl.doSelect (EPollSelectorImpl.java:79)
	  [4] sun.nio.ch.SelectorImpl.lockAndDoSelect (SelectorImpl.java:87)
	  [5] sun.nio.ch.SelectorImpl.select (SelectorImpl.java:98)
	  [6] org.apache.kafka.common.network.Selector.select (Selector.java:328)
	  [7] org.apache.kafka.common.network.Selector.poll (Selector.java:218)
	  [8] org.apache.kafka.clients.NetworkClient.poll (NetworkClient.java:192)
	  [9] org.apache.kafka.clients.producer.internals.Sender.run (Sender.java:191)
	  [10] org.apache.kafka.clients.producer.internals.Sender.run (Sender.java:135)
	  [11] java.lang.Thread.run (Thread.java:745)

到这里暂时不知道该怎么继续往下查了，知道在这里 hang 住了，但是暂时不知道怎么继续往下查，看着屏幕发呆，然后想着这个问题或许别人也遇到过，就从上面的 栈信息 中抽取一部分关键词进行 google，得到信息在 kafka 0.8.2.1 中 producer.close() 在某些情况下会 hang 住，详情参考 KIP-19，在 kafka 0.9.0.0 中提供一个带超时的 close 方法进行修复。

问题复盘

  1. 在知道作业 hang 住的情况，又不了解相应调试的情况下，能否快速了解定位问题的方法，能否询问其他人快速的定位问题，或者如何通过搜索引擎快速的获取自己需要的知识。这里自己有个小私心 – 觉得这是测试的作业，想保留现场，通过自己的努力完全把问题解决，好提升自己的能力。另外自己如何在平时积累一些查问题的经验（这次发现官方文档真是个好东西）

  2. 通过微信群询问是一个方法，但是提问需要有技巧，要能够提炼出自己的问题，以及自己进行了哪些尝试，有什么思考，而不是做伸手党。

  3. 为什么到最后才想着 Google 相关信息，而不是在知道 producer.close() 导致作业 hang 住的时候就 Google 相关信息。
  4. 对 Java 排查问题的工具非常不熟练，在平时需要自己模拟各种 case 进行练手。jstack, jvisualvm, jdb 等都是第一次使用，这些工具需要在平时进行熟练。
  5. 对常见的库或通用的写法要有一定的了解，比如看到 org.apache.kafka.common.network.Selector.poll 是否能想到没有超时而导致一直 hang 住，这些平时需要积累（思考这个怎么积累？）

Comments

▲