spark 1.6.2（apache）kafka 2.1.1（cdh 5.7.2）

val conf = new   SparkConf().setAppName("Test").setMaster("local[2]")
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(15))
val kafkaDStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

我想用spark流媒体来传播Kafka的消息。但是当程序运行一段时间后，我得到以下信息，然后总是输出信息simpleconsumer:由于套接字错误而重新连接：java.nio.channels.closedchannelexception：

16/10/08 18:28:00 INFO SimpleConsumer: Reconnect due to socket error:                      java.nio.channels.ClosedChannelException
16/10/08 18:28:00 INFO JobScheduler: Added jobs for time 1475922480000 ms
16/10/08 18:28:15 INFO JobScheduler: Added jobs for time 1475922495000 ms
16/10/08 18:28:30 INFO JobScheduler: Added jobs for time 1475922510000 ms
16/10/08 18:28:45 INFO JobScheduler: Added jobs for time 1475922525000 ms
16/10/08 18:29:00 INFO SimpleConsumer: Reconnect due to socket error:     java.nio.channels.ClosedChannelException
16/10/08 18:29:00 INFO JobScheduler: Added jobs for time 1475922540000 ms
16/10/08 18:29:15 INFO SimpleConsumer: Reconnect due to socket error:     java.nio.channels.ClosedChannelException
16/10/08 18:29:15 INFO JobScheduler: Added jobs for time 1475922555000 ms
16/10/08 18:29:30 INFO JobScheduler: Added jobs for time 1475922570000 ms
16/10/08 18:29:45 INFO JobScheduler: Added jobs for time 1475922585000 ms
16/10/08 18:30:00 INFO JobScheduler: Added jobs for time 1475922600000 ms
16/10/08 18:30:15 INFO JobScheduler: Added jobs for time 1475922615000 ms
16/10/08 18:30:30 INFO JobScheduler: Added jobs for time 1475922630000 ms
16/10/08 18:30:45 INFO JobScheduler: Added jobs for time 1475922645000 ms
16/10/08 18:31:00 INFO SimpleConsumer: Reconnect due to socket error: java.nio.channels.ClosedChannelException
16/10/08 18:31:00 INFO JobScheduler: Added jobs for time 1475922660000 ms
16/10/08 18:31:15 INFO SimpleConsumer: Reconnect due to socket error: java.nio.channels.ClosedChannelException
16/10/08 18:31:15 INFO JobScheduler: Added jobs for time 1475922675000 ms
16/10/08 18:31:30 INFO JobScheduler: Added jobs for time 1475922690000 ms
16/10/08 18:31:45 INFO JobScheduler: Added jobs for time 1475922705000 ms
16/10/08 18:32:00 INFO JobScheduler: Added jobs for time 1475922720000 ms
16/10/08 18:32:15 INFO JobScheduler: Added jobs for time 1475922735000 ms
16/10/08 18:32:30 INFO JobScheduler: Added jobs for time 1475922750000 ms
16/10/08 18:32:45 INFO JobScheduler: Added jobs for time 1475922765000 ms
16/10/08 18:33:00 INFO JobScheduler: Added jobs for time 1475922780000 ms
16/10/08 18:33:15 INFO SimpleConsumer: Reconnect due to socket error: java.nio.channels.ClosedChannelException
16/10/08 18:33:15 INFO JobScheduler: Added jobs for time 1475922795000 ms
16/10/08 18:33:30 INFO JobScheduler: Added jobs for time 1475922810000 ms
16/10/08 18:33:45 INFO JobScheduler: Added jobs for time 1475922825000 ms
16/10/08 18:34:00 INFO JobScheduler: Added jobs for time 1475922840000 ms
16/10/08 18:34:15 INFO JobScheduler: Added jobs for time 1475922855000 ms
16/10/08 18:34:30 INFO JobScheduler: Added jobs for time 1475922870000 ms
16/10/08 18:34:45 INFO SimpleConsumer: Reconnect due to socket error: java.nio.channels.ClosedChannelException

有人知道问题出在哪里吗？