test_class.file: flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueueTest.java
test_class.identifier: ClosableBlockingQueueTest
test_case.body: @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) {
  				// expected
  			}
  		}
  		catch (Exception e) {
  			e.printStackTrace();
  			fail(e.getMessage());
  		}
  	}
test_case.class_method_signature: ClosableBlockingQueueTest.testPollBatch()
test_case.full_signature: @Test public void testPollBatch()
test_case.identifier: testPollBatch
test_case.invocations: assertNull pollBatch add add assertEquals asList pollBatch assertNull pollBatch add assertEquals singletonList pollBatch assertNull pollBatch assertTrue close pollBatch fail printStackTrace fail getMessage
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testPollBatch()
focal_class.declarator: lock
focal_class.modifier: private final
focal_class.original_string: private final ReentrantLock lock;
focal_class.type: ReentrantLock
focal_class.var_name: lock
focal_class.declarator: nonEmpty
focal_class.modifier: private final
focal_class.original_string: private final Condition nonEmpty;
focal_class.type: Condition
focal_class.var_name: nonEmpty
focal_class.declarator: elements
focal_class.modifier: private final
focal_class.original_string: private final ArrayDeque<E> elements;
focal_class.type: ArrayDeque<E>
focal_class.var_name: elements
focal_class.declarator: open
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean open;
focal_class.type: boolean
focal_class.var_name: open
focal_class.file: flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java
focal_class.identifier: ClosableBlockingQueue
focal_class.class_method_signature: ClosableBlockingQueue.ClosableBlockingQueue()
focal_class.full_signature: public  ClosableBlockingQueue()
focal_class.identifier: ClosableBlockingQueue
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.signature:  ClosableBlockingQueue()
focal_class.class_method_signature: ClosableBlockingQueue.ClosableBlockingQueue(int initialSize)
focal_class.full_signature: public  ClosableBlockingQueue(int initialSize)
focal_class.identifier: ClosableBlockingQueue
focal_class.modifiers: public
focal_class.parameters: (int initialSize)
focal_class.signature:  ClosableBlockingQueue(int initialSize)
focal_class.class_method_signature: ClosableBlockingQueue.ClosableBlockingQueue(Collection<? extends E> initialElements)
focal_class.full_signature: public  ClosableBlockingQueue(Collection<? extends E> initialElements)
focal_class.identifier: ClosableBlockingQueue
focal_class.modifiers: public
focal_class.parameters: (Collection<? extends E> initialElements)
focal_class.signature:  ClosableBlockingQueue(Collection<? extends E> initialElements)
focal_class.class_method_signature: ClosableBlockingQueue.size()
focal_class.full_signature: public int size()
focal_class.identifier: size
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: int
focal_class.signature: int size()
focal_class.class_method_signature: ClosableBlockingQueue.isEmpty()
focal_class.full_signature: public boolean isEmpty()
focal_class.identifier: isEmpty
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean isEmpty()
focal_class.class_method_signature: ClosableBlockingQueue.isOpen()
focal_class.full_signature: public boolean isOpen()
focal_class.identifier: isOpen
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean isOpen()
focal_class.class_method_signature: ClosableBlockingQueue.close()
focal_class.full_signature: public boolean close()
focal_class.identifier: close
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean close()
focal_class.class_method_signature: ClosableBlockingQueue.addIfOpen(E element)
focal_class.full_signature: public boolean addIfOpen(E element)
focal_class.identifier: addIfOpen
focal_class.modifiers: public
focal_class.parameters: (E element)
focal_class.return: boolean
focal_class.signature: boolean addIfOpen(E element)
focal_class.class_method_signature: ClosableBlockingQueue.add(E element)
focal_class.full_signature: public void add(E element)
focal_class.identifier: add
focal_class.modifiers: public
focal_class.parameters: (E element)
focal_class.return: void
focal_class.signature: void add(E element)
focal_class.class_method_signature: ClosableBlockingQueue.peek()
focal_class.full_signature: public E peek()
focal_class.identifier: peek
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: E
focal_class.signature: E peek()
focal_class.class_method_signature: ClosableBlockingQueue.poll()
focal_class.full_signature: public E poll()
focal_class.identifier: poll
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: E
focal_class.signature: E poll()
focal_class.class_method_signature: ClosableBlockingQueue.pollBatch()
focal_class.full_signature: public List<E> pollBatch()
focal_class.identifier: pollBatch
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: List<E>
focal_class.signature: List<E> pollBatch()
focal_class.class_method_signature: ClosableBlockingQueue.getElementBlocking()
focal_class.full_signature: public E getElementBlocking()
focal_class.identifier: getElementBlocking
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: E
focal_class.signature: E getElementBlocking()
focal_class.class_method_signature: ClosableBlockingQueue.getElementBlocking(long timeoutMillis)
focal_class.full_signature: public E getElementBlocking(long timeoutMillis)
focal_class.identifier: getElementBlocking
focal_class.modifiers: public
focal_class.parameters: (long timeoutMillis)
focal_class.return: E
focal_class.signature: E getElementBlocking(long timeoutMillis)
focal_class.class_method_signature: ClosableBlockingQueue.getBatchBlocking()
focal_class.full_signature: public List<E> getBatchBlocking()
focal_class.identifier: getBatchBlocking
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: List<E>
focal_class.signature: List<E> getBatchBlocking()
focal_class.class_method_signature: ClosableBlockingQueue.getBatchBlocking(long timeoutMillis)
focal_class.full_signature: public List<E> getBatchBlocking(long timeoutMillis)
focal_class.identifier: getBatchBlocking
focal_class.modifiers: public
focal_class.parameters: (long timeoutMillis)
focal_class.return: List<E>
focal_class.signature: List<E> getBatchBlocking(long timeoutMillis)
focal_class.class_method_signature: ClosableBlockingQueue.hashCode()
focal_class.full_signature: @Override public int hashCode()
focal_class.identifier: hashCode
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: int
focal_class.signature: int hashCode()
focal_class.class_method_signature: ClosableBlockingQueue.equals(Object obj)
focal_class.full_signature: @Override public boolean equals(Object obj)
focal_class.identifier: equals
focal_class.modifiers: @Override public
focal_class.parameters: (Object obj)
focal_class.return: boolean
focal_class.signature: boolean equals(Object obj)
focal_class.class_method_signature: ClosableBlockingQueue.toString()
focal_class.full_signature: @Override public String toString()
focal_class.identifier: toString
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String toString()
focal_method.body: public List<E> pollBatch() {
  		lock.lock();
  		try {
  			if (open) {
  				if (elements.size() > 0) {
  					ArrayList<E> result = new ArrayList<>(elements);
  					elements.clear();
  					return result;
  				} else {
  					return null;
  				}
  			} else {
  				throw new IllegalStateException("queue is closed");
  			}
  		} finally {
  			lock.unlock();
  		}
  	}
focal_method.class_method_signature: ClosableBlockingQueue.pollBatch()
focal_method.full_signature: public List<E> pollBatch()
focal_method.identifier: pollBatch
focal_method.invocations: lock size clear unlock
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: List<E>
focal_method.signature: List<E> pollBatch()