Merge pull request #533 from oracle/fix-take6

Suppress outdated watch events

Author: rjeberhard

operator/src/main/java/oracle/kubernetes/operator/Watcher.java

@@ -29,10 +29,10 @@
 abstract class Watcher<T> {
   static final String HAS_NEXT_EXCEPTION_MESSAGE = "IO Exception during hasNext method.";
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
-  private static final String IGNORED_RESOURCE_VERSION = "0";
+  private static final long IGNORED_RESOURCE_VERSION = 0;
 
   private final AtomicBoolean isDraining = new AtomicBoolean(false);
-  private String resourceVersion;
+  private Long resourceVersion;
   private AtomicBoolean stopping;
   private WatchListener<T> listener;
   private Thread thread = null;
@@ -45,7 +45,8 @@ abstract class Watcher<T> {
    * @param stopping an atomic boolean to watch to determine when to stop the watcher
    */
   Watcher(String resourceVersion, AtomicBoolean stopping) {
-    this.resourceVersion = resourceVersion;
+    this.resourceVersion =
+        !isNullOrEmptyString(resourceVersion) ? Long.parseLong(resourceVersion) : 0;
     this.stopping = stopping;
   }
 
@@ -110,7 +111,8 @@ protected boolean isStopping() {
   }
 
   private void watchForEvents() {
-    try (WatchI<T> watch = initiateWatch(new WatchBuilder().withResourceVersion(resourceVersion))) {
+    try (WatchI<T> watch =
+        initiateWatch(new WatchBuilder().withResourceVersion(resourceVersion.toString()))) {
       while (watch.hasNext()) {
         Watch.Response<T> item = watch.next();
 
@@ -139,9 +141,11 @@ private boolean isError(Watch.Response<T> item) {
   }
 
   private void handleRegularUpdate(Watch.Response<T> item) {
-    LOGGER.fine(MessageKeys.WATCH_EVENT, item.type, item.object);
-    trackResourceVersion(item.type, item.object);
-    if (listener != null) listener.receivedResponse(item);
+    if (isFresh(item.type, item.object)) {
+      LOGGER.fine(MessageKeys.WATCH_EVENT, item.type, item.object);
+      trackResourceVersion(item.type, item.object);
+      if (listener != null) listener.receivedResponse(item);
+    }
   }
 
   private void handleErrorResponse(Watch.Response<T> item) {
@@ -152,7 +156,8 @@ private void handleErrorResponse(Watch.Response<T> item) {
       if (index1 > 0) {
         int index2 = message.indexOf(')', index1 + 1);
         if (index2 > 0) {
-          resourceVersion = message.substring(index1 + 1, index2);
+          String val = message.substring(index1 + 1, index2);
+          resourceVersion = !isNullOrEmptyString(val) ? Long.parseLong(val) : 0;
         }
       }
     }
@@ -170,27 +175,34 @@ private void trackResourceVersion(String type, Object object) {
     updateResourceVersion(getNewResourceVersion(type, object));
   }
 
-  private String getNewResourceVersion(String type, Object object) {
-    if (type.equalsIgnoreCase("DELETED"))
-      return Integer.toString(1 + Integer.parseInt(resourceVersion));
+  private long getNewResourceVersion(String type, Object object) {
+    if (type.equalsIgnoreCase("DELETED")) return 1 + resourceVersion;
     else return getResourceVersionFromMetadata(object);
   }
 
-  private String getResourceVersionFromMetadata(Object object) {
+  private long getResourceVersionFromMetadata(Object object) {
     try {
       Method getMetadata = object.getClass().getDeclaredMethod("getMetadata");
       V1ObjectMeta metadata = (V1ObjectMeta) getMetadata.invoke(object);
-      return metadata.getResourceVersion();
+      String val = metadata.getResourceVersion();
+      return !isNullOrEmptyString(val) ? Long.parseLong(val) : 0;
     } catch (Exception e) {
       LOGGER.warning(MessageKeys.EXCEPTION, e);
       return IGNORED_RESOURCE_VERSION;
     }
   }
 
-  private void updateResourceVersion(String newResourceVersion) {
-    if (isNullOrEmptyString(resourceVersion)) resourceVersion = newResourceVersion;
-    else if (newResourceVersion.compareTo(resourceVersion) > 0)
-      resourceVersion = newResourceVersion;
+  private boolean isFresh(String type, Object object) {
+    if (resourceVersion == 0) return true;
+    long newResourceVersion = getResourceVersionFromMetadata(object);
+    return type.equalsIgnoreCase("DELETED")
+        ? newResourceVersion >= resourceVersion
+        : newResourceVersion > resourceVersion;
+  }
+
+  private void updateResourceVersion(long newResourceVersion) {
+    if (resourceVersion == 0) resourceVersion = newResourceVersion;
+    else if (newResourceVersion > resourceVersion) resourceVersion = newResourceVersion;
   }
 
   private static boolean isNullOrEmptyString(String s) {

operator/src/test/java/oracle/kubernetes/operator/WatcherTestBase.java

@@ -5,7 +5,6 @@
 package oracle.kubernetes.operator;
 
 import static java.net.HttpURLConnection.HTTP_GONE;
-import static oracle.kubernetes.operator.builders.EventMatcher.addEvent;
 import static oracle.kubernetes.operator.builders.EventMatcher.modifyEvent;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.contains;
@@ -107,12 +106,13 @@ private Watch.Response createHttpGoneErrorResponse(int nextResourceVersion) {
   @SuppressWarnings({"unchecked", "rawtypes"})
   @Test
   public void receivedEvents_areSentToListeners() {
-    Object object = createObjectWithMetaData();
-    StubWatchFactory.addCallResponses(createAddResponse(object), createModifyResponse(object));
+    Object object1 = createObjectWithMetaData();
+    Object object2 = createObjectWithMetaData();
+    StubWatchFactory.addCallResponses(createAddResponse(object1), createModifyResponse(object2));
 
     createAndRunWatcher(NAMESPACE, stopping, INITIAL_RESOURCE_VERSION);
 
-    assertThat(callBacks, contains(addEvent(object), modifyEvent(object)));
+    assertThat(callBacks, contains(modifyEvent(object2)));
   }
 
   @SuppressWarnings({"rawtypes", "unchecked"})
@@ -121,15 +121,14 @@ public void afterFirstSetOfEvents_nextRequestSendsLastResourceVersion() {
     Object object1 = createObjectWithMetaData();
     Object object2 = createObjectWithMetaData();
     Watch.Response[] firstSet = {createAddResponse(object1), createModifyResponse(object2)};
-    int resourceAfterFirstSet = resourceVersion - 1;
     StubWatchFactory.addCallResponses(firstSet);
     scheduleAddResponse(createObjectWithMetaData());
 
     createAndRunWatcher(NAMESPACE, stopping, INITIAL_RESOURCE_VERSION);
 
     assertThat(
         StubWatchFactory.getRequestParameters().get(1),
-        hasEntry("resourceVersion", Integer.toString(resourceAfterFirstSet)));
+        hasEntry("resourceVersion", String.valueOf(resourceVersion - 2)));
   }
 
   @SuppressWarnings({"unchecked", "rawtypes"})