Custom resource informer's event handler's onUpdate stops working after pod is restarted

[asardaes]

Hello, I am currently trying to write a Kubernetes controller using io.fabric8:kubernetes-client:5.5.0. As of now, I have only tested basic logic with an informer:

flinkJobInformer = sharedInformerFactory.sharedIndexInformerForCustomResource(FlinkJobCustomResource.class, resyncPeriodSeconds * 1000L);
flinkJobInformer.addEventHandler(new FlinkJobHandler());
sharedInformerFactory.startAllRegisteredInformers().get();

The handler does nothing except log what it received.

After initial deployment, everything works as expected. But if I manually restart the pod where my controller is running after I already created some custom resources, I start seeing these exceptions after every resync:

2021-07-02T11:46:50,645+00 ERROR io.fabric8.kubernetes.client.informers.cache.ProcessorListener [pool-2-thread-1] Failed invoking com.dummy.kubernetes.controller.v1.FlinkJobHandler@60861d36 event handler:
io.fabric8.kubernetes.api.model.GenericKubernetesResource cannot be cast to com.dummy.kubernetes.model.v1.FlinkJobCustomResource
java.lang.ClassCastException: io.fabric8.kubernetes.api.model.GenericKubernetesResource cannot be cast to com.dummy.kubernetes.model.v1.FlinkJobCustomResource
        at com.dummy.kubernetes.controller.v1.FlinkJobHandler.onUpdate(FlinkJobHandler.java:9)
        at io.fabric8.kubernetes.client.informers.cache.ProcessorListener$UpdateNotification.handle(ProcessorListener.java:92)
        at io.fabric8.kubernetes.client.informers.cache.ProcessorListener.add(ProcessorListener.java:51)
        at io.fabric8.kubernetes.client.informers.cache.SharedProcessor.lambda$distribute$0(SharedProcessor.java:90)
        at io.fabric8.kubernetes.client.utils.SerialExecutor.lambda$execute$0(SerialExecutor.java:40)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

Does anyone know why this happens?

[shawkins]

Does anyone know why this happens?

No, that seems odd.

What needs to happen for this to occur is that the deserializer for FlinkJobCustomResource is not getting registered. That is supposed to happen automatically prior to any list/watch operation by calls in SharedInformerFactory.

If it is not registered, that will cause the parsing of the list operation to return GenericKubernetesResource instances instead. Those values then make it into the informer cache, and then cause the exception on resync - you should see the same problem if you try to use the cache/store.

This cannot happen with watch events as there's some code that will convert the WatchEvent objects to the expected type. Ideally we can figure out why the deserializer is not getting registered - if not though we can add a similar type conversion logic at other places.

@rohanKanojia any ideas?

[rohanKanojia]

strange, I'm not sure right now what might be causing this. I'll try to reproduce this. So you're saying I can reproduce it after restarting the controller pod, right?

[asardaes]

Yes, I can consistently reproduce this if I install my controller, create a custom resource, and then restart the controller's pod.

[asardaes]

And for completeness, here's an overview of my CR class

@Group(FlinkJobConstants.GROUP)
@Version("v1")
@Kind(FlinkJobConstants.KIND)
@Plural(FlinkJobConstants.PLURAL)
@ShortNames({FlinkJobConstants.SHORT_NAME})
public class FlinkJobCustomResource extends CustomResource<FlinkJobSpec, FlinkJobStatus> implements Namespaced {
    private static final long serialVersionUID = 1L;

    @Override
    protected FlinkJobSpec initSpec() {
        return new FlinkJobSpec();
    }

    @Override
    protected FlinkJobStatus initStatus() {
        return new FlinkJobStatus();
    }
}

[asardaes]

I was looking for ways to make the informer namespace-specific (in the future) and I changed the code to the following:

MixedOperation<FlinkJobCustomResource, KubernetesResourceList<FlinkJobCustomResource>, Resource<FlinkJobCustomResource>> flinkJobOperation =
        k8sClient.customResources(FlinkJobCustomResource.class);

FlinkJobHandler handler = new FlinkJobHandler();
SharedIndexInformer<FlinkJobCustomResource> flinkJobInformer = flinkJobOperation
        .inAnyNamespace()
        .inform(handler, resyncPeriodSeconds * 1000L);

I don't see the issue after restarting the pod in this case.

[shawkins]

What is the FlinkJobConstants.KIND constant value?

I don't see the issue after restarting the pod in this case.

The problem may be with the deserializer registration logic:

kubernetes-client/kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/SharedInformerFactory.java

Line 430 in 1f5061f

KubernetesDeserializer.registerCustomKind(HasMetadata.getApiVersion(apiTypeClass), apiTypeClass.getSimpleName(), apiTypeClass);

vs

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

Line 53 in 74cc63d

KubernetesDeserializer.registerCustomKind(apiVersion, kind(context.getCrdContext()), type);

Note that factory infers the kind from the class name, but the dsl logic will take it from the context which was constructed by consulting the annotations.

This issue should occur for you anytime you start your controller and your already have instances of the custom resource that exist. If you create them while the informer is running, the aforementioned conversion in the watch logic will properly type them before they get into the cache - until the first http gone exception in which case you should start seeing the same exception.

[asardaes]

Kind is FlinkJob, and yes, I think you're right about the issue. Thanks for looking into this.

[shawkins]

Great it will be a simple fix. For now you can either change the class name to match the kind, or just use the inform approach - it is available at any context where a list is available, the namespace is not required.

[rohanKanojia]

After initial deployment, everything works as expected. But if I manually restart the pod where my controller is running after I already created some custom resources, I start seeing these exceptions after every resync:

Hello, I'm trying to reproduce your issue with this reproducer https://github.com/r0haaaan/flinkjobinformer-controller. I can reproduce your issue but I'm not even able to get factory's informer working(even locally) with v5.5.0 with your setup. I can verify that the fix suggested by Steven works.