Merge pull request #669 from cizezsy/fix/capacity-limit

Fix: remove the capacity limit of ProcessorListener

Author: k8s-ci-robot

util/src/main/java/io/kubernetes/client/informer/cache/ProcessorListener.java

@@ -2,23 +2,20 @@
 
 import io.kubernetes.client.informer.ResourceEventHandler;
 import io.kubernetes.client.informer.exception.BadNotificationException;
-import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 import org.joda.time.DateTime;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * ProcessorListener implements Runnable interface. It's supposed to run in background and actually
- * executes its event handler on notification. Note that it allows 1000 pending notifications at
- * maximum.
+ * executes its event handler on notification.
  */
 public class ProcessorListener<ApiType> implements Runnable {
 
   private static final Logger log = LoggerFactory.getLogger(ProcessorListener.class);
 
-  private static final int DEFAULT_QUEUE_CAPACITY = 1000;
-
   // resyncPeriod is how frequently the listener wants a full resync from the shared informer. This
   // value may differ from requestedResyncPeriod if the shared informer adjusts it to align with the
   // informer's overall resync check period.
@@ -33,7 +30,7 @@ public ProcessorListener(ResourceEventHandler<ApiType> handler, long resyncPerio
     this.resyncPeriod = resyncPeriod;
     this.handler = handler;
 
-    this.queue = new ArrayBlockingQueue<>(DEFAULT_QUEUE_CAPACITY);
+    this.queue = new LinkedBlockingQueue<>();
 
     determineNextResync(DateTime.now());
   }
@@ -90,9 +87,7 @@ public void add(Notification<ApiType> obj) {
     if (obj == null) {
       return;
     }
-    if (!this.queue.offer(obj)) {
-      log.warn("notification queue full!");
-    }
+    this.queue.add(obj);
   }
 
   public void determineNextResync(DateTime now) {

util/src/test/java/io/kubernetes/client/informer/cache/ProcessorListenerTest.java

@@ -60,4 +60,39 @@ public void onDelete(V1Pod obj, boolean deletedFinalStateUnknown) {
     assertTrue(updateNotificationReceived);
     assertTrue(deleteNotificationReceived);
   }
+
+  @Test
+  public void testMultipleNotificationsHandling() throws InterruptedException {
+    V1Pod pod = new V1Pod().metadata(new V1ObjectMeta().name("foo").namespace("default"));
+    final int[] count = {0};
+
+    ProcessorListener<V1Pod> listener =
+        new ProcessorListener<>(
+            new ResourceEventHandler<V1Pod>() {
+              @Override
+              public void onAdd(V1Pod obj) {
+                assertEquals(pod, obj);
+                count[0]++;
+              }
+
+              @Override
+              public void onUpdate(V1Pod oldObj, V1Pod newObj) {}
+
+              @Override
+              public void onDelete(V1Pod obj, boolean deletedFinalStateUnknown) {}
+            },
+            0);
+
+    for (int i = 0; i < 2000; i++) {
+      listener.add(new ProcessorListener.AddNotification<>(pod));
+    }
+
+    Thread listenerThread = new Thread(listener);
+    listenerThread.setDaemon(true);
+    listenerThread.start();
+
+    Thread.sleep(2000);
+
+    assertEquals(count[0], 2000);
+  }
 }