@Override
    public List<HasMetadata> buildAccompanyingKubernetesResources() throws IOException {
        return getTaskManagerPodTemplateFile()
                .map(
                        FunctionUtils.uncheckedFunction(
                                file -> {
                                    final Map<String, String> data = new HashMap<>();
                                    data.put(
                                            TASK_MANAGER_POD_TEMPLATE_FILE_NAME,
                                            Files.toString(file, StandardCharsets.UTF_8));
                                    final HasMetadata flinkConfConfigMap =
                                            new ConfigMapBuilder()
                                                    .withApiVersion(Constants.API_VERSION)
                                                    .withNewMetadata()
                                                    .withName(podTemplateConfigMapName)
                                                    .withLabels(
                                                            kubernetesComponentConf
                                                                    .getCommonLabels())
                                                    .endMetadata()
                                                    .addToData(data)
                                                    .build();
                                    return Collections.singletonList(flinkConfConfigMap);
                                }))
                .orElse(Collections.emptyList());
    }
@Test
    void testDecoratorShouldFailWhenPodTemplateFileNotExist() {
        final String msg =
                String.format(
                        "Pod template file %s does not exist.",
                        new File(flinkConfDir, POD_TEMPLATE_FILE_NAME));
        assertThatThrownBy(
                        () -> podTemplateMountDecorator.buildAccompanyingKubernetesResources(),
                        "Decorator should fail when the pod template file does not exist.")
                .satisfies(FlinkAssertions.anyCauseMatches(msg));
    }