@Override
	public void run() {
		startTime = System.currentTimeMillis();

		List<QueueEntry> batch = new ArrayList<QueueEntry>();

		while (true) {
			batch.clear();

			// Try to take as many as possible without blocking
			queue.drainTo(batch, batchSize);

			// LOG.info("In batch: " + batch.size() + " size after drain: " +
			// queue.size());

			// Read batchSize messages
			while (batch.size() < batchSize) {
				try {
					QueueEntry entry = queue.take();
					batch.add(entry);
				} catch (InterruptedException e) {
					LOG.error("Failed to get message of queue: " + e.getMessage());
				}
			}
			
			// Handle messages
			try {
				batchHandle(batch);
			} catch (Exception e) {
				e.printStackTrace();
				LOG.error("Error while handling batch: " + e.getMessage());				
			}

		}

	}