Added updates from code review

belugabehr · belugabehr · commit c63ab852c954 · 2019-10-10T14:57:06.000-04:00
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java b/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java
@@ -54,12 +54,15 @@ public class CircularBlockingQueue<E> implements BlockingQueue<E> {
 
   private final int maxSize;
 
+  private long droppedCount;
+
   public CircularBlockingQueue(int queueSize) {
     this.queue = new ArrayDeque<>(queueSize);
     this.maxSize = queueSize;
 
     this.lock =  new ReentrantLock();
     this.notEmpty = this.lock.newCondition();
+    this.droppedCount = 0L;
   }
 
   /**
@@ -78,7 +81,9 @@ public boolean offer(E e) {
       try {
           if (this.queue.size() == this.maxSize) {
               final E discard = this.queue.remove();
-              LOG.debug("Queue if full. Discarding oldest element: {}", discard);
+              this.droppedCount++;
+              LOG.debug("Queue is full. Discarding oldest element [count={}]: {}",
+                  this.droppedCount, discard);
           }
           this.queue.add(e);
           this.notEmpty.signal();
@@ -142,11 +147,38 @@ public int size() {
     }
   }
 
+  /**
+   * Returns the number of elements that were dropped from the queue because the
+   * queue was full when a new element was offered.
+   *
+   * @return The number of elements dropped (lost) from the queue
+   */
+  public long getDroppedCount() {
+    return this.droppedCount;
+  }
+
+  /**
+   * For testing purposes only.
+   *
+   * @return True if a thread is blocked waiting for a new element to be offered
+   *         to the queue
+   */
+  boolean isConsumerThreadBlocked() {
+    final ReentrantLock lock = this.lock;
+    lock.lock();
+    try {
+      return lock.getWaitQueueLength(this.notEmpty) > 0;
+    } finally {
+      lock.unlock();
+    }
+  }
+
   @Override
   public int drainTo(Collection<? super E> c) {
     throw new UnsupportedOperationException();
   }
 
+
   @Override
   public E poll() {
     throw new UnsupportedOperationException();
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java b/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java
@@ -16,23 +16,23 @@
  * limitations under the License.
  */
 
-package org.apache.zookeeper.server.util;
+package org.apache.zookeeper.util;
 
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
-import org.apache.zookeeper.util.CircularBlockingQueue;
 import org.junit.Assert;
 import org.junit.Test;
 
 public class TestCircularBlockingQueue {
 
   @Test
   public void testCircularBlockingQueue() throws InterruptedException {
-    final BlockingQueue<Integer> testQueue = new CircularBlockingQueue<>(2);
+    final CircularBlockingQueue<Integer> testQueue =
+        new CircularBlockingQueue<>(2);
+
     testQueue.offer(1);
     testQueue.offer(2);
     testQueue.offer(3);
@@ -42,6 +42,7 @@ public void testCircularBlockingQueue() throws InterruptedException {
     Assert.assertEquals(2, testQueue.take().intValue());
     Assert.assertEquals(3, testQueue.take().intValue());
 
+    Assert.assertEquals(1L, testQueue.getDroppedCount());
     Assert.assertEquals(0, testQueue.size());
     Assert.assertEquals(true, testQueue.isEmpty());
   }
@@ -50,7 +51,7 @@ public void testCircularBlockingQueue() throws InterruptedException {
   public void testCircularBlockingQueueTakeBlock()
       throws InterruptedException, ExecutionException {
 
-    final BlockingQueue<Integer> testQueue = new CircularBlockingQueue<>(2);
+    final CircularBlockingQueue<Integer> testQueue = new CircularBlockingQueue<>(2);
 
     ExecutorService executor = Executors.newSingleThreadExecutor();
 
@@ -59,7 +60,9 @@ public void testCircularBlockingQueueTakeBlock()
     });
 
     // Allow the other thread to get into position; waiting for item to be inserted
-    Thread.sleep(2000L);
+    while (!testQueue.isConsumerThreadBlocked()) {
+      Thread.sleep(50L);
+    }
 
     testQueue.offer(10);
 
