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
/**
	 * Test that ensures the KafkaConsumer is properly failing if the topic doesnt exist
	 * and a wrong broker was specified.
	 *
	 * @throws Exception
	 */
Test that ensures the KafkaConsumer is properly failing if the topic doesnt exist
and a wrong broker was specified.

@throws Exception
Test that ensures the KafkaConsumer is properly failing if the topic doesnt exist and a wrong broker was specified . @throws Exception
public void runFailOnNoBrokerTest() throws Exception {
		try {
			Properties properties = new Properties();
			StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
			see.getConfig().disableSysoutLogging();
			see.setRestartStrategy(RestartStrategies.noRestart());
			see.setParallelism(1);
			properties.setProperty("bootstrap.servers", "localhost:80");
			properties.setProperty("zookeeper.connect", "localhost:80");
			properties.setProperty("group.id", "test");
			properties.setProperty("request.timeout.ms", "3000"); 
			properties.setProperty("socket.timeout.ms", "3000");
			properties.setProperty("session.timeout.ms", "2000");
			properties.setProperty("fetch.max.wait.ms", "2000");
			properties.setProperty("heartbeat.interval.ms", "1000");
			properties.putAll(secureProps);
			FlinkKafkaConsumerBase<String> source = kafkaServer.getConsumer("doesntexist", new SimpleStringSchema(), properties);
			DataStream<String> stream = see.addSource(source);
			stream.print();
			see.execute("No broker test");
		} catch (JobExecutionException jee) {
			if (kafkaServer.getVersion().equals("0.9") || kafkaServer.getVersion().equals("0.10") || kafkaServer.getVersion().equals("0.11")) {
				assertTrue(jee.getCause() instanceof TimeoutException);
				TimeoutException te = (TimeoutException) jee.getCause();
				assertEquals("Timeout expired while fetching topic metadata", te.getMessage());
			} else {
				assertTrue(jee.getCause() instanceof RuntimeException);
				RuntimeException re = (RuntimeException) jee.getCause();
				assertTrue(re.getMessage().contains("Unable to retrieve any partitions"));
			}
		}
	}
public void runFailOnNoBrokerTest ( ) throws Exception { try { Properties properties = new Properties ( ) ; StreamExecutionEnvironment see = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; see . getConfig ( ) . disableSysoutLogging ( ) ; see . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; see . setParallelism ( 1 ) ; properties . setProperty ( "bootstrap.servers" , "localhost:80" ) ; properties . setProperty ( "zookeeper.connect" , "localhost:80" ) ; properties . setProperty ( "group.id" , "test" ) ; properties . setProperty ( "request.timeout.ms" , "3000" ) ; properties . setProperty ( "socket.timeout.ms" , "3000" ) ; properties . setProperty ( "session.timeout.ms" , "2000" ) ; properties . setProperty ( "fetch.max.wait.ms" , "2000" ) ; properties . setProperty ( "heartbeat.interval.ms" , "1000" ) ; properties . putAll ( secureProps ) ; FlinkKafkaConsumerBase < String > source = kafkaServer . getConsumer ( "doesntexist" , new SimpleStringSchema ( ) , properties ) ; DataStream < String > stream = see . addSource ( source ) ; stream . print ( ) ; see . execute ( "No broker test" ) ; } catch ( JobExecutionException jee ) { if ( kafkaServer . getVersion ( ) . equals ( "0.9" ) || kafkaServer . getVersion ( ) . equals ( "0.10" ) || kafkaServer . getVersion ( ) . equals ( "0.11" ) ) { assertTrue ( jee . getCause ( ) instanceof TimeoutException ) ; TimeoutException te = ( TimeoutException ) jee . getCause ( ) ; assertEquals ( "Timeout expired while fetching topic metadata" , te . getMessage ( ) ) ; } else { assertTrue ( jee . getCause ( ) instanceof RuntimeException ) ; RuntimeException re = ( RuntimeException ) jee . getCause ( ) ; assertTrue ( re . getMessage ( ) . contains ( "Unable to retrieve any partitions" ) ) ; } } }
Test that ensures the KafkaConsumer is properly failing if the topic doesnt exist
 and a wrong broker was specified.
Test that ensures the KafkaConsumer is properly failing if the topic doesnt exist and a wrong broker was specified .
// use wrong ports for the consumers // let the test fail fast