removing the deltafifo to simplify the informer code

it's not in these changes, but it will allow the cache interface to
become fully type safe

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

@@ -372,10 +372,6 @@ public void addIndexFunc(String indexName, Function<T, List<String>> indexFunc)
    * @return the key
    */
   public static <T> String deletionHandlingMetaNamespaceKeyFunc(T object) {
-    if (object instanceof DeltaFIFO.DeletedFinalStateUnknown) {
-      DeltaFIFO.DeletedFinalStateUnknown deleteObj = (DeltaFIFO.DeletedFinalStateUnknown) object;
-      return deleteObj.getKey();
-    }
     return metaNamespaceKeyFunc(object);
   }
 

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

@@ -24,14 +24,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.AbstractMap;
-import java.util.Deque;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 /**
@@ -48,19 +45,14 @@ public class Controller<T extends HasMetadata, L extends KubernetesResourceList<
    */
   private final long fullResyncPeriod;
 
-  /**
-   * Queue stores deltas produced by reflector.
-   */
-  private final DeltaFIFO<T> queue;
+  private final Store<T> store;
 
   private final ListerWatcher<T, L> listerWatcher;
 
   private Reflector<T, L> reflector;
 
   private final Supplier<Boolean> resyncFunc;
 
-  private final Consumer<Deque<AbstractMap.SimpleEntry<DeltaFIFO.DeltaType, Object>>> processFunc;
-
   private final ScheduledExecutorService resyncExecutor;
 
   private ScheduledFuture resyncFuture;
@@ -71,13 +63,10 @@ public class Controller<T extends HasMetadata, L extends KubernetesResourceList<
 
   private final Class<T> apiTypeClass;
 
-  private volatile boolean running;
-
-  public Controller(Class<T> apiTypeClass, DeltaFIFO<T> queue, ListerWatcher<T, L> listerWatcher, Consumer<Deque<AbstractMap.SimpleEntry<DeltaFIFO.DeltaType, Object>>> processFunc, Supplier<Boolean> resyncFunc, long fullResyncPeriod, OperationContext context, ConcurrentLinkedQueue<SharedInformerEventListener> eventListeners) {
-    this.queue = queue;
+  public Controller(Class<T> apiTypeClass, Store<T> store, ListerWatcher<T, L> listerWatcher, Supplier<Boolean> resyncFunc, long fullResyncPeriod, OperationContext context, ConcurrentLinkedQueue<SharedInformerEventListener> eventListeners) {
+    this.store = store;
     this.listerWatcher = listerWatcher;
     this.apiTypeClass = apiTypeClass;
-    this.processFunc = processFunc;
     this.resyncFunc = resyncFunc;
     if (fullResyncPeriod < 0) {
       throw new IllegalArgumentException("Invalid resync period provided, It should be a non-negative value");
@@ -88,14 +77,14 @@ public Controller(Class<T> apiTypeClass, DeltaFIFO<T> queue, ListerWatcher<T, L>
 
     // Starts one daemon thread for resync
     this.resyncExecutor = Executors.newSingleThreadScheduledExecutor();
-    this.reflector = new Reflector<>(apiTypeClass, listerWatcher, queue, operationContext);
+    this.reflector = new Reflector<>(apiTypeClass, listerWatcher, store, operationContext);
   }
 
   public void run() {
     log.info("informer#Controller: ready to run resync and reflector runnable");
     // Start the resync runnable
     if (fullResyncPeriod > 0) {
-      ResyncRunnable resyncRunnable = new ResyncRunnable(queue, resyncFunc);
+      ResyncRunnable resyncRunnable = new ResyncRunnable(store, resyncFunc);
       if(!resyncExecutor.isShutdown()) {
         resyncFuture = resyncExecutor.scheduleAtFixedRate(resyncRunnable, fullResyncPeriod, fullResyncPeriod, TimeUnit.MILLISECONDS);
       }
@@ -104,16 +93,10 @@ public void run() {
     }
 
     try {
-      running = true;
       reflector.listAndWatch();
-
-      // Start the process loop
-      this.processLoop();
     } catch (Exception exception) {
       log.warn("Reflector list-watching job exiting because the thread-pool is shutting down", exception);
       this.eventListeners.forEach(listener -> listener.onException(exception));
-    } finally {
-      running = false;
     }
   }
 
@@ -135,7 +118,7 @@ public void stop() {
    * @return boolean value about queue sync status
    */
   public boolean hasSynced() {
-    return this.queue.hasSynced();
+    return this.store.hasSynced();
   }
 
   /**
@@ -150,30 +133,12 @@ Reflector<T, L> getReflector() {
     return reflector;
   }
 
-  /**
-   * drains the work queue.
-   */
-  private void processLoop() throws Exception {
-    while (!Thread.currentThread().isInterrupted()) {
-      try {
-        this.queue.pop(this.processFunc);
-      } catch (InterruptedException t) {
-        log.debug("DefaultController#processLoop got interrupted: {}", t.getMessage());
-        Thread.currentThread().interrupt();
-        return;
-      } catch (Exception e) {
-        log.error("DefaultController#processLoop recovered from crashing {} ", e.getMessage(), e);
-        throw e;
-      }
-    }
-  }
-
   ScheduledExecutorService getResyncExecutor() {
     return this.resyncExecutor;
   }
 
   public boolean isRunning() {
-    return running && this.reflector.isRunning();
+    return this.reflector.isRunning();
   }
 
   public long getFullResyncPeriod() {