Kafka can normal consumption, consumers can take to the data from the kafka cluster, the logic of consumer is such,

KafkaStreamConsumerIteratorWhile (it. HasNext ())
System. The out. Println (new String (it. The next (). The message ()));

//now launch all the threads
Executor=Executors. NewFixedThreadPool (numThreads);

//now create an object to consume the messages
//
Int threadNumber=0;
For (final KafkaStream stream: streams) {
Executor. Submit (new ConsumerMsgTask (stream, threadNumber));
ThreadNumber++;
}
}

But after all data consumption, execution to the while (it. HasNext ()), will appear this question:

2016-10-23 16:36:04 [main - SendThread (localhost: 2181) : 1414357] - [the DEBUG] Got ping response for sessionid: 0 x157f072a16d000f after 0 ms
The 2016-10-23 16:36:07 [main - SendThread (localhost: 2181) : 1417025] - [the DEBUG] Got ping response for sessionid: 0 x157f072a16d000f after 1 ms
The 2016-10-23 16:36:10 [main - SendThread (localhost: 2181) : 1419692] - [the DEBUG] Got ping response for sessionid: 0 x157f072a16d000f after 1 ms
The 2016-10-23 16:36:12 [main - SendThread (localhost: 2181) : 1422359] - [the DEBUG] Got ping response for sessionid: 0 x157f072a16d000f after 1 ms

And every 1 seconds to print once this, the program is not down, kafka even not zookeeper on?

The great god help see ah,,,, this don't know how to solve,,

CodePudding user response:

Looks like your log4j set is the DEBUG log, adjustable high,