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
/**
	 * Ensures that the committed offsets to Kafka are the offsets of "the next record to process".
	 */
Ensures that the committed offsets to Kafka are the offsets of "the next record to process".
Ensures that the committed offsets to Kafka are the offsets of " the next record to process " .
public void runCommitOffsetsToKafka() throws Exception {
		final int parallelism = 3;
		final int recordsInEachPartition = 50;
		final String topicName = writeSequence("testCommitOffsetsToKafkaTopic", recordsInEachPartition, parallelism, 1);
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.getConfig().disableSysoutLogging();
		env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
		env.setParallelism(parallelism);
		env.enableCheckpointing(200);
		DataStream<String> stream = env.addSource(kafkaServer.getConsumer(topicName, new SimpleStringSchema(), standardProps));
		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();
		final Long l50 = 50L; 
		final long deadline = 30_000_000_000L + System.nanoTime();
		KafkaTestEnvironment.KafkaOffsetHandler kafkaOffsetHandler = kafkaServer.createOffsetHandler();
		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 runCommitOffsetsToKafka ( ) throws Exception { final int parallelism = 3 ; final int recordsInEachPartition = 50 ; final String topicName = writeSequence ( "testCommitOffsetsToKafkaTopic" , recordsInEachPartition , parallelism , 1 ) ; final StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . getConfig ( ) . disableSysoutLogging ( ) ; env . getConfig ( ) . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; env . setParallelism ( parallelism ) ; env . enableCheckpointing ( 200 ) ; DataStream < String > stream = env . addSource ( kafkaServer . getConsumer ( topicName , new SimpleStringSchema ( ) , standardProps ) ) ; 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 ( ) ; final Long l50 = 50L ; final long deadline = 30_000_000_000L + System . nanoTime ( ) ; KafkaTestEnvironment . KafkaOffsetHandler kafkaOffsetHandler = kafkaServer . createOffsetHandler ( ) ; 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 ) ; }
Ensures that the committed offsets to Kafka are the offsets of "the next record to process".
Ensures that the committed offsets to Kafka are the offsets of " the next record to process " .
// 3 partitions with 50 records each (0-49, so the expected commit offset of each partition should be 50) // 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