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 earliest 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 earliest 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 earliest record the consumer ignores the " auto . offset . reset " behaviour as well as any committed group offsets in Kafka .
public void runStartFromEarliestOffsets() throws Exception {
		final int parallelism = 3;
		final int recordsInEachPartition = 50;
		final String topicName = writeSequence("testStartFromEarliestOffsetsTopic", recordsInEachPartition, parallelism, 1);
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.getConfig().disableSysoutLogging();
		env.setParallelism(parallelism);
		Properties readProps = new Properties();
		readProps.putAll(standardProps);
		readProps.setProperty("auto.offset.reset", "latest"); 
		KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
		kafkaOffsetHandler.setCommittedOffset(topicName, 0, 23);
		kafkaOffsetHandler.setCommittedOffset(topicName, 1, 31);
		kafkaOffsetHandler.setCommittedOffset(topicName, 2, 43);
		readSequence(env, StartupMode.EARLIEST, null, readProps, parallelism, topicName, recordsInEachPartition, 0);
		kafkaOffsetHandler.close();
		deleteTestTopic(topicName);
	}
public void runStartFromEarliestOffsets ( ) throws Exception { final int parallelism = 3 ; final int recordsInEachPartition = 50 ; final String topicName = writeSequence ( "testStartFromEarliestOffsetsTopic" , recordsInEachPartition , parallelism , 1 ) ; final StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; env . setParallelism ( parallelism ) ; Properties readProps = new Properties ( ) ; readProps . putAll ( standardProps ) ; readProps . setProperty ( "auto.offset.reset" , "latest" ) ; KafkaTestEnvironment . KafkaOffsetHandler kafkaOffsetHandler = kafkaServer . createOffsetHandler ( ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 0 , 23 ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 1 , 31 ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 2 , 43 ) ; readSequence ( env , StartupMode . EARLIEST , null , readProps , parallelism , topicName , recordsInEachPartition , 0 ) ; kafkaOffsetHandler . close ( ) ; deleteTestTopic ( topicName ) ; }
This test ensures that when explicitly set to start from earliest 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 earliest record the consumer ignores the " auto . offset . reset " behaviour as well as any committed group offsets in Kafka .
// 3 partitions with 50 records each (0-49, so the expected commit offset of each partition should be 50) // this should be ignored // the committed offsets should be ignored