fix #5015: executing resync as a locked operation

CHANGELOG.md

@@ -12,6 +12,7 @@
 * Fix #5000: Remove clashing `v1alpha` apigroup packages in `istio-model-v1beta1`
 * fix #5002: Jetty response completion accounts for header processing
 * Fix #5009: addressing issue with serialization of wrapped polymorphic types
+* Fix #5015: executing resync as a locking operation to ensure resync event ordering
 * Fix #5020: updating the resourceVersion on a delete with finalizers
 * Fix #5033: port forwarding for clients other than okhttp needs to specify the subprotocol
 * Fix #5044: disable Vert.x instance file caching

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/impl/cache/CacheImpl.java

@@ -396,4 +396,8 @@ public boolean isFullState() {
     return items.isFullState();
   }
 
+  public Object getLockObject() {
+    return this;
+  }
+
 }

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/impl/cache/ProcessorStore.java

@@ -130,8 +130,11 @@ public String getKey(T obj) {
 
   @Override
   public void resync() {
-    this.cache.list()
-        .forEach(i -> this.processor.distribute(new ProcessorListener.UpdateNotification<>(i, i), true));
+    // lock to ensure the ordering wrt other events
+    synchronized (cache.getLockObject()) {
+      this.cache.list()
+          .forEach(i -> this.processor.distribute(new ProcessorListener.UpdateNotification<>(i, i), true));
+    }
   }
 
 }

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/impl/cache/SharedProcessor.java

@@ -56,6 +56,10 @@ public SharedProcessor() {
   }
 
   public SharedProcessor(Executor executor, String informerDescription) {
+    // serialexecutors are by default unbounded, we expect this behavior here
+    // because resync may flood the executor with events for large caches
+    // if we ever need to limit the queue size, we have to revisit the
+    // resync locking behavior
     this.executor = new SerialExecutor(executor);
     this.informerDescription = informerDescription;
   }

kubernetes-client/src/test/java/io/fabric8/kubernetes/client/informers/impl/cache/ProcessorStoreTest.java

@@ -118,4 +118,25 @@ void testSyncEvents() {
     assertTrue(syncCaptor.getAllValues().subList(4, 6).stream().allMatch(s -> !s.booleanValue()));
   }
 
+  @Test
+  void testResyncLock() throws InterruptedException {
+    CacheImpl<Pod> podCache = new CacheImpl<>();
+    SharedProcessor<Pod> processor = Mockito.mock(SharedProcessor.class);
+
+    ProcessorStore<Pod> processorStore = new ProcessorStore<>(podCache, processor);
+
+    Pod pod = new PodBuilder().withNewMetadata().withName("pod1").withResourceVersion("1").endMetadata().build();
+
+    List<Pod> pods = Arrays.asList(pod);
+    processorStore.update(pods);
+
+    Mockito.doAnswer(invocation -> {
+      assertTrue(Thread.holdsLock(podCache.getLockObject()));
+      return null;
+    }).when(processor).distribute(Mockito.any(ProcessorListener.Notification.class), Mockito.anyBoolean());
+    processorStore.resync();
+
+    Mockito.verify(processor).distribute(Mockito.any(ProcessorListener.Notification.class), Mockito.anyBoolean());
+  }
+
 }