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 the consumers retrieve some start offset from kafka (earliest, latest), that this offset
	 * is committed to Kafka, even if some partitions are not read.
	 *
	 * <p>Test:
	 * - Create 3 partitions
	 * - write 50 messages into each.
	 * - Start three consumers with auto.offset.reset='latest' and wait until they committed into Kafka.
	 * - Check if the offsets in Kafka are set to 50 for the three partitions
	 *
	 * <p>See FLINK-3440 as well
	 */
This test ensures that when the consumers retrieve some start offset from kafka (earliest, latest), that this offset
is committed to Kafka, even if some partitions are not read.

Test:
- Create 3 partitions
- write 50 messages into each.
- Start three consumers with auto.offset.reset='latest' and wait until they committed into Kafka.
- Check if the offsets in Kafka are set to 50 for the three partitions

See FLINK-3440 as well
This test ensures that when the consumers retrieve some start offset from kafka ( earliest latest ) that this offset is committed to Kafka even if some partitions are not read . Test : - Create 3 partitions - write 50 messages into each . - Start three consumers with auto . offset . reset = ' latest ' and wait until they committed into Kafka . - Check if the offsets in Kafka are set to 50 for the three partitions See FLINK - 3440 as well
public void runAutoOffsetRetrievalAndCommitToKafka() throws Exception {
		final int parallelism = 3;
		final int recordsInEachPartition = 50;
		final String topicName = writeSequence("testAutoOffsetRetrievalAndCommitToKafkaTopic", recordsInEachPartition, parallelism, 1);
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.getConfig().disableSysoutLogging();
		env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
		env.setParallelism(parallelism);
		env.enableCheckpointing(200);
		Properties readProps = new Properties();
		readProps.putAll(standardProps);
		readProps.setProperty("auto.offset.reset", "latest"); 
		DataStream<String> stream = env.addSource(kafkaServer.getConsumer(topicName, new SimpleStringSchema(), readProps));
		stream.addSink(new DiscardingSink<String>());
		final AtomicReference<Throwable> errorRef = new AtomicReference<>();
		final Thread runner = new Thread("runner") {
			@Override
			public void run() {
				try {
					env.execute();
				}
				catch (Throwable t) {
					if (!(t instanceof JobCancellationException)) {
						errorRef.set(t);
					}
				}
			}
		};
		runner.start();
		KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
		final Long l50 = 50L; 
		final long deadline = 30_000_000_000L + System.nanoTime();
		do {
			Long o1 = kafkaOffsetHandler.getCommittedOffset(topicName, 0);
			Long o2 = kafkaOffsetHandler.getCommittedOffset(topicName, 1);
			Long o3 = kafkaOffsetHandler.getCommittedOffset(topicName, 2);
			if (l50.equals(o1) && l50.equals(o2) && l50.equals(o3)) {
				break;
			}
			Thread.sleep(100);
		}
		while (System.nanoTime() < deadline);
		JobManagerCommunicationUtils.cancelCurrentJob(flink.getLeaderGateway(timeout));
		runner.join();
		final Throwable t = errorRef.get();
		if (t != null) {
			throw new RuntimeException("Job failed with an exception", t);
		}
		Long o1 = kafkaOffsetHandler.getCommittedOffset(topicName, 0);
		Long o2 = kafkaOffsetHandler.getCommittedOffset(topicName, 1);
		Long o3 = kafkaOffsetHandler.getCommittedOffset(topicName, 2);
		Assert.assertEquals(Long.valueOf(50L), o1);
		Assert.assertEquals(Long.valueOf(50L), o2);
		Assert.assertEquals(Long.valueOf(50L), o3);
		kafkaOffsetHandler.close();
		deleteTestTopic(topicName);
	}
public void runAutoOffsetRetrievalAndCommitToKafka ( ) throws Exception { final int parallelism = 3 ; final int recordsInEachPartition = 50 ; final String topicName = writeSequence ( "testAutoOffsetRetrievalAndCommitToKafkaTopic" , recordsInEachPartition , parallelism , 1 ) ; final StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; env . getConfig ( ) . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; env . setParallelism ( parallelism ) ; env . enableCheckpointing ( 200 ) ; Properties readProps = new Properties ( ) ; readProps . putAll ( standardProps ) ; readProps . setProperty ( "auto.offset.reset" , "latest" ) ; DataStream < String > stream = env . addSource ( kafkaServer . getConsumer ( topicName , new SimpleStringSchema ( ) , readProps ) ) ; stream . addSink ( new DiscardingSink < String > ( ) ) ; final AtomicReference < Throwable > errorRef = new AtomicReference < > ( ) ; final Thread runner = new Thread ( "runner" ) { @ Override public void run ( ) { try { env . execute ( ) ; } catch ( Throwable t ) { if ( ! ( t instanceof JobCancellationException ) ) { errorRef . set ( t ) ; } } } } ; runner . start ( ) ; KafkaTestEnvironment . KafkaOffsetHandler kafkaOffsetHandler = kafkaServer . createOffsetHandler ( ) ; final Long l50 = 50L ; final long deadline = 30_000_000_000L + System . nanoTime ( ) ; do { Long o1 = kafkaOffsetHandler . getCommittedOffset ( topicName , 0 ) ; Long o2 = kafkaOffsetHandler . getCommittedOffset ( topicName , 1 ) ; Long o3 = kafkaOffsetHandler . getCommittedOffset ( topicName , 2 ) ; if ( l50 . equals ( o1 ) && l50 . equals ( o2 ) && l50 . equals ( o3 ) ) { break ; } Thread . sleep ( 100 ) ; } while ( System . nanoTime ( ) < deadline ) ; JobManagerCommunicationUtils . cancelCurrentJob ( flink . getLeaderGateway ( timeout ) ) ; runner . join ( ) ; final Throwable t = errorRef . get ( ) ; if ( t != null ) { throw new RuntimeException ( "Job failed with an exception" , t ) ; } Long o1 = kafkaOffsetHandler . getCommittedOffset ( topicName , 0 ) ; Long o2 = kafkaOffsetHandler . getCommittedOffset ( topicName , 1 ) ; Long o3 = kafkaOffsetHandler . getCommittedOffset ( topicName , 2 ) ; Assert . assertEquals ( Long . valueOf ( 50L ) , o1 ) ; Assert . assertEquals ( Long . valueOf ( 50L ) , o2 ) ; Assert . assertEquals ( Long . valueOf ( 50L ) , o3 ) ; kafkaOffsetHandler . close ( ) ; deleteTestTopic ( topicName ) ; }
This test ensures that when the consumers retrieve some start offset from kafka (earliest, latest), that this offset
 is committed to Kafka, even if some partitions are not read.
This test ensures that when the consumers retrieve some start offset from kafka ( earliest latest ) that this offset is committed to Kafka even if some partitions are not read .
// 3 partitions with 50 records each (0-49, so the expected commit offset of each partition should be 50) // set to reset to latest, so that partitions are initially not read // the final committed offset in Kafka should be 50 // cancel the job & wait for the job to finish // final check to see if offsets are correctly in Kafka