Deserialize WatchEvents using the specific object type (#3786)

* Deserialize WatchEvents using the specific object type

* fallback

* tested

* sonar

Author: andreaTP

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

@@ -15,12 +15,12 @@
  */
 package io.fabric8.kubernetes.client.dsl.internal;
 
-import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.KubernetesResource;
-import io.fabric8.kubernetes.api.model.KubernetesResourceList;
-import io.fabric8.kubernetes.api.model.ListOptions;
-import io.fabric8.kubernetes.api.model.Status;
-import io.fabric8.kubernetes.api.model.WatchEvent;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.fabric8.kubernetes.api.model.*;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
@@ -232,9 +232,31 @@ public void close() {
     closeRequest();
     cancelReconnect();
   }
+
+  private WatchEvent contextAwareWatchEventDeserializer(String messageSource) {
+    try {
+      return Serialization.unmarshal(messageSource, WatchEvent.class);
+    } catch (Exception ex1) {
+      try {
+        JsonNode json = Serialization.jsonMapper().readTree(messageSource);
+        JsonNode objectJson = null;
+        if (json instanceof ObjectNode && json.has("object")) {
+          objectJson = ((ObjectNode) json).remove("object");
+        }
+
+        WatchEvent watchEvent = Serialization.jsonMapper().treeToValue(json, WatchEvent.class);
+        KubernetesResource object = Serialization.jsonMapper().treeToValue(objectJson, baseOperation.getType());
+
+        watchEvent.setObject(object);
+        return watchEvent;
+      } catch (JsonProcessingException ex2) {
+        throw new IllegalArgumentException("Failed to deserialize WatchEvent", ex2);
+      }
+    }
+  }
 
   protected WatchEvent readWatchEvent(String messageSource) {
-    WatchEvent event = Serialization.unmarshal(messageSource, WatchEvent.class);
+    WatchEvent event = contextAwareWatchEventDeserializer(messageSource);
     KubernetesResource object = null;
     if (event != null) {
       object = event.getObject();
@@ -261,12 +283,10 @@ protected void onMessage(String message) {
     try {
       WatchEvent event = readWatchEvent(message);
       Object object = event.getObject();
-      if (object instanceof HasMetadata) {
-        @SuppressWarnings("unchecked")
-        T obj = (T) object;
-        updateResourceVersion(obj.getMetadata().getResourceVersion());
-        Action action = Action.valueOf(event.getType());
-        eventReceived(action, obj);
+      if (object instanceof Status) {
+        Status status = (Status) object;
+
+        onStatus(status);
       } else if (object instanceof KubernetesResourceList) {
         // Dirty cast - should always be valid though
         KubernetesResourceList list = (KubernetesResourceList) object;
@@ -278,10 +298,12 @@ protected void onMessage(String message) {
             eventReceived(action, item);
           }
         }
-      } else if (object instanceof Status) {
-        Status status = (Status) object;
-      
-        onStatus(status);
+      } else if (object instanceof HasMetadata) {
+        @SuppressWarnings("unchecked")
+        T obj = (T) object;
+        updateResourceVersion(obj.getMetadata().getResourceVersion());
+        Action action = Action.valueOf(event.getType());
+        eventReceived(action, obj);
       } else {
         logger.error("Unknown message received: {}", message);
       }

kubernetes-tests/src/test/java/io/fabric8/kubernetes/client/mock/DefaultSharedIndexInformerTest.java

@@ -54,6 +54,7 @@
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
 import io.fabric8.kubernetes.client.utils.Utils;
+import io.fabric8.kubernetes.internal.KubernetesDeserializer;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
@@ -927,6 +928,23 @@ void testGenericKubernetesResourceSharedIndexInformerWithNamespaceConfigured() t
     assertEquals(0, foundExistingAnimal.getCount());
   }
 
+  @Test
+  void testGenericKubernetesResourceSharedIndexInformerWithAdditionalDeserializers() throws InterruptedException {
+    // Given
+    setupMockServerExpectations(Animal.class, "ns1", this::getList, r -> new WatchEvent(getAnimal("red-panda", "Carnivora", r), "ADDED"), null, null);
+
+    // When
+    KubernetesDeserializer.registerCustomKind("jungle.example.com/v1","Animal", CronTab.class);
+    SharedIndexInformer<GenericKubernetesResource> animalSharedIndexInformer = factory.inNamespace("ns1").sharedIndexInformerForCustomResource(animalContext, 60 * WATCH_EVENT_EMIT_TIME);
+    CountDownLatch foundExistingAnimal = new CountDownLatch(1);
+    animalSharedIndexInformer.addEventHandler(new TestResourceHandler<>(foundExistingAnimal, "red-panda"));
+    factory.startAllRegisteredInformers();
+    foundExistingAnimal.await(LATCH_AWAIT_PERIOD_IN_SECONDS, TimeUnit.SECONDS);
+
+    // Then
+    assertEquals(0, foundExistingAnimal.getCount());
+  }
+
   private KubernetesResource getAnimal(String name, String order, String resourceVersion) {
     AnimalSpec animalSpec = new AnimalSpec();
     animalSpec.setOrder(order);