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 the consumer correctly uses group offsets in Kafka, and defaults to "auto.offset.reset"
	 * behaviour when necessary, when explicitly configured to start from group offsets.
	 *
	 * <p>The partitions and their committed group offsets are setup as:
	 * 	partition 0 --> committed offset 23
	 * 	partition 1 --> no commit offset
	 * 	partition 2 --> committed offset 43
	 *
	 * <p>When configured to start from group offsets, each partition should read:
	 * 	partition 0 --> start from offset 23, read to offset 49 (27 records)
	 * 	partition 1 --> default to "auto.offset.reset" (set to earliest), so start from offset 0, read to offset 49 (50 records)
	 * 	partition 2 --> start from offset 43, read to offset 49 (7 records)
	 */
This test ensures that the consumer correctly uses group offsets in Kafka, and defaults to "auto.offset.reset"
behaviour when necessary, when explicitly configured to start from group offsets.




This test ensures that the consumer correctly uses group offsets in Kafka and defaults to " auto . offset . reset " behaviour when necessary when explicitly configured to start from group offsets .
public void runStartFromGroupOffsets() throws Exception {
		final int parallelism = 3;
		final int recordsInEachPartition = 50;
		final String topicName = writeSequence("testStartFromGroupOffsetsTopic", 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", "earliest");
		KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
		kafkaOffsetHandler.setCommittedOffset(topicName, 0, 23);
		kafkaOffsetHandler.setCommittedOffset(topicName, 2, 43);
		Map<Integer, Tuple2<Integer, Integer>> partitionsToValueCountAndStartOffsets = new HashMap<>();
		partitionsToValueCountAndStartOffsets.put(0, new Tuple2<>(27, 23)); 
		partitionsToValueCountAndStartOffsets.put(1, new Tuple2<>(50, 0)); 
		partitionsToValueCountAndStartOffsets.put(2, new Tuple2<>(7, 43));	
		readSequence(env, StartupMode.GROUP_OFFSETS, null, readProps, topicName, partitionsToValueCountAndStartOffsets);
		kafkaOffsetHandler.close();
		deleteTestTopic(topicName);
	}
public void runStartFromGroupOffsets ( ) throws Exception { final int parallelism = 3 ; final int recordsInEachPartition = 50 ; final String topicName = writeSequence ( "testStartFromGroupOffsetsTopic" , 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" , "earliest" ) ; KafkaTestEnvironment . KafkaOffsetHandler kafkaOffsetHandler = kafkaServer . createOffsetHandler ( ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 0 , 23 ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 2 , 43 ) ; Map < Integer , Tuple2 < Integer , Integer > > partitionsToValueCountAndStartOffsets = new HashMap < > ( ) ; partitionsToValueCountAndStartOffsets . put ( 0 , new Tuple2 < > ( 27 , 23 ) ) ; partitionsToValueCountAndStartOffsets . put ( 1 , new Tuple2 < > ( 50 , 0 ) ) ; partitionsToValueCountAndStartOffsets . put ( 2 , new Tuple2 < > ( 7 , 43 ) ) ; readSequence ( env , StartupMode . GROUP_OFFSETS , null , readProps , topicName , partitionsToValueCountAndStartOffsets ) ; kafkaOffsetHandler . close ( ) ; deleteTestTopic ( topicName ) ; }
This test ensures that the consumer correctly uses group offsets in Kafka, and defaults to "auto.offset.reset"
 behaviour when necessary, when explicitly configured to start from group offsets.
This test ensures that the consumer correctly uses group offsets in Kafka and defaults to " auto . offset . reset " behaviour when necessary when explicitly configured to start from group offsets .
// 3 partitions with 50 records each (offsets 0-49) // the committed group offsets should be used as starting points // only partitions 0 and 2 have group offsets committed // partition 0 should read offset 23-49 // partition 1 should read offset 0-49 // partition 2 should read offset 43-49