@Override
    public void eventReceived(Action action, Pod pod) {
        logger.debug(
                "Received {} event for pod {}, details: {}{}",
                action,
                pod.getMetadata().getName(),
                System.lineSeparator(),
                KubernetesUtils.tryToGetPrettyPrintYaml(pod.getStatus()));
        final List<KubernetesPod> pods = Collections.singletonList(new KubernetesPod(pod));
        switch (action) {
            case ADDED:
                callbackHandler.onAdded(pods);
                break;
            case MODIFIED:
                callbackHandler.onModified(pods);
                break;
            case ERROR:
                callbackHandler.onError(pods);
                break;
            case DELETED:
                callbackHandler.onDeleted(pods);
                break;
            default:
                logger.debug(
                        "Ignore handling {} event for pod {}", action, pod.getMetadata().getName());
                break;
        }
    }
@Test
    void testCallbackHandler() {
        FlinkPod pod = new FlinkPod.Builder().build();
        final KubernetesPodsWatcher podsWatcher =
                new KubernetesPodsWatcher(
                        TestingWatchCallbackHandler.<KubernetesPod>builder()
                                .setOnAddedConsumer(pods -> podAddedList.addAll(pods))
                                .setOnModifiedConsumer(pods -> podModifiedList.addAll(pods))
                                .setOnDeletedConsumer(pods -> podDeletedList.addAll(pods))
                                .setOnErrorConsumer(pods -> podErrorList.addAll(pods))
                                .build());
        podsWatcher.eventReceived(Watcher.Action.ADDED, pod.getPodWithoutMainContainer());
        podsWatcher.eventReceived(Watcher.Action.MODIFIED, pod.getPodWithoutMainContainer());
        podsWatcher.eventReceived(Watcher.Action.DELETED, pod.getPodWithoutMainContainer());
        podsWatcher.eventReceived(Watcher.Action.ERROR, pod.getPodWithoutMainContainer());

        assertThat(podAddedList).hasSize(1);
        assertThat(podModifiedList).hasSize(1);
        assertThat(podDeletedList).hasSize(1);
        assertThat(podErrorList).hasSize(1);
    }