flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-tests/src/test/java/org/apache/flink/test/operators/RemoteEnvironmentITCase.java
/**
	 * Ensure that the program parallelism can be set even if the configuration is supplied.
	 */
Ensure that the program parallelism can be set even if the configuration is supplied.
Ensure that the program parallelism can be set even if the configuration is supplied .
@Test
	public void testUserSpecificParallelism() throws Exception {
		Configuration config = new Configuration();
		config.setString(AkkaOptions.STARTUP_TIMEOUT, VALID_STARTUP_TIMEOUT);
		final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(
				cluster.getHostname(),
				cluster.getPort(),
				config
		);
		env.setParallelism(USER_DOP);
		env.getConfig().disableSysoutLogging();
		DataSet<Integer> result = env.createInput(new ParallelismDependentInputFormat())
				.rebalance()
				.mapPartition(new RichMapPartitionFunction<Integer, Integer>() {
					@Override
					public void mapPartition(Iterable<Integer> values, Collector<Integer> out) throws Exception {
						out.collect(getRuntimeContext().getIndexOfThisSubtask());
					}
				});
		List<Integer> resultCollection = result.collect();
		assertEquals(USER_DOP, resultCollection.size());
	}
@ Test public void testUserSpecificParallelism ( ) throws Exception { Configuration config = new Configuration ( ) ; config . setString ( AkkaOptions . STARTUP_TIMEOUT , VALID_STARTUP_TIMEOUT ) ; final ExecutionEnvironment env = ExecutionEnvironment . createRemoteEnvironment ( cluster . getHostname ( ) , cluster . getPort ( ) , config ) ; env . setParallelism ( USER_DOP ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; DataSet < Integer > result = env . createInput ( new ParallelismDependentInputFormat ( ) ) . rebalance ( ) . mapPartition ( new RichMapPartitionFunction < Integer , Integer > ( ) { @ Override public void mapPartition ( Iterable < Integer > values , Collector < Integer > out ) throws Exception { out . collect ( getRuntimeContext ( ) . getIndexOfThisSubtask ( ) ) ; } } ) ; List < Integer > resultCollection = result . collect ( ) ; assertEquals ( USER_DOP , resultCollection . size ( ) ) ; }
Ensure that the program parallelism can be set even if the configuration is supplied.
Ensure that the program parallelism can be set even if the configuration is supplied .