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
/**
	 * Test metrics reporting for consumer.
	 *
	 * @throws Exception
	 */
Test metrics reporting for consumer.
@throws Exception
Test metrics reporting for consumer . @throws Exception
public void runMetricsTest() throws Throwable {
		final String topic = "metricsStream";
		createTestTopic(topic, 5, 1);
		final Tuple1<Throwable> error = new Tuple1<>(null);
		Runnable job = new Runnable() {
			@Override
			public void run() {
				try {
					final StreamExecutionEnvironment env1 = StreamExecutionEnvironment.getExecutionEnvironment();
					env1.setParallelism(1);
					env1.getConfig().setRestartStrategy(RestartStrategies.noRestart());
					env1.getConfig().disableSysoutLogging();
					env1.disableOperatorChaining(); 
					Properties props = new Properties();
					props.putAll(standardProps);
					props.putAll(secureProps);
					TypeInformationSerializationSchema<Tuple2<Integer, Integer>> schema = new TypeInformationSerializationSchema<>(TypeInfoParser.<Tuple2<Integer, Integer>>parse("Tuple2<Integer, Integer>"), env1.getConfig());
					DataStream<Tuple2<Integer, Integer>> fromKafka = env1.addSource(kafkaServer.getConsumer(topic, schema, standardProps));
					fromKafka.flatMap(new FlatMapFunction<Tuple2<Integer, Integer>, Void>() {
						@Override
						public void flatMap(Tuple2<Integer, Integer> value, Collector<Void> out) throws Exception {
						}
					});
					DataStream<Tuple2<Integer, Integer>> fromGen = env1.addSource(new RichSourceFunction<Tuple2<Integer, Integer>>() {
						boolean running = true;
						@Override
						public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
							int i = 0;
							while (running) {
								ctx.collect(Tuple2.of(i++, getRuntimeContext().getIndexOfThisSubtask()));
								Thread.sleep(1);
							}
						}
						@Override
						public void cancel() {
							running = false;
						}
					});
					kafkaServer.produceIntoKafka(fromGen, topic, new KeyedSerializationSchemaWrapper<>(schema), standardProps, null);
					env1.execute("Metrics test job");
				} catch (Throwable t) {
					if (!(t instanceof JobCancellationException)) { 
						LOG.warn("Got exception during execution", t);
						error.f0 = t;
					}
				}
			}
		};
		Thread jobThread = new Thread(job);
		jobThread.start();
		try {
			MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
			Set<ObjectName> offsetMetrics = mBeanServer.queryNames(new ObjectName("*current-offsets*:*"), null);
			while (offsetMetrics.size() < 5) { 
				if (error.f0 != null) {
					throw error.f0;
				}
				offsetMetrics = mBeanServer.queryNames(new ObjectName("*current-offsets*:*"), null);
				Thread.sleep(50);
			}
			Assert.assertEquals(5, offsetMetrics.size());
			while (true) {
				int numPosOffsets = 0;
				for (ObjectName object : offsetMetrics) {
					Object offset = mBeanServer.getAttribute(object, "Value");
					if ((long) offset >= 0) {
						numPosOffsets++;
					}
				}
				if (numPosOffsets == 5) {
					break;
				}
				Thread.sleep(50);
			}
			Set<ObjectName> producerMetrics = mBeanServer.queryNames(new ObjectName("*KafkaProducer*:*"), null);
			Assert.assertTrue("No producer metrics found", producerMetrics.size() > 30);
			LOG.info("Found all JMX metrics. Cancelling job.");
		} finally {
			JobManagerCommunicationUtils.cancelCurrentJob(flink.getLeaderGateway(timeout));
			jobThread.join();
		}
		if (error.f0 != null) {
			throw error.f0;
		}
		deleteTestTopic(topic);
	}
public void runMetricsTest ( ) throws Throwable { final String topic = "metricsStream" ; createTestTopic ( topic , 5 , 1 ) ; final Tuple1 < Throwable > error = new Tuple1 < > ( null ) ; Runnable job = new Runnable ( ) { @ Override public void run ( ) { try { final StreamExecutionEnvironment env1 = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env1 . setParallelism ( 1 ) ; env1 . getConfig ( ) . setRestartStrategy ( RestartStrategies . noRestart ( ) ) ; env1 . getConfig ( ) . disableSysoutLogging ( ) ; env1 . disableOperatorChaining ( ) ; Properties props = new Properties ( ) ; props . putAll ( standardProps ) ; props . putAll ( secureProps ) ; TypeInformationSerializationSchema < Tuple2 < Integer , Integer > > schema = new TypeInformationSerializationSchema < > ( TypeInfoParser . < Tuple2 < Integer , Integer > > parse ( "Tuple2<Integer, Integer>" ) , env1 . getConfig ( ) ) ; DataStream < Tuple2 < Integer , Integer > > fromKafka = env1 . addSource ( kafkaServer . getConsumer ( topic , schema , standardProps ) ) ; fromKafka . flatMap ( new FlatMapFunction < Tuple2 < Integer , Integer > , Void > ( ) { @ Override public void flatMap ( Tuple2 < Integer , Integer > value , Collector < Void > out ) throws Exception { } } ) ; DataStream < Tuple2 < Integer , Integer > > fromGen = env1 . addSource ( new RichSourceFunction < Tuple2 < Integer , Integer > > ( ) { boolean running = true ; @ Override public void run ( SourceContext < Tuple2 < Integer , Integer > > ctx ) throws Exception { int i = 0 ; while ( running ) { ctx . collect ( Tuple2 . of ( i ++ , getRuntimeContext ( ) . getIndexOfThisSubtask ( ) ) ) ; Thread . sleep ( 1 ) ; } } @ Override public void cancel ( ) { running = false ; } } ) ; kafkaServer . produceIntoKafka ( fromGen , topic , new KeyedSerializationSchemaWrapper < > ( schema ) , standardProps , null ) ; env1 . execute ( "Metrics test job" ) ; } catch ( Throwable t ) { if ( ! ( t instanceof JobCancellationException ) ) { LOG . warn ( "Got exception during execution" , t ) ; error . f0 = t ; } } } } ; Thread jobThread = new Thread ( job ) ; jobThread . start ( ) ; try { MBeanServer mBeanServer = ManagementFactory . getPlatformMBeanServer ( ) ; Set < ObjectName > offsetMetrics = mBeanServer . queryNames ( new ObjectName ( "*current-offsets*:*" ) , null ) ; while ( offsetMetrics . size ( ) < 5 ) { if ( error . f0 != null ) { throw error . f0 ; } offsetMetrics = mBeanServer . queryNames ( new ObjectName ( "*current-offsets*:*" ) , null ) ; Thread . sleep ( 50 ) ; } Assert . assertEquals ( 5 , offsetMetrics . size ( ) ) ; while ( true ) { int numPosOffsets = 0 ; for ( ObjectName object : offsetMetrics ) { Object offset = mBeanServer . getAttribute ( object , "Value" ) ; if ( ( long ) offset >= 0 ) { numPosOffsets ++ ; } } if ( numPosOffsets == 5 ) { break ; } Thread . sleep ( 50 ) ; } Set < ObjectName > producerMetrics = mBeanServer . queryNames ( new ObjectName ( "*KafkaProducer*:*" ) , null ) ; Assert . assertTrue ( "No producer metrics found" , producerMetrics . size ( ) > 30 ) ; LOG . info ( "Found all JMX metrics. Cancelling job." ) ; } finally { JobManagerCommunicationUtils . cancelCurrentJob ( flink . getLeaderGateway ( timeout ) ) ; jobThread . join ( ) ; } if ( error . f0 != null ) { throw error . f0 ; } deleteTestTopic ( topic ) ; }
Test metrics reporting for consumer.
Test metrics reporting for consumer .
// create a stream with 5 topics // start job writing & reading data. // let the source read everything into the network buffers // no op // we'll cancel the job // connect to JMX // wait until we've found all 5 offset metrics // test will time out if metrics are not properly working // fail test early // we can't rely on the consumer to have touched all the partitions already // that's why we'll wait until all five partitions have a positive offset. // The test will fail if we never meet the condition // check that offsets are correctly reported // wait for the consumer to consume on all partitions // check if producer metrics are also available. // cancel // wait for the job to finish (it should due to the cancel command above)