flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
/**
	 * Tests that the source can be properly canceled when reading full partitions.
	 */
Tests that the source can be properly canceled when reading full partitions.
Tests that the source can be properly canceled when reading full partitions .
public void runFailOnDeployTest() throws Exception {
		final String topic = "failOnDeployTopic";
		createTestTopic(topic, 2, 1);
		DeserializationSchema<Integer> schema =
				new TypeInformationSerializationSchema<>(BasicTypeInfo.INT_TYPE_INFO, new ExecutionConfig());
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(12); 
		env.getConfig().disableSysoutLogging();
		Properties props = new Properties();
		props.putAll(standardProps);
		props.putAll(secureProps);
		FlinkKafkaConsumerBase<Integer> kafkaSource = kafkaServer.getConsumer(topic, schema, props);
		env
				.addSource(kafkaSource)
				.addSink(new DiscardingSink<Integer>());
		try {
			env.execute("test fail on deploy");
			fail("this test should fail with an exception");
		}
		catch (JobExecutionException e) {
			Throwable cause = e.getCause();
			int depth = 0;
			boolean foundResourceException = false;
			while (cause != null && depth++ < 20) {
				if (cause instanceof NoResourceAvailableException) {
					foundResourceException = true;
					break;
				}
				cause = cause.getCause();
			}
			assertTrue("Wrong exception", foundResourceException);
		}
		deleteTestTopic(topic);
	}
public void runFailOnDeployTest ( ) throws Exception { final String topic = "failOnDeployTopic" ; createTestTopic ( topic , 2 , 1 ) ; DeserializationSchema < Integer > schema = new TypeInformationSerializationSchema < > ( BasicTypeInfo . INT_TYPE_INFO , new ExecutionConfig ( ) ) ; StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setParallelism ( 12 ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; Properties props = new Properties ( ) ; props . putAll ( standardProps ) ; props . putAll ( secureProps ) ; FlinkKafkaConsumerBase < Integer > kafkaSource = kafkaServer . getConsumer ( topic , schema , props ) ; env . addSource ( kafkaSource ) . addSink ( new DiscardingSink < Integer > ( ) ) ; try { env . execute ( "test fail on deploy" ) ; fail ( "this test should fail with an exception" ) ; } catch ( JobExecutionException e ) { Throwable cause = e . getCause ( ) ; int depth = 0 ; boolean foundResourceException = false ; while ( cause != null && depth ++ < 20 ) { if ( cause instanceof NoResourceAvailableException ) { foundResourceException = true ; break ; } cause = cause . getCause ( ) ; } assertTrue ( "Wrong exception" , foundResourceException ) ; } deleteTestTopic ( topic ) ; }
Tests that the source can be properly canceled when reading full partitions.
Tests that the source can be properly canceled when reading full partitions .
// needs to be more that the mini cluster has slots // validate that we failed due to a NoResourceAvailableException