public synchronized void close() {
		// lock the thread for closing
		closingLock.writeLock().lock();
		List<Callable<Boolean>> taskList = new ArrayList<>();
		int size = queue.size();
		final CountDownLatch count = new CountDownLatch(queue.size());
		for(int i=0;i<size;i++)
		{
			taskList.add(new Callable<Boolean>() {

				@Override
				public Boolean call() throws Exception {
					MessageSender sender = queue.poll();
					if(sender!=null)
					{
						sender.shutDown();
						count.countDown();
					}
					return true;
				}
			});
		}
			
		try {
			pool.invokeAll(taskList);
			count.await(KafkaConstants.INIT_TIMEOUT_MIN, TimeUnit.MINUTES);
			pool.shutdownNow();
		} catch (InterruptedException e) {
			LOGGER.error("Failed to close the MessageSenderPool", e);
		}
		closingLock.writeLock().unlock();
	}