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 empty partitions.
	 */
Tests that the source can be properly canceled when reading empty partitions.
Tests that the source can be properly canceled when reading empty partitions .
public void runCancelingOnEmptyInputTest() throws Exception {
		final String topic = "cancelingOnEmptyInputTopic";
		final int parallelism = 3;
		createTestTopic(topic, parallelism, 1);
		final AtomicReference<Throwable> error = new AtomicReference<>();
		final Runnable jobRunner = new Runnable() {
			@Override
			public void run() {
				try {
					final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
					env.setParallelism(parallelism);
					env.enableCheckpointing(100);
					env.getConfig().disableSysoutLogging();
					Properties props = new Properties();
					props.putAll(standardProps);
					props.putAll(secureProps);
					FlinkKafkaConsumerBase<String> source = kafkaServer.getConsumer(topic, new SimpleStringSchema(), props);
					env.addSource(source).addSink(new DiscardingSink<String>());
					env.execute("CancelingOnEmptyInputTest");
				}
				catch (Throwable t) {
					LOG.error("Job Runner failed with exception", t);
					error.set(t);
				}
			}
		};
		Thread runnerThread = new Thread(jobRunner, "program runner thread");
		runnerThread.start();
		Thread.sleep(2000);
		Throwable failueCause = error.get();
		if (failueCause != null) {
			failueCause.printStackTrace();
			Assert.fail("Test failed prematurely with: " + failueCause.getMessage());
		}
		JobManagerCommunicationUtils.cancelCurrentJob(flink.getLeaderGateway(timeout));
		runnerThread.join();
		failueCause = error.get();
		assertNotNull("program did not fail properly due to canceling", failueCause);
		assertTrue(failueCause.getMessage().contains("Job was cancelled"));
		deleteTestTopic(topic);
	}
public void runCancelingOnEmptyInputTest ( ) throws Exception { final String topic = "cancelingOnEmptyInputTopic" ; final int parallelism = 3 ; createTestTopic ( topic , parallelism , 1 ) ; final AtomicReference < Throwable > error = new AtomicReference < > ( ) ; final Runnable jobRunner = new Runnable ( ) { @ Override public void run ( ) { try { final StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setParallelism ( parallelism ) ; env . enableCheckpointing ( 100 ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; Properties props = new Properties ( ) ; props . putAll ( standardProps ) ; props . putAll ( secureProps ) ; FlinkKafkaConsumerBase < String > source = kafkaServer . getConsumer ( topic , new SimpleStringSchema ( ) , props ) ; env . addSource ( source ) . addSink ( new DiscardingSink < String > ( ) ) ; env . execute ( "CancelingOnEmptyInputTest" ) ; } catch ( Throwable t ) { LOG . error ( "Job Runner failed with exception" , t ) ; error . set ( t ) ; } } } ; Thread runnerThread = new Thread ( jobRunner , "program runner thread" ) ; runnerThread . start ( ) ; Thread . sleep ( 2000 ) ; Throwable failueCause = error . get ( ) ; if ( failueCause != null ) { failueCause . printStackTrace ( ) ; Assert . fail ( "Test failed prematurely with: " + failueCause . getMessage ( ) ) ; } JobManagerCommunicationUtils . cancelCurrentJob ( flink . getLeaderGateway ( timeout ) ) ; runnerThread . join ( ) ; failueCause = error . get ( ) ; assertNotNull ( "program did not fail properly due to canceling" , failueCause ) ; assertTrue ( failueCause . getMessage ( ) . contains ( "Job was cancelled" ) ) ; deleteTestTopic ( topic ) ; }
Tests that the source can be properly canceled when reading empty partitions.
Tests that the source can be properly canceled when reading empty partitions .
// wait a bit before canceling // cancel // wait for the program to be done and validate that we failed with the right exception