Fix some problems with watches that return a v1.Status

Author: brendandburns

util/src/main/java/io/kubernetes/client/util/Watch.java

@@ -12,16 +12,17 @@
 */
 package io.kubernetes.client.util;
 
-import com.google.gson.JsonParseException;
 import com.google.gson.annotations.SerializedName;
 import com.google.gson.reflect.TypeToken;
+import com.google.gson.stream.JsonReader;
 import com.squareup.okhttp.Call;
 import com.squareup.okhttp.ResponseBody;
 import io.kubernetes.client.ApiClient;
 import io.kubernetes.client.ApiException;
 import io.kubernetes.client.JSON;
 import io.kubernetes.client.models.V1Status;
 import java.io.IOException;
+import java.io.StringReader;
 import java.lang.reflect.Type;
 import java.util.Iterator;
 import org.slf4j.Logger;
@@ -112,7 +113,7 @@ public static <T> Watch<T> createWatch(ApiClient client, Call call, Type watchTy
     }
   }
 
-  private Watch(JSON json, ResponseBody body, Type watchType, Call call) {
+  protected Watch(JSON json, ResponseBody body, Type watchType, Call call) {
     this.response = body;
     this.watchType = watchType;
     this.json = json;
@@ -125,18 +126,60 @@ public Response<T> next() {
       if (line == null) {
         throw new RuntimeException("Null response from the server.");
       }
-      try {
-        return json.deserialize(line, watchType);
-      } catch (JsonParseException ex) {
-        Type statusType = new TypeToken<Response<V1Status>>() {}.getType();
-        Response<V1Status> status = json.deserialize(line, statusType);
-        return new Response<T>(status.type, status.object);
-      }
+      return parseLine(line);
     } catch (IOException e) {
       throw new RuntimeException("IO Exception during next method.", e);
     }
   }
 
+  protected boolean isStatus(String line) throws IOException {
+    boolean found = false;
+    JsonReader reader = new JsonReader(new StringReader(line));
+    reader.beginObject();
+    // extract object data.
+    while (reader.hasNext()) {
+      String name = reader.nextName();
+      if (name.equals("object")) {
+        found = true;
+        break;
+      }
+      reader.skipValue();
+    }
+    if (!found) {
+      return false;
+    }
+
+    String kind = null;
+    String apiVersion = null;
+    reader.beginObject();
+    while (reader.hasNext()) {
+      String name = reader.nextName();
+      if (name.equals("kind")) {
+        kind = reader.nextString();
+      } else if (name.equals("apiVersion")) {
+        apiVersion = reader.nextString();
+      } else {
+        reader.skipValue();
+      }
+      if (apiVersion != null && kind != null) {
+        break;
+      }
+    }
+    if ("Status".equals(kind) && "v1".equals(apiVersion)) {
+      return true;
+    }
+    return false;
+  }
+
+  protected Response<T> parseLine(String line) throws IOException {
+    if (!isStatus(line)) {
+      return json.deserialize(line, watchType);
+    }
+    Type statusType = new TypeToken<Response<V1Status>>() {}.getType();
+    Response<V1Status> status = json.deserialize(line, statusType);
+    return new Response<T>(status.type, status.object);
+  }
+
   public boolean hasNext() {
     try {
       return !response.source().exhausted();

util/src/test/java/io/kubernetes/client/util/WatchTest.java

@@ -0,0 +1,52 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+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.kubernetes.client.util;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.gson.JsonObject;
+import com.google.gson.reflect.TypeToken;
+import io.kubernetes.client.JSON;
+import io.kubernetes.client.models.V1ConfigMap;
+import java.io.IOException;
+import org.junit.Test;
+
+/** Tests for the Watch helper class */
+public class WatchTest {
+  @Test
+  public void testWatchEnd() throws IOException {
+    JSON json = new JSON();
+    Watch<V1ConfigMap> watch =
+        new Watch<V1ConfigMap>(
+            json, null, new TypeToken<Watch.Response<V1ConfigMap>>() {}.getType(), null);
+    JsonObject metadata = new JsonObject();
+    metadata.addProperty("name", "foo");
+    metadata.addProperty("namespace", "bar");
+
+    JsonObject status = new JsonObject();
+    status.add("metadata", metadata);
+    status.addProperty("kind", "Status");
+    status.addProperty("apiVersion", "v1");
+    status.addProperty("status", "failure");
+    status.addProperty("message", "too old resource version");
+    status.addProperty("reason", "Gone");
+    status.addProperty("code", 410);
+
+    JsonObject obj = new JsonObject();
+    obj.addProperty("type", "ERR");
+    obj.add("object", status);
+    String data = json.getGson().toJson(obj);
+    Watch.Response<V1ConfigMap> response = watch.parseLine(data);
+    assertEquals(null, response.object);
+  }
+}