public MessageSender getSender(long waitTimeMillis) {
		try {
            if (!freeSender.tryAcquire(waitTimeMillis, TimeUnit.MILLISECONDS))
                throw new RuntimeException("Timeout waiting for idle object in the pool.");

        } catch (InterruptedException e) {
            throw new RuntimeException("Interrupted waiting for idle object in the pool .");
        } 
		closingLock.readLock().lock();
		MessageSender sender = queue.poll();
		if(sender==null)
		{
			sender = new MessageSenderImpl(props, topic, this);
			if(sender!=null)
			{
				LOGGER.info("Add new sender to the pool.");
				queue.offer(sender);
			}
		}
		closingLock.readLock().unlock();
		return sender;
	}