@Test public void testStreamConfig() { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSource<String> input = env.fromElements(""); BootstrapTransformation<String> transformation = OperatorTransformation .bootstrapWith(input) .keyBy(new CustomKeySelector()) .transform(new ExampleKeyedStateBootstrapFunction()); StreamConfig config = transformation.getConfig(OperatorIDGenerator.fromUid("uid"), new MemoryStateBackend(), null); KeySelector selector = config.getStatePartitioner(0, Thread.currentThread().getContextClassLoader()); Assert.assertEquals("Incorrect key selector forwarded to stream operator", CustomKeySelector.class, selector.getClass()); }
@VisibleForTesting StreamConfig getConfig(OperatorID operatorID, StateBackend stateBackend, StreamOperator<TaggedOperatorSubtaskState> operator) { final StreamConfig config; if (keyType == null) { config = new BoundedStreamConfig(); } else { TypeSerializer<?> keySerializer = keyType.createSerializer(dataSet.getExecutionEnvironment().getConfig()); config = new BoundedStreamConfig(keySerializer, originalKeySelector); } config.setStreamOperator(operator); config.setOperatorName(operatorID.toHexString()); config.setOperatorID(operatorID); config.setStateBackend(stateBackend); return config; }
BootstrapTransformation { @VisibleForTesting StreamConfig getConfig(OperatorID operatorID, StateBackend stateBackend, StreamOperator<TaggedOperatorSubtaskState> operator) { final StreamConfig config; if (keyType == null) { config = new BoundedStreamConfig(); } else { TypeSerializer<?> keySerializer = keyType.createSerializer(dataSet.getExecutionEnvironment().getConfig()); config = new BoundedStreamConfig(keySerializer, originalKeySelector); } config.setStreamOperator(operator); config.setOperatorName(operatorID.toHexString()); config.setOperatorID(operatorID); config.setStateBackend(stateBackend); return config; } }
BootstrapTransformation { @VisibleForTesting StreamConfig getConfig(OperatorID operatorID, StateBackend stateBackend, StreamOperator<TaggedOperatorSubtaskState> operator) { final StreamConfig config; if (keyType == null) { config = new BoundedStreamConfig(); } else { TypeSerializer<?> keySerializer = keyType.createSerializer(dataSet.getExecutionEnvironment().getConfig()); config = new BoundedStreamConfig(keySerializer, 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); }
BootstrapTransformation { @VisibleForTesting StreamConfig getConfig(OperatorID operatorID, StateBackend stateBackend, StreamOperator<TaggedOperatorSubtaskState> operator) { final StreamConfig config; if (keyType == null) { config = new BoundedStreamConfig(); } else { TypeSerializer<?> keySerializer = keyType.createSerializer(dataSet.getExecutionEnvironment().getConfig()); config = new BoundedStreamConfig(keySerializer, 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);  }
BootstrapTransformation { @VisibleForTesting StreamConfig getConfig(OperatorID operatorID, StateBackend stateBackend, StreamOperator<TaggedOperatorSubtaskState> operator) { final StreamConfig config; if (keyType == null) { config = new BoundedStreamConfig(); } else { TypeSerializer<?> keySerializer = keyType.createSerializer(dataSet.getExecutionEnvironment().getConfig()); config = new BoundedStreamConfig(keySerializer, 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);   }