SynchronizedQueue implements Queue<I> { public ArrayList<I> trim(int size) { if (trimming.compareAndSet(false, true)) { try { queueLock.lock(); if (queueSize > size) { ArrayList<I> evictedItems = new ArrayList<I>(queueSize - size); while (queueSize > size) { I last = (I)tail.previous; remove(last); evictedItems.add(last); } return evictedItems; } } finally { trimming.set(false); queueLock.unlock(); } } return null; } SynchronizedQueue(Log log); int size(); void assertConsistency(); boolean remove(I item); void add(I item); ArrayList<I> trim(int size);  }
@Test public void testTrim() throws Exception { SynchronizedQueue queue = new SynchronizedQueue(null); CountDownLatch start = new CountDownLatch(1); CountDownLatch finish = new CountDownLatch(1000); for (int i = 0; i < 1000; i++) { Thread t = new Thread(new Runnable() { @Override public void run() { try { start.await(10, TimeUnit.SECONDS); } catch (InterruptedException e) { Assert.fail("CountDownLatch fail to wait"); } for (int j = 0; j < 10000; j++) { if (queue.size() <= 1000) { queue.add(new Item()); queue.trim(1000); } } finish.countDown(); } }); t.start(); } start.countDown(); finish.await(1, TimeUnit.MINUTES); Assert.assertEquals(1000, queue.size()); }