public Plan generate() {
        final Plan plan = createPlan();
        registerGenericTypeInfoIfConfigured(plan);
        registerCachedFiles(plan);

        logTypeRegistrationDetails();
        return plan;
    }
@Test
    void testGenerate() {

        final String fileA = "fileA";
        final String fileB = "fileB";

        final Map<String, DistributedCache.DistributedCacheEntry> originalArtifacts =
                Stream.of(
                                Tuple2.of(
                                        fileA,
                                        new DistributedCache.DistributedCacheEntry("test1", true)),
                                Tuple2.of(
                                        fileB,
                                        new DistributedCache.DistributedCacheEntry("test2", false)))
                        .collect(Collectors.toMap(x -> x.f0, x -> x.f1));

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(10);
        env.registerCachedFile("test1", fileA, true);
        env.registerCachedFile("test2", fileB, false);

        env.fromElements(1, 3, 5)
                .map((MapFunction<Integer, String>) value -> String.valueOf(value + 1))
                .writeAsText("/tmp/csv");

        final Plan generatedPlanUnderTest = env.createProgramPlan("test");

        final Map<String, DistributedCache.DistributedCacheEntry> retrievedArtifacts =
                generatedPlanUnderTest.getCachedFiles().stream()
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        assertThat(generatedPlanUnderTest.getDataSinks()).hasSize(1);
        assertThat(generatedPlanUnderTest.getDefaultParallelism()).isEqualTo(10);
        assertThat(generatedPlanUnderTest.getExecutionConfig()).isEqualTo(env.getConfig());
        assertThat(generatedPlanUnderTest.getJobName()).isEqualTo("test");

        assertThat(retrievedArtifacts)
                .hasSameSizeAs(originalArtifacts)
                .containsEntry(fileA, originalArtifacts.get(fileA))
                .containsEntry(fileB, originalArtifacts.get(fileB));
    }