@Test
public void testConcurrentOperations() throws InterruptedException {

  ActionQueue aq = new ActionQueue();
  String[] hosts = new String[] { "h0", "h1", "h2", "h3", "h4", "h5", "h6",
      "h7", "h8", "h9" };
  ActionQueueOperation[] enqueOperators = new ActionQueueOperation[threadCount];
  ActionQueueOperation[] dequeOperators = new ActionQueueOperation[threadCount];
  ActionQueueOperation[] dequeAllOperators = new ActionQueueOperation[threadCount];
  List<Thread> producers = new ArrayList<Thread>();
  List<Thread> consumers = new ArrayList<Thread>();
  for (int i = 0; i < threadCount; i++) {
    dequeOperators[i] = new ActionQueueOperation(aq, hosts,
        ActionQueueOperation.OpType.DEQUEUE);
    Thread t = new Thread(dequeOperators[i]);
    consumers.add(t);
    t.start();
  }
  for (int i = 0; i < threadCount; i++) {
    enqueOperators[i] = new ActionQueueOperation(aq, hosts,
        ActionQueueOperation.OpType.ENQUEUE);
    Thread t = new Thread(enqueOperators[i]);
    producers.add(t);
    t.start();
  }
  for (int i = 0; i < threadCount; i++) {
    dequeAllOperators[i] = new ActionQueueOperation(aq, hosts,
        ActionQueueOperation.OpType.DEQUEUEALL);
    Thread t = new Thread(dequeAllOperators[i]);
    consumers.add(t);
    t.start();
  }
  // Run for some time
  Thread.sleep(100);
  // Stop the enqueue
  for (int i = 0; i < threadCount; i++) {
    enqueOperators[i].stop();
  }
  for (Thread producer : producers) {
    producer.join();
  }
  // Give time to get everything dequeued
  boolean allDequeued = false;
  while (!allDequeued) {
    Thread.sleep(10);
    allDequeued = true;
    for (String host: hosts) {
      if (aq.size(host) > 0) {
        allDequeued = false;
        break;
      }
    }
  }
  // Stop all threads
  for (int i = 0; i < threadCount; i++) {
    dequeOperators[i].stop();
    dequeAllOperators[i].stop();
  }
  for (Thread consumer : consumers) {
    consumer.join();
  }
  for (int h = 0; h<hosts.length; h++) {
    long opsEnqueued = 0;
    long opsDequeued = 0;
    for (int i = 0; i < threadCount; i++) {
      opsEnqueued += enqueOperators[i].getOpCounts()[h];
      opsDequeued += dequeOperators[i].getOpCounts()[h];
      opsDequeued += dequeAllOperators[i].getOpCounts()[h];
    }
    assertTrue(opsEnqueued != 0); //Prevent degenerate case of all zeros.
    assertEquals(0, aq.size(hosts[h])); //Everything should be dequeued
    LOG.info("Host: " + hosts[h] + ", opsEnqueued: " + opsEnqueued
        + ", opsDequeued: " + opsDequeued);
    assertEquals(opsDequeued, opsEnqueued);
  }
}