@Override
    public List<HasMetadata> buildAccompanyingKubernetesResources() throws IOException {

        final List<HasMetadata> resources = new ArrayList<>();

        if (!StringUtils.isNullOrWhitespaceOnly(securityConfig.getKeytab())
                && !StringUtils.isNullOrWhitespaceOnly(securityConfig.getPrincipal())) {
            final File keytab = new File(securityConfig.getKeytab());
            if (!keytab.exists()) {
                LOG.warn(
                        "Could not found the kerberos keytab file in {}.",
                        keytab.getAbsolutePath());
            } else {
                resources.add(
                        new SecretBuilder()
                                .withNewMetadata()
                                .withName(
                                        getKerberosKeytabSecretName(
                                                kubernetesParameters.getClusterId()))
                                .endMetadata()
                                .addToData(
                                        keytab.getName(),
                                        Base64.getEncoder()
                                                .encodeToString(Files.toByteArray(keytab)))
                                .build());

                // Set keytab path in the container. One should make sure this decorator is
                // triggered before FlinkConfMountDecorator.
                kubernetesParameters
                        .getFlinkConfiguration()
                        .set(
                                SecurityOptions.KERBEROS_LOGIN_KEYTAB,
                                String.format(
                                        "%s/%s",
                                        Constants.KERBEROS_KEYTAB_MOUNT_POINT, keytab.getName()));
            }
        }

        if (!StringUtils.isNullOrWhitespaceOnly(
                kubernetesParameters
                        .getFlinkConfiguration()
                        .get(SecurityOptions.KERBEROS_KRB5_PATH))) {
            final File krb5Conf =
                    new File(
                            kubernetesParameters
                                    .getFlinkConfiguration()
                                    .get(SecurityOptions.KERBEROS_KRB5_PATH));
            if (!krb5Conf.exists()) {
                LOG.warn(
                        "Could not found the kerberos config file in {}.",
                        krb5Conf.getAbsolutePath());
            } else {
                resources.add(
                        new ConfigMapBuilder()
                                .withNewMetadata()
                                .withName(
                                        getKerberosKrb5confConfigMapName(
                                                kubernetesParameters.getClusterId()))
                                .endMetadata()
                                .addToData(
                                        krb5Conf.getName(),
                                        Files.toString(krb5Conf, StandardCharsets.UTF_8))
                                .build());
            }
        }
        return resources;
    }
@Test
    void testConfEditWhenBuildAccompanyingKubernetesResources() throws IOException {
        kerberosMountDecorator.buildAccompanyingKubernetesResources();

        assertThat(
                        this.testingKubernetesParameters
                                .getFlinkConfiguration()
                                .get(SecurityOptions.KERBEROS_LOGIN_KEYTAB))
                .isEqualTo(
                        String.format("%s/%s", Constants.KERBEROS_KEYTAB_MOUNT_POINT, KEYTAB_FILE));
    }