}
catch(WakeupException e)
{
- log.info("{} - RIIING!", id);
+ log.info("{} - Closing the KafkaConsumer", id);
+ consumer.close();
}
catch(Exception e)
{
}
finally
{
+ log.info("{} - Shutting down...");
this.lock.lock();
try
{
- log.info("{} - Closing the KafkaConsumer", id);
- consumer.close();
log.info("{} - DONE!");
stopped.signal();
}