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)
I am trying to use Spark Streaming for comsumer messages from Kafka. But when the program starts for a while, I get the following information, then always output INFO SimpleConsumer: I reconnect due to a socket error: 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
Does anyone know what could be the problem?
Z.min source share