refactor: Watcher.onClose has dedicated WatcherException as parameter.

Author: manusa

CHANGELOG.md

@@ -5,6 +5,7 @@
 #### Bugs
 
 #### Improvements
+* Fix #2614: Watcher.onClose has dedicated WatcherException as parameter.
 
 #### Dependency Upgrade
 

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/Watcher.java

@@ -24,7 +24,7 @@ public interface Watcher<T> {
    *
    * @param cause What caused the watcher to be closed. Null means normal close.
    */
-  void onClose(KubernetesClientException cause);
+  void onClose(WatcherException cause);
 
   enum Action {
     ADDED, MODIFIED, DELETED, ERROR

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/WatcherException.java

@@ -0,0 +1,45 @@
+/**
+ * Copyright (C) 2015 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.fabric8.kubernetes.client;
+
+import java.net.HttpURLConnection;
+
+public class WatcherException extends Exception {
+
+  public WatcherException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public WatcherException(String message) {
+    super(message);
+  }
+
+  public KubernetesClientException asClientException() {
+    final Throwable cause = getCause();
+    return getCause() instanceof KubernetesClientException ?
+      (KubernetesClientException) cause : new KubernetesClientException(getMessage(), cause);
+  }
+
+  public boolean isHttpGone() {
+    final KubernetesClientException cause = asClientException();
+    return cause.getCode() == HttpURLConnection.HTTP_GONE
+      || (cause.getStatus() != null && cause.getStatus().getCode() == HttpURLConnection.HTTP_GONE);
+  }
+
+  public boolean isShouldRetry() {
+    return getCause() == null || !isHttpGone();
+  }
+}

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/base/BaseOperation.java

@@ -16,6 +16,7 @@
 package io.fabric8.kubernetes.client.dsl.base;
 
 import io.fabric8.kubernetes.api.model.ObjectReference;
+import io.fabric8.kubernetes.client.WatcherException;
 import io.fabric8.kubernetes.client.utils.CreateOrReplaceHelper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,7 +46,6 @@
 import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
 import io.fabric8.kubernetes.client.dsl.Replaceable;
 import io.fabric8.kubernetes.client.dsl.Resource;
-import io.fabric8.kubernetes.client.dsl.base.WaitForConditionWatcher.WatchException;
 import io.fabric8.kubernetes.client.dsl.internal.DefaultOperationInfo;
 import io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager;
 import io.fabric8.kubernetes.client.dsl.internal.WatchHTTPManager;
@@ -1104,7 +1104,7 @@ private T waitUntilConditionWithRetries(Predicate<T> condition, long timeoutNano
         return watcher.getFuture().get(remainingNanosToWait, NANOSECONDS);
       } catch (ExecutionException e) {
         Throwable cause = e.getCause();
-        if (cause instanceof WatchException && ((WatchException) cause).isShouldRetry()) {
+        if (cause instanceof WatcherException && ((WatcherException) cause).isShouldRetry()) {
           LOG.debug("retryable watch exception encountered, retrying after {} millis", currentBackOff, cause);
           Thread.sleep(currentBackOff);
           currentBackOff *= watchRetryBackoffMultiplier;

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/base/WaitForConditionWatcher.java

@@ -15,13 +15,12 @@
  */
 package io.fabric8.kubernetes.client.dsl.base;
 
-import java.net.HttpURLConnection;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Predicate;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
 
 public class WaitForConditionWatcher<T extends HasMetadata> implements Watcher<T> {
 
@@ -52,34 +51,14 @@ public void eventReceived(Action action, T resource) {
         }
         break;
       case ERROR:
-        future.completeExceptionally(new WatchException("Action.ERROR received"));
+        future.completeExceptionally(new WatcherException("Action.ERROR received"));
         break;
     }
   }
 
   @Override
-  public void onClose(KubernetesClientException cause) {
-    future.completeExceptionally(new WatchException("Watcher closed", cause));
+  public void onClose(WatcherException cause) {
+    future.completeExceptionally(cause == null ? new WatcherException("Watcher closed") : cause);
   }
 
-  public static class WatchException extends Exception {
-
-    public WatchException(String message, KubernetesClientException cause) {
-      super(message, cause);
-    }
-
-    public WatchException(String message) {
-      super(message);
-    }
-
-    public boolean isShouldRetry() {
-      return getCause() == null || !isHttpGone();
-    }
-
-    private boolean isHttpGone() {
-      KubernetesClientException cause = ((KubernetesClientException) getCause());
-      return cause.getCode() == HttpURLConnection.HTTP_GONE
-        || (cause.getStatus() != null && cause.getStatus().getCode() == HttpURLConnection.HTTP_GONE);
-    }
-  }
 }

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/AbstractWatchManager.java

@@ -0,0 +1,78 @@
+/**
+ * Copyright (C) 2015 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.fabric8.kubernetes.client.dsl.internal;
+
+import io.fabric8.kubernetes.api.model.ListOptions;
+import io.fabric8.kubernetes.client.Watch;
+import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
+import okhttp3.OkHttpClient;
+import okhttp3.WebSocket;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+public abstract class AbstractWatchManager<T> implements Watch {
+
+  private static final Logger logger = LoggerFactory.getLogger(AbstractWatchManager.class);
+
+  final Watcher<T> watcher;
+  final int reconnectLimit;
+  final int reconnectInterval;
+  final int maxIntervalExponent;
+  final ListOptions listOptions;
+  final AtomicReference<String> resourceVersion;
+  final OkHttpClient clonedClient;
+
+  private final AtomicBoolean forceClosed;
+
+  AbstractWatchManager(
+    Watcher<T> watcher, ListOptions listOptions, int reconnectLimit, int reconnectInterval, int maxIntervalExponent,
+    OkHttpClient clonedClient
+  ) {
+    this.watcher = watcher;
+    this.listOptions = listOptions;
+    this.reconnectLimit = reconnectLimit;
+    this.reconnectInterval = reconnectInterval;
+    this.maxIntervalExponent = maxIntervalExponent;
+    this.clonedClient = clonedClient;
+    this.resourceVersion = new AtomicReference<>(listOptions.getResourceVersion());
+    this.forceClosed = new AtomicBoolean();
+  }
+
+  final void closeEvent(WatcherException cause) {
+    if (forceClosed.getAndSet(true)) {
+      logger.debug("Ignoring duplicate firing of onClose event");
+      return;
+    }
+    watcher.onClose(cause);
+  }
+
+  static void closeWebSocket(WebSocket webSocket) {
+    if (webSocket != null) {
+      logger.debug("Closing websocket {}", webSocket);
+      try {
+        if (!webSocket.close(1000, null)) {
+          logger.warn("Failed to close websocket");
+        }
+      } catch (IllegalStateException e) {
+        logger.error("Called close on already closed websocket: {} {}", e.getClass(), e.getMessage());
+      }
+    }
+  }
+}

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/RawWatchConnectionManager.java

@@ -19,8 +19,8 @@
 import io.fabric8.kubernetes.api.model.ListOptions;
 import io.fabric8.kubernetes.api.model.Status;
 import io.fabric8.kubernetes.client.KubernetesClientException;
-import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
 import io.fabric8.kubernetes.client.dsl.base.OperationSupport;
 import io.fabric8.kubernetes.client.utils.Utils;
 import okhttp3.HttpUrl;
@@ -53,17 +53,12 @@
  * instead of using a solid type for deserializing events, it uses plain strings.
  *
  */
-public class RawWatchConnectionManager implements Watch {
+public class RawWatchConnectionManager extends AbstractWatchManager<String> {
   private static final Logger logger = LoggerFactory.getLogger(RawWatchConnectionManager.class);
   private ObjectMapper objectMapper;
-  private Watcher<String> watcher;
   private HttpUrl.Builder watchUrlBuilder;
 
   private final AtomicBoolean forceClosed = new AtomicBoolean();
-  private final AtomicReference<String> resourceVersion;
-  private final int reconnectLimit;
-  private final int reconnectInterval;
-  private int maxIntervalExponent;
   private final AtomicInteger currentReconnectAttempt = new AtomicInteger(0);
   private final AtomicReference<WebSocket> webSocketRef = new AtomicReference<>();
   // single threaded serial executor
@@ -75,17 +70,14 @@ public class RawWatchConnectionManager implements Watch {
   private final ArrayBlockingQueue<Object> queue = new ArrayBlockingQueue<>(1);
 
   private WebSocket webSocket;
-  private OkHttpClient clonedClient;
 
   public RawWatchConnectionManager(OkHttpClient okHttpClient, HttpUrl.Builder watchUrlBuilder, ListOptions listOptions, ObjectMapper objectMapper, final Watcher<String> watcher, int reconnectLimit, int reconnectInterval, int maxIntervalExponent)  {
-    this.clonedClient = okHttpClient;
+    super(
+      watcher, listOptions, reconnectLimit, reconnectInterval, maxIntervalExponent,
+      okHttpClient.newBuilder().build()
+    );
     this.watchUrlBuilder = watchUrlBuilder;
     this.objectMapper = objectMapper;
-    this.watcher = watcher;
-    this.reconnectLimit = reconnectLimit;
-    this.reconnectInterval = reconnectInterval;
-    this.resourceVersion = new AtomicReference<>(listOptions.getResourceVersion());
-    this.maxIntervalExponent = maxIntervalExponent;
     executor = Executors.newSingleThreadScheduledExecutor(r -> {
       Thread ret = new Thread(r, "Executor for Watch " + System.identityHashCode(RawWatchConnectionManager.this));
       ret.setDaemon(true);
@@ -156,7 +148,7 @@ public void onClosed(WebSocket webSocket, int code, String reason) {
           return;
         }
         if (currentReconnectAttempt.get() >= reconnectLimit && reconnectLimit >= 0) {
-          closeEvent(new KubernetesClientException("Connection unexpectedly closed"));
+          closeEvent(new WatcherException("Connection unexpectedly closed"));
           return;
         }
         scheduleReconnect();
@@ -204,7 +196,7 @@ public void onFailure(WebSocket webSocket, Throwable t, Response response) {
         }
 
         if (currentReconnectAttempt.get() >= reconnectLimit && reconnectLimit >= 0) {
-          closeEvent(new KubernetesClientException("Connection failure", t));
+          closeEvent(new WatcherException("Connection failure", t));
           return;
         }
 
@@ -242,7 +234,7 @@ public void execute() {
                   // An unexpected error occurred and we didn't even get an onFailure callback.
                   logger.error("Exception in reconnect", e);
                   webSocketRef.set(null);
-                  closeEvent(new KubernetesClientException("Unhandled exception in reconnect attempt", e));
+                  closeEvent(new WatcherException("Unhandled exception in reconnect attempt", e));
                   close();
                 }
               }
@@ -277,27 +269,6 @@ public void close() {
     }
   }
 
-  private void closeEvent(KubernetesClientException cause) {
-    if (forceClosed.getAndSet(true)) {
-      logger.debug("Ignoring duplicate firing of onClose event");
-      return;
-    }
-    watcher.onClose(cause);
-  }
-
-  private void closeWebSocket(WebSocket ws) {
-    if (ws != null) {
-      logger.debug("Closing websocket {}", ws);
-      try {
-        if (!ws.close(1000, null)) {
-          logger.warn("Failed to close websocket");
-        }
-      } catch (IllegalStateException e) {
-        logger.error("Called close on already closed websocket: {} {}", e.getClass(), e.getMessage());
-      }
-    }
-  }
-
   private long nextReconnectInterval() {
     int exponentOfTwo = currentReconnectAttempt.getAndIncrement();
     if (exponentOfTwo > maxIntervalExponent)