这是一个在yarn集群模式下运行的spark流应用程序，它在三个kafka代理中生成消息。
一旦达到150k打开文件，它就会失败：

There is insufficient memory for the Java Runtime Environment to continue
Native memory allocation (mmap) failed  to map 12288 bytes for committing reserved memory.

Job aborted due to stage failure ... : 
org.apache.kafka.common.KafkaException: Failed to construct kafka producer
.....
Caused by: java.lang.OutOfMemoryError: unable to create new native thread

什么时候做 lsof -p <PID> 对于运行该executor的java进程，我可以在kafka代理中看到来自主机服务器的成吨（高达90k）tcp连接： host:portXXX->kafkabroker1:XmlIpcRegSvc (ESTABLISHED) host:portYYY->kafkabroker2:XmlIpcRegSvc (ESTABLISHED) host:portZZZ->kafkabroker3:XmlIpcRegSvc (ESTABLISHED) 我尝试将executor内核的数量从8个减少到6个，但是在打开的文件数量上没有任何差别（仍然达到150k），然后一直失败。
从spark streaming连接到kafka的库有：

org.apache.spark.streaming.kafka010.KafkaUtils
org.apache.spark.streaming.dstream.InputDStream
org.apache.kafka.clients.producer.kafkaproducer

代码：

foreachRDD{
   get kafkaProducer
   do some work on each RDD...
   foreach( record => {
      kafkaProducer.send(record._1,record._2)
   }
  kafkaProducer.close()
}