class BootstrapTransformation {

	@VisibleForTesting
	StreamConfig getConfig(OperatorID operatorID, StateBackend stateBackend, 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(dataSet.getExecutionEnvironment().getConfiguration());
		final StreamConfig config = new StreamConfig(deepCopy);
		config.setChainStart();
		config.setCheckpointingEnabled(true);
		config.setCheckpointMode(CheckpointingMode.EXACTLY_ONCE);

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

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

		config.setStreamOperator(operator);
		config.setOperatorName(operatorID.toHexString());
		config.setOperatorID(operatorID);
		config.setStateBackend(stateBackend);
		return config;
	}

	  BootstrapTransformation(
		DataSet<T> dataSet,
		OptionalInt operatorMaxParallelism,
		SavepointWriterOperatorFactory factory);
	 <K> BootstrapTransformation(
		DataSet<T> dataSet,
		OptionalInt operatorMaxParallelism,
		SavepointWriterOperatorFactory factory,
		@Nonnull KeySelector<T, K> keySelector,
		@Nonnull TypeInformation<K> keyType);

}