@Override
    public List<KubernetesPod> getPodsWithLabels(Map<String, String> labels) {
        final List<Pod> podList =
                this.internalClient
                        .pods()
                        .withLabels(labels)
                        .list(
                                new ListOptionsBuilder()
                                        .withResourceVersion(KUBERNETES_ZERO_RESOURCE_VERSION)
                                        .build())
                        .getItems();

        if (podList == null || podList.isEmpty()) {
            return new ArrayList<>();
        }

        return podList.stream().map(KubernetesPod::new).collect(Collectors.toList());
    }
@Test
    void testGetPodsWithLabels() {
        final String podName = "pod-with-labels";
        final Pod pod =
                new PodBuilder()
                        .editOrNewMetadata()
                        .withName(podName)
                        .withLabels(TESTING_LABELS)
                        .endMetadata()
                        .editOrNewSpec()
                        .endSpec()
                        .build();
        this.kubeClient.pods().inNamespace(NAMESPACE).create(pod);
        List<KubernetesPod> kubernetesPods = this.flinkKubeClient.getPodsWithLabels(TESTING_LABELS);
        assertThat(kubernetesPods)
                .satisfiesExactly(
                        kubernetesPod -> assertThat(kubernetesPod.getName()).isEqualTo(podName));
    }