@VisibleForTesting
    StreamConfig getConfig(
            OperatorID operatorID,
            StateBackend stateBackend,
            Configuration additionalConfig,
            StreamOperator<TaggedOperatorSubtaskState> operator) {
        // Eagerly perform a deep copy of the configuration, otherwise it will result in undefined
        // behavior when deploying with multiple bootstrap transformations.
        Configuration deepCopy =
                new Configuration(
                        MutableConfig.of(stream.getExecutionEnvironment().getConfiguration()));
        deepCopy.addAll(additionalConfig);

        final StreamConfig config = new StreamConfig(deepCopy);
        config.setChainStart();
        config.setCheckpointingEnabled(true);
        config.setCheckpointMode(CheckpointingMode.EXACTLY_ONCE);

        if (keyType != null) {
            TypeSerializer<?> keySerializer =
                    keyType.createSerializer(
                            stream.getExecutionEnvironment().getConfig().getSerializerConfig());

            config.setStateKeySerializer(keySerializer);
            config.setStatePartitioner(0, keySelector);
        }

        config.setStreamOperator(operator);
        config.setOperatorName(operatorID.toHexString());
        config.setOperatorID(operatorID);
        config.setStateBackend(stateBackend);
        config.setManagedMemoryFractionOperatorOfUseCase(ManagedMemoryUseCase.STATE_BACKEND, 1.0);
        config.serializeAllConfigs();
        return config;
    }
@Test
    public void testStreamConfig() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> input = env.fromData("");

        StateBootstrapTransformation<String> transformation =
                OperatorTransformation.bootstrapWith(input)
                        .keyBy(new CustomKeySelector())
                        .transform(new ExampleKeyedStateBootstrapFunction());

        StreamConfig config =
                transformation.getConfig(
                        OperatorIDGenerator.fromUid("uid"),
                        new HashMapStateBackend(),
                        new Configuration(),
                        null);
        KeySelector selector =
                config.getStatePartitioner(0, Thread.currentThread().getContextClassLoader());

        Assert.assertEquals(
                "Incorrect key selector forwarded to stream operator",
                CustomKeySelector.class,
                selector.getClass());
    }