ClosableBlockingQueue { public List<E> pollBatch() { lock.lock(); try { if (open) { if (elements.size() > 0) { ArrayList<E> result = new ArrayList<E>(elements); elements.clear(); return result; } else { return null; } } else { throw new IllegalStateException("queue is closed"); } } finally { lock.unlock(); } } ClosableBlockingQueue();  ClosableBlockingQueue(int initialSize);  ClosableBlockingQueue(Collection<? extends E> initialElements); int size(); boolean isEmpty(); boolean isOpen(); boolean close(); boolean addIfOpen(E element); void add(E element); E peek(); E poll(); List<E> pollBatch(); E getElementBlocking(); E getElementBlocking(long timeoutMillis); List<E> getBatchBlocking(); List<E> getBatchBlocking(long timeoutMillis); @Override int hashCode(); @Override boolean equals(Object obj); @Override String toString();  }
@Test public void testPollBatch() { try { ClosableBlockingQueue<String> queue = new ClosableBlockingQueue<>(); assertNull(queue.pollBatch()); queue.add("a"); queue.add("b"); assertEquals(asList("a", "b"), queue.pollBatch()); assertNull(queue.pollBatch()); queue.add("c"); assertEquals(singletonList("c"), queue.pollBatch()); assertNull(queue.pollBatch()); assertTrue(queue.close()); try { queue.pollBatch(); fail("should cause an exception"); } catch (IllegalStateException ignored) { } } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }