public StreamGraph generate() {
        streamGraph =
                new StreamGraph(
                        configuration, executionConfig, checkpointConfig, savepointRestoreSettings);
        shouldExecuteInBatchMode = shouldExecuteInBatchMode();
        configureStreamGraph(streamGraph);

        alreadyTransformed = new IdentityHashMap<>();

        for (Transformation<?> transformation : transformations) {
            transform(transformation);
        }
        streamGraph.setSlotSharingGroupResource(slotSharingGroupResources);

        setFineGrainedGlobalStreamExchangeMode(streamGraph);

        LineageGraph lineageGraph = LineageGraphUtils.convertToLineageGraph(transformations);
        streamGraph.setLineageGraph(lineageGraph);

        for (StreamNode node : streamGraph.getStreamNodes()) {
            if (node.getInEdges().stream().anyMatch(this::shouldDisableUnalignedCheckpointing)) {
                for (StreamEdge edge : node.getInEdges()) {
                    edge.setSupportsUnalignedCheckpoints(false);
                }
            }
        }

        final StreamGraph builtStreamGraph = streamGraph;

        alreadyTransformed.clear();
        alreadyTransformed = null;
        streamGraph = null;

        return builtStreamGraph;
    }
@Test
    void testSettingSavepointRestoreSettings() {
        Configuration config = new Configuration();
        config.set(StateRecoveryOptions.SAVEPOINT_PATH, "/tmp/savepoint");

        final StreamGraph streamGraph =
                new StreamGraphGenerator(
                                Collections.emptyList(),
                                new ExecutionConfig(),
                                new CheckpointConfig(),
                                config)
                        .generate();

        SavepointRestoreSettings savepointRestoreSettings =
                streamGraph.getSavepointRestoreSettings();
        assertThat(savepointRestoreSettings)
                .isEqualTo(SavepointRestoreSettings.forPath("/tmp/savepoint"));
    }