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
/**
	 * This test ensures that when explicitly set to start from latest record, the consumer
	 * ignores the "auto.offset.reset" behaviour as well as any committed group offsets in Kafka.
	 */
This test ensures that when explicitly set to start from latest record, the consumer
ignores the "auto.offset.reset" behaviour as well as any committed group offsets in Kafka.
This test ensures that when explicitly set to start from latest record the consumer ignores the " auto . offset . reset " behaviour as well as any committed group offsets in Kafka .
public void runStartFromLatestOffsets() throws Exception {
		final int parallelism = 3;
		final int recordsInEachPartition = 50;
		final int extraRecordsInEachPartition = 200;
		final String topicName = writeSequence("testStartFromLatestOffsetsTopic", recordsInEachPartition, parallelism, 1);
		KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
		kafkaOffsetHandler.setCommittedOffset(topicName, 0, 23);
		kafkaOffsetHandler.setCommittedOffset(topicName, 1, 31);
		kafkaOffsetHandler.setCommittedOffset(topicName, 2, 43);
		final String consumeExtraRecordsJobName = "Consume Extra Records Job";
		final String writeExtraRecordsJobName = "Write Extra Records Job";
		final TypeInformation<Tuple2<Integer, Integer>> resultType =
			TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {});
		final KeyedSerializationSchema<Tuple2<Integer, Integer>> serSchema =
			new KeyedSerializationSchemaWrapper<>(
				new TypeInformationSerializationSchema<>(resultType, new ExecutionConfig()));
		final KeyedDeserializationSchema<Tuple2<Integer, Integer>> deserSchema =
			new KeyedDeserializationSchemaWrapper<>(
				new TypeInformationSerializationSchema<>(resultType, new ExecutionConfig()));
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.getConfig().disableSysoutLogging();
		env.setParallelism(parallelism);
		final Properties readProps = new Properties();
		readProps.putAll(standardProps);
		readProps.setProperty("auto.offset.reset", "earliest"); 
		FlinkKafkaConsumerBase<Tuple2<Integer, Integer>> latestReadingConsumer =
			kafkaServer.getConsumer(topicName, deserSchema, readProps);
		latestReadingConsumer.setStartFromLatest();
		env
			.addSource(latestReadingConsumer).setParallelism(parallelism)
			.flatMap(new FlatMapFunction<Tuple2<Integer, Integer>, Object>() {
				@Override
				public void flatMap(Tuple2<Integer, Integer> value, Collector<Object> out) throws Exception {
					if (value.f1 - recordsInEachPartition < 0) {
						throw new RuntimeException("test failed; consumed a record that was previously written: " + value);
					}
				}
			}).setParallelism(1)
			.addSink(new DiscardingSink<>());
		final AtomicReference<Throwable> error = new AtomicReference<>();
		Thread consumeThread = new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					env.execute(consumeExtraRecordsJobName);
				} catch (Throwable t) {
					if (!(t instanceof JobCancellationException)) {
						error.set(t);
					}
				}
			}
		});
		consumeThread.start();
		JobManagerCommunicationUtils.waitUntilJobIsRunning(
			flink.getLeaderGateway(timeout),
			consumeExtraRecordsJobName);
		final StreamExecutionEnvironment env2 = StreamExecutionEnvironment.getExecutionEnvironment();
		env2.setParallelism(parallelism);
		DataStream<Tuple2<Integer, Integer>> extraRecordsStream = env2
			.addSource(new RichParallelSourceFunction<Tuple2<Integer, Integer>>() {
				private boolean running = true;
				@Override
				public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
					int count = recordsInEachPartition; 
					int partition = getRuntimeContext().getIndexOfThisSubtask();
					while (running && count < recordsInEachPartition + extraRecordsInEachPartition) {
						ctx.collect(new Tuple2<>(partition, count));
						count++;
					}
				}
				@Override
				public void cancel() {
					running = false;
				}
			});
		kafkaServer.produceIntoKafka(extraRecordsStream, topicName, serSchema, readProps, null);
		try {
			env2.execute(writeExtraRecordsJobName);
		}
		catch (Exception e) {
			throw new RuntimeException("Writing extra records failed", e);
		}
		JobManagerCommunicationUtils.cancelCurrentJob(
			flink.getLeaderGateway(timeout),
			consumeExtraRecordsJobName);
		consumeThread.join();
		kafkaOffsetHandler.close();
		deleteTestTopic(topicName);
		final Throwable consumerError = error.get();
		if (consumerError != null) {
			throw new Exception("Exception in the consuming thread", consumerError);
		}
	}
public void runStartFromLatestOffsets ( ) throws Exception { final int parallelism = 3 ; final int recordsInEachPartition = 50 ; final int extraRecordsInEachPartition = 200 ; final String topicName = writeSequence ( "testStartFromLatestOffsetsTopic" , recordsInEachPartition , parallelism , 1 ) ; KafkaTestEnvironment . KafkaOffsetHandler kafkaOffsetHandler = kafkaServer . createOffsetHandler ( ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 0 , 23 ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 1 , 31 ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 2 , 43 ) ; final String consumeExtraRecordsJobName = "Consume Extra Records Job" ; final String writeExtraRecordsJobName = "Write Extra Records Job" ; final TypeInformation < Tuple2 < Integer , Integer > > resultType = TypeInformation . of ( new TypeHint < Tuple2 < Integer , Integer > > ( ) { } ) ; final KeyedSerializationSchema < Tuple2 < Integer , Integer > > serSchema = new KeyedSerializationSchemaWrapper < > ( new TypeInformationSerializationSchema < > ( resultType , new ExecutionConfig ( ) ) ) ; final KeyedDeserializationSchema < Tuple2 < Integer , Integer > > deserSchema = new KeyedDeserializationSchemaWrapper < > ( new TypeInformationSerializationSchema < > ( resultType , new ExecutionConfig ( ) ) ) ; final StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; env . setParallelism ( parallelism ) ; final Properties readProps = new Properties ( ) ; readProps . putAll ( standardProps ) ; readProps . setProperty ( "auto.offset.reset" , "earliest" ) ; FlinkKafkaConsumerBase < Tuple2 < Integer , Integer > > latestReadingConsumer = kafkaServer . getConsumer ( topicName , deserSchema , readProps ) ; latestReadingConsumer . setStartFromLatest ( ) ; env . addSource ( latestReadingConsumer ) . setParallelism ( parallelism ) . flatMap ( new FlatMapFunction < Tuple2 < Integer , Integer > , Object > ( ) { @ Override public void flatMap ( Tuple2 < Integer , Integer > value , Collector < Object > out ) throws Exception { if ( value . f1 - recordsInEachPartition < 0 ) { throw new RuntimeException ( "test failed; consumed a record that was previously written: " + value ) ; } } } ) . setParallelism ( 1 ) . addSink ( new DiscardingSink < > ( ) ) ; final AtomicReference < Throwable > error = new AtomicReference < > ( ) ; Thread consumeThread = new Thread ( new Runnable ( ) { @ Override public void run ( ) { try { env . execute ( consumeExtraRecordsJobName ) ; } catch ( Throwable t ) { if ( ! ( t instanceof JobCancellationException ) ) { error . set ( t ) ; } } } } ) ; consumeThread . start ( ) ; JobManagerCommunicationUtils . waitUntilJobIsRunning ( flink . getLeaderGateway ( timeout ) , consumeExtraRecordsJobName ) ; final StreamExecutionEnvironment env2 = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env2 . setParallelism ( parallelism ) ; DataStream < Tuple2 < Integer , Integer > > extraRecordsStream = env2 . addSource ( new RichParallelSourceFunction < Tuple2 < Integer , Integer > > ( ) { private boolean running = true ; @ Override public void run ( SourceContext < Tuple2 < Integer , Integer > > ctx ) throws Exception { int count = recordsInEachPartition ; int partition = getRuntimeContext ( ) . getIndexOfThisSubtask ( ) ; while ( running && count < recordsInEachPartition + extraRecordsInEachPartition ) { ctx . collect ( new Tuple2 < > ( partition , count ) ) ; count ++ ; } } @ Override public void cancel ( ) { running = false ; } } ) ; kafkaServer . produceIntoKafka ( extraRecordsStream , topicName , serSchema , readProps , null ) ; try { env2 . execute ( writeExtraRecordsJobName ) ; } catch ( Exception e ) { throw new RuntimeException ( "Writing extra records failed" , e ) ; } JobManagerCommunicationUtils . cancelCurrentJob ( flink . getLeaderGateway ( timeout ) , consumeExtraRecordsJobName ) ; consumeThread . join ( ) ; kafkaOffsetHandler . close ( ) ; deleteTestTopic ( topicName ) ; final Throwable consumerError = error . get ( ) ; if ( consumerError != null ) { throw new Exception ( "Exception in the consuming thread" , consumerError ) ; } }
This test ensures that when explicitly set to start from latest record, the consumer
 ignores the "auto.offset.reset" behaviour as well as any committed group offsets in Kafka.
This test ensures that when explicitly set to start from latest record the consumer ignores the " auto . offset . reset " behaviour as well as any committed group offsets in Kafka .
// 50 records written to each of 3 partitions before launching a latest-starting consuming job // each partition will be written an extra 200 records // all already existing data in the topic, before the consuming topology has started, should be ignored // the committed offsets should be ignored // job names for the topologies for writing and consuming the extra records // seriliazation / deserialization schemas for writing and consuming the extra records // setup and run the latest-consuming job // this should be ignored // wait until the consuming job has started, to be extra safe // setup the extra records writing job // the extra records should start from the last written value // cancel the consume job after all extra records are written // check whether the consuming thread threw any test errors; // test will fail here if the consume job had incorrectly read any records other than the extra records