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 delete behavior and metrics for producer.
	 * @throws Exception
	 */
Test delete behavior and metrics for producer.
@throws Exception
Test delete behavior and metrics for producer . @throws Exception
public void runAllDeletesTest() throws Exception {
		final String topic = "alldeletestest";
		createTestTopic(topic, 1, 1);
		final int elementCount = 300;
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);
		env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
		env.getConfig().disableSysoutLogging();
		DataStream<Tuple2<byte[], PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<byte[], PojoValue>>() {
			@Override
			public void run(SourceContext<Tuple2<byte[], PojoValue>> ctx) throws Exception {
				Random rnd = new Random(1337);
				for (long i = 0; i < elementCount; i++) {
					final byte[] key = new byte[200];
					rnd.nextBytes(key);
					ctx.collect(new Tuple2<>(key, (PojoValue) null));
				}
			}
			@Override
			public void cancel() {
			}
		});
		TypeInformationKeyValueSerializationSchema<byte[], PojoValue> schema = new TypeInformationKeyValueSerializationSchema<>(byte[].class, PojoValue.class, env.getConfig());
		Properties producerProperties = FlinkKafkaProducerBase.getPropertiesFromBrokerList(brokerConnectionStrings);
		producerProperties.setProperty("retries", "3");
		producerProperties.putAll(secureProps);
		kafkaServer.produceIntoKafka(kvStream, topic, schema, producerProperties, null);
		env.execute("Write deletes to Kafka");
		env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);
		env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
		env.getConfig().disableSysoutLogging();
		Properties props = new Properties();
		props.putAll(standardProps);
		props.putAll(secureProps);
		DataStream<Tuple2<byte[], PojoValue>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, schema, props));
		fromKafka.flatMap(new RichFlatMapFunction<Tuple2<byte[], PojoValue>, Object>() {
			long counter = 0;
			@Override
			public void flatMap(Tuple2<byte[], PojoValue> value, Collector<Object> out) throws Exception {
				assertNull(value.f1);
				counter++;
				if (counter == elementCount) {
					throw new SuccessException();
				}
			}
		});
		tryExecute(env, "Read deletes from Kafka");
		deleteTestTopic(topic);
	}
public void runAllDeletesTest ( ) throws Exception { final String topic = "alldeletestest" ; createTestTopic ( topic , 1 , 1 ) ; final int elementCount = 300 ; StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setParallelism ( 1 ) ; env . getConfig ( ) . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; DataStream < Tuple2 < byte [ ] , PojoValue > > kvStream = env . addSource ( new SourceFunction < Tuple2 < byte [ ] , PojoValue > > ( ) { @ Override public void run ( SourceContext < Tuple2 < byte [ ] , PojoValue > > ctx ) throws Exception { Random rnd = new Random ( 1337 ) ; for ( long i = 0 ; i < elementCount ; i ++ ) { final byte [ ] key = new byte [ 200 ] ; rnd . nextBytes ( key ) ; ctx . collect ( new Tuple2 < > ( key , ( PojoValue ) null ) ) ; } } @ Override public void cancel ( ) { } } ) ; TypeInformationKeyValueSerializationSchema < byte [ ] , PojoValue > schema = new TypeInformationKeyValueSerializationSchema < > ( byte [ ] . class , PojoValue . class , env . getConfig ( ) ) ; Properties producerProperties = FlinkKafkaProducerBase . getPropertiesFromBrokerList ( brokerConnectionStrings ) ; producerProperties . setProperty ( "retries" , "3" ) ; producerProperties . putAll ( secureProps ) ; kafkaServer . produceIntoKafka ( kvStream , topic , schema , producerProperties , null ) ; env . execute ( "Write deletes to Kafka" ) ; env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setParallelism ( 1 ) ; env . getConfig ( ) . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; Properties props = new Properties ( ) ; props . putAll ( standardProps ) ; props . putAll ( secureProps ) ; DataStream < Tuple2 < byte [ ] , PojoValue > > fromKafka = env . addSource ( kafkaServer . getConsumer ( topic , schema , props ) ) ; fromKafka . flatMap ( new RichFlatMapFunction < Tuple2 < byte [ ] , PojoValue > , Object > ( ) { long counter = 0 ; @ Override public void flatMap ( Tuple2 < byte [ ] , PojoValue > value , Collector < Object > out ) throws Exception { assertNull ( value . f1 ) ; counter ++ ; if ( counter == elementCount ) { throw new SuccessException ( ) ; } } } ) ; tryExecute ( env , "Read deletes from Kafka" ) ; deleteTestTopic ( topic ) ; }
Test delete behavior and metrics for producer.
Test delete behavior and metrics for producer .
// ----------- Write some data into Kafka ------------------- // ----------- Read the data again ------------------- // ensure that deleted messages are passed as nulls // we got the right number of elements