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 that DeserializationSchema.isEndOfStream() is properly evaluated.
	 *
	 * @throws Exception
	 */
Test that ensures that DeserializationSchema.isEndOfStream() is properly evaluated.
@throws Exception
Test that ensures that DeserializationSchema . isEndOfStream () is properly evaluated . @throws Exception
public void runEndOfStreamTest() throws Exception {
		final int elementCount = 300;
		final String topic = writeSequence("testEndOfStream", elementCount, 1, 1);
		final StreamExecutionEnvironment env1 = StreamExecutionEnvironment.getExecutionEnvironment();
		env1.setParallelism(1);
		env1.getConfig().setRestartStrategy(RestartStrategies.noRestart());
		env1.getConfig().disableSysoutLogging();
		Properties props = new Properties();
		props.putAll(standardProps);
		props.putAll(secureProps);
		DataStream<Tuple2<Integer, Integer>> fromKafka = env1.addSource(kafkaServer.getConsumer(topic, new FixedNumberDeserializationSchema(elementCount), props));
		fromKafka.flatMap(new FlatMapFunction<Tuple2<Integer, Integer>, Void>() {
			@Override
			public void flatMap(Tuple2<Integer, Integer> value, Collector<Void> out) throws Exception {
			}
		});
		tryExecute(env1, "Consume " + elementCount + " elements from Kafka");
		deleteTestTopic(topic);
	}
public void runEndOfStreamTest ( ) throws Exception { final int elementCount = 300 ; final String topic = writeSequence ( "testEndOfStream" , elementCount , 1 , 1 ) ; final StreamExecutionEnvironment env1 = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env1 . setParallelism ( 1 ) ; env1 . getConfig ( ) . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; env1 . getConfig ( ) . disableSysoutLogging ( ) ; Properties props = new Properties ( ) ; props . putAll ( standardProps ) ; props . putAll ( secureProps ) ; DataStream < Tuple2 < Integer , Integer > > fromKafka = env1 . addSource ( kafkaServer . getConsumer ( topic , new FixedNumberDeserializationSchema ( elementCount ) , props ) ) ; fromKafka . flatMap ( new FlatMapFunction < Tuple2 < Integer , Integer > , Void > ( ) { @ Override public void flatMap ( Tuple2 < Integer , Integer > value , Collector < Void > out ) throws Exception { } } ) ; tryExecute ( env1 , "Consume " + elementCount + " elements from Kafka" ) ; deleteTestTopic ( topic ) ; }
Test that ensures that DeserializationSchema.isEndOfStream() is properly evaluated.
Test that ensures that DeserializationSchema . isEndOfStream () is properly evaluated .
// read using custom schema // noop ;)