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 user-supplied specific offsets when explicitly configured to
	 * start from specific offsets. For partitions which a specific offset can not be found for, the starting position
	 * for them should fallback to the group offsets behaviour.
	 *
	 * <p>4 partitions will have 50 records with offsets 0 to 49. The supplied specific offsets map is:
	 * 	partition 0 --> start from offset 19
	 * 	partition 1 --> not set
	 * 	partition 2 --> start from offset 22
	 * 	partition 3 --> not set
	 * 	partition 4 --> start from offset 26 (this should be ignored because the partition does not exist)
	 *
	 * <p>The partitions and their committed group offsets are setup as:
	 * 	partition 0 --> committed offset 23
	 * 	partition 1 --> committed offset 31
	 * 	partition 2 --> committed offset 43
	 * 	partition 3 --> no commit offset
	 *
	 * <p>When configured to start from these specific offsets, each partition should read:
	 * 	partition 0 --> start from offset 19, read to offset 49 (31 records)
	 * 	partition 1 --> fallback to group offsets, so start from offset 31, read to offset 49 (19 records)
	 * 	partition 2 --> start from offset 22, read to offset 49 (28 records)
	 * 	partition 3 --> fallback to group offsets, but since there is no group offset for this partition,
	 * 	                will default to "auto.offset.reset" (set to "earliest"),
	 * 	                so start from offset 0, read to offset 49 (50 records)
	 */
This test ensures that the consumer correctly uses user-supplied specific offsets when explicitly configured to
start from specific offsets. For partitions which a specific offset can not be found for, the starting position
for them should fallback to the group offsets behaviour.

4 partitions will have 50 records with offsets 0 to 49.




This test ensures that the consumer correctly uses user - supplied specific offsets when explicitly configured to start from specific offsets . For partitions which a specific offset can not be found for the starting position for them should fallback to the group offsets behaviour . 4 partitions will have 50 records with offsets 0 to 49 .
public void runStartFromSpecificOffsets() throws Exception {
		final int parallelism = 4;
		final int recordsInEachPartition = 50;
		final String topicName = writeSequence("testStartFromSpecificOffsetsTopic", 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"); 
		Map<KafkaTopicPartition, Long> specificStartupOffsets = new HashMap<>();
		specificStartupOffsets.put(new KafkaTopicPartition(topicName, 0), 19L);
		specificStartupOffsets.put(new KafkaTopicPartition(topicName, 2), 22L);
		specificStartupOffsets.put(new KafkaTopicPartition(topicName, 4), 26L); 
		KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
		kafkaOffsetHandler.setCommittedOffset(topicName, 0, 23);
		kafkaOffsetHandler.setCommittedOffset(topicName, 1, 31);
		kafkaOffsetHandler.setCommittedOffset(topicName, 2, 43);
		Map<Integer, Tuple2<Integer, Integer>> partitionsToValueCountAndStartOffsets = new HashMap<>();
		partitionsToValueCountAndStartOffsets.put(0, new Tuple2<>(31, 19)); 
		partitionsToValueCountAndStartOffsets.put(1, new Tuple2<>(19, 31)); 
		partitionsToValueCountAndStartOffsets.put(2, new Tuple2<>(28, 22));	
		partitionsToValueCountAndStartOffsets.put(3, new Tuple2<>(50, 0));	
		readSequence(env, StartupMode.SPECIFIC_OFFSETS, specificStartupOffsets, readProps, topicName, partitionsToValueCountAndStartOffsets);
		kafkaOffsetHandler.close();
		deleteTestTopic(topicName);
	}
public void runStartFromSpecificOffsets ( ) throws Exception { final int parallelism = 4 ; final int recordsInEachPartition = 50 ; final String topicName = writeSequence ( "testStartFromSpecificOffsetsTopic" , 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" ) ; Map < KafkaTopicPartition , Long > specificStartupOffsets = new HashMap < > ( ) ; specificStartupOffsets . put ( new KafkaTopicPartition ( topicName , 0 ) , 19L ) ; specificStartupOffsets . put ( new KafkaTopicPartition ( topicName , 2 ) , 22L ) ; specificStartupOffsets . put ( new KafkaTopicPartition ( topicName , 4 ) , 26L ) ; KafkaTestEnvironment . KafkaOffsetHandler kafkaOffsetHandler = kafkaServer . createOffsetHandler ( ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 0 , 23 ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 1 , 31 ) ; kafkaOffsetHandler . setCommittedOffset ( topicName , 2 , 43 ) ; Map < Integer , Tuple2 < Integer , Integer > > partitionsToValueCountAndStartOffsets = new HashMap < > ( ) ; partitionsToValueCountAndStartOffsets . put ( 0 , new Tuple2 < > ( 31 , 19 ) ) ; partitionsToValueCountAndStartOffsets . put ( 1 , new Tuple2 < > ( 19 , 31 ) ) ; partitionsToValueCountAndStartOffsets . put ( 2 , new Tuple2 < > ( 28 , 22 ) ) ; partitionsToValueCountAndStartOffsets . put ( 3 , new Tuple2 < > ( 50 , 0 ) ) ; readSequence ( env , StartupMode . SPECIFIC_OFFSETS , specificStartupOffsets , readProps , topicName , partitionsToValueCountAndStartOffsets ) ; kafkaOffsetHandler . close ( ) ; deleteTestTopic ( topicName ) ; }
This test ensures that the consumer correctly uses user-supplied specific offsets when explicitly configured to
 start from specific offsets.
This test ensures that the consumer correctly uses user - supplied specific offsets when explicitly configured to start from specific offsets .
// 4 partitions with 50 records each (offsets 0-49) // partition 3 should default back to this behaviour // non-existing partition, should be ignored // only the committed offset for partition 1 should be used, because partition 1 has no entry in specific offset map // partition 0 should read offset 19-49 // partition 1 should read offset 31-49 // partition 2 should read offset 22-49 // partition 3 should read offset 0-49