@Override
    public FlinkPod decorateFlinkPod(FlinkPod flinkPod) {
        PodBuilder podBuilder = new PodBuilder(flinkPod.getPodWithoutMainContainer());
        ContainerBuilder containerBuilder = new ContainerBuilder(flinkPod.getMainContainer());

        if (!StringUtils.isNullOrWhitespaceOnly(securityConfig.getKeytab())
                && !StringUtils.isNullOrWhitespaceOnly(securityConfig.getPrincipal())) {
            podBuilder =
                    podBuilder
                            .editOrNewSpec()
                            .addNewVolume()
                            .withName(Constants.KERBEROS_KEYTAB_VOLUME)
                            .withNewSecret()
                            .withSecretName(
                                    getKerberosKeytabSecretName(
                                            kubernetesParameters.getClusterId()))
                            .endSecret()
                            .endVolume()
                            .endSpec();

            containerBuilder =
                    containerBuilder
                            .addNewVolumeMount()
                            .withName(Constants.KERBEROS_KEYTAB_VOLUME)
                            .withMountPath(Constants.KERBEROS_KEYTAB_MOUNT_POINT)
                            .endVolumeMount();
        }

        if (!StringUtils.isNullOrWhitespaceOnly(
                kubernetesParameters
                        .getFlinkConfiguration()
                        .get(SecurityOptions.KERBEROS_KRB5_PATH))) {
            final File krb5Conf =
                    new File(
                            kubernetesParameters
                                    .getFlinkConfiguration()
                                    .get(SecurityOptions.KERBEROS_KRB5_PATH));
            podBuilder =
                    podBuilder
                            .editOrNewSpec()
                            .addNewVolume()
                            .withName(Constants.KERBEROS_KRB5CONF_VOLUME)
                            .withNewConfigMap()
                            .withName(
                                    getKerberosKrb5confConfigMapName(
                                            kubernetesParameters.getClusterId()))
                            .withItems(
                                    new KeyToPathBuilder()
                                            .withKey(krb5Conf.getName())
                                            .withPath(KERBEROS_KRB5CONF_FILE)
                                            .build())
                            .endConfigMap()
                            .endVolume()
                            .endSpec();

            containerBuilder =
                    containerBuilder
                            .addNewVolumeMount()
                            .withName(Constants.KERBEROS_KRB5CONF_VOLUME)
                            .withMountPath(
                                    Constants.KERBEROS_KRB5CONF_MOUNT_DIR
                                            + "/"
                                            + KERBEROS_KRB5CONF_FILE)
                            .withSubPath(KERBEROS_KRB5CONF_FILE)
                            .endVolumeMount();
        }

        return new FlinkPod(podBuilder.build(), containerBuilder.build());
    }
@Test
    void testWhetherPodOrContainerIsDecorated() {
        final FlinkPod resultFlinkPod = kerberosMountDecorator.decorateFlinkPod(baseFlinkPod);
        assertThat(resultFlinkPod.getPodWithoutMainContainer())
                .isNotEqualTo(baseFlinkPod.getPodWithoutMainContainer());
        assertThat(resultFlinkPod.getMainContainer()).isNotEqualTo(baseFlinkPod.getMainContainer());
    }