public void shutdown ( ) { for ( Topic topic : _topics . values ( ) ) { if ( topic . getConsumerConnector ( ) != null ) { topic . getConsumerConnector ( ) . shutdown ( ) ; } topic . getStreamExecutorService ( ) . shutdownNow ( ) ; try { topic . getStreamExecutorService ( ) . awaitTermination ( 60 , TimeUnit . SECONDS ) ; } catch ( InterruptedException e ) { _logger . warn ( "Stream executor service was interrupted while awaiting termination. This should never happen." ) ; } } _logger . debug ( "Pushing unflushed messages back to Kafka." ) ; Producer producer = new Producer ( _configuration ) ; for ( Map . Entry < String , Topic > entry : _topics . entrySet ( ) ) { String topicName = entry . getKey ( ) ; Topic topic = entry . getValue ( ) ; List < String > unflushedMessages = new ArrayList < String > ( ) ; if ( ! topic . getMessages ( ) . isEmpty ( ) ) { topic . getMessages ( ) . drainTo ( unflushedMessages ) ; producer . enqueue ( topicName , unflushedMessages ) ; } _logger . debug ( "{} messages for topic {} enqueued on Kafka queue" , unflushedMessages . size ( ) , topicName ) ; } producer . shutdown ( ) ; }
Enqueue un - flushed messages back on to Kafka .