import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Test
public void stringifyingResultsShouldIncorporateAccumulatorLocalValueDirectly() {
	final String name = "a";
	final int targetValue = 314159;
	final IntCounter acc = new IntCounter();
	acc.add(targetValue);
	final Map<String, Accumulator<?, ?>> accumulatorMap = new HashMap<>();
	accumulatorMap.put(name, acc);

	final StringifiedAccumulatorResult[] results = StringifiedAccumulatorResult.stringifyAccumulatorResults(accumulatorMap);

	assertEquals(1, results.length);

	final StringifiedAccumulatorResult firstResult = results[0];
	assertEquals(name, firstResult.getName());
	assertEquals("IntCounter", firstResult.getType());
	assertEquals(Integer.toString(targetValue), firstResult.getValue());
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Override
public void open(Configuration parameters) {

	// Add counters using convenience functions
	this.cntNumLines = getRuntimeContext().getIntCounter("num-lines");
	this.wordsPerLineDistribution = getRuntimeContext().getHistogram("words-per-line");

	// Add built-in accumulator without convenience function
	getRuntimeContext().addAccumulator("open-close-counter", this.openCloseCounter);

	// Add custom counter
	this.distinctWords = new SetAccumulator<>();
	this.getRuntimeContext().addAccumulator("distinct-words", distinctWords);

	// Create counter and test increment
	IntCounter simpleCounter = getRuntimeContext().getIntCounter("simple-counter");
	simpleCounter.add(1);
	Assert.assertEquals(simpleCounter.getLocalValue().intValue(), 1);

	// Test if we get the same counter
	IntCounter simpleCounter2 = getRuntimeContext().getIntCounter("simple-counter");
	Assert.assertEquals(simpleCounter.getLocalValue(), simpleCounter2.getLocalValue());

	// Should fail if we request it with different type
	try {
		@SuppressWarnings("unused")
		DoubleCounter simpleCounter3 = getRuntimeContext().getDoubleCounter("simple-counter");
		// DoubleSumAggregator longAggregator3 = (DoubleSumAggregator)
		// getRuntimeContext().getAggregator("custom",
		// DoubleSumAggregator.class);
		Assert.fail("Should not be able to obtain previously created counter with different type");
	}
	catch (UnsupportedOperationException ex) {
		// expected!
	}

	// Test counter used in open() and closed()
	this.openCloseCounter.add(0.5);
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Override
public void open() throws Exception {
	super.open();

	timerService = getInternalTimerService(
			"timer",
			LongSerializer.INSTANCE,
			this);

	getRuntimeContext().addAccumulator(SUCCESSFUL_PROCESS_CHECK_ACCUMULATOR, new IntCounter());
	getRuntimeContext().addAccumulator(SUCCESSFUL_EVENT_TIME_CHECK_ACCUMULATOR, new IntCounter());
	getRuntimeContext().addAccumulator(SUCCESSFUL_PROCESSING_TIME_CHECK_ACCUMULATOR, new IntCounter());
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Override
public void open(Configuration parameters) {
  
	// Add counters using convenience functions
	this.cntNumLines = getRuntimeContext().getIntCounter("num-lines");
	this.wordsPerLineDistribution = getRuntimeContext().getHistogram("words-per-line");

	// Add built-in accumulator without convenience function
	getRuntimeContext().addAccumulator("open-close-counter", this.openCloseCounter);

	// Add custom counter. Didn't find a way to do this with
	// getAccumulator()
	this.distinctWords = new SetAccumulator<StringValue>();
	this.getRuntimeContext().addAccumulator("distinct-words", distinctWords);

	// Create counter and test increment
	IntCounter simpleCounter = getRuntimeContext().getIntCounter("simple-counter");
	simpleCounter.add(1);
	Assert.assertEquals(simpleCounter.getLocalValue().intValue(), 1);

	// Test if we get the same counter
	IntCounter simpleCounter2 = getRuntimeContext().getIntCounter("simple-counter");
	Assert.assertEquals(simpleCounter.getLocalValue(), simpleCounter2.getLocalValue());

	// Should fail if we request it with different type
	try {
		@SuppressWarnings("unused")
		DoubleCounter simpleCounter3 = getRuntimeContext().getDoubleCounter("simple-counter");
		// DoubleSumAggregator longAggregator3 = (DoubleSumAggregator)
		// getRuntimeContext().getAggregator("custom",
		// DoubleSumAggregator.class);
		Assert.fail("Should not be able to obtain previously created counter with different type");
	} catch (UnsupportedOperationException ex) {
	}

	// Test counter used in open() and closed()
	this.openCloseCounter.add(0.5);
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
private static boolean checkUserAccumulators(int expected, Map<String, Accumulator<?, ?>> accumulatorMap) {
	LOG.info("checking user accumulators");
	return accumulatorMap.containsKey(ACCUMULATOR_NAME) && expected == ((IntCounter) accumulatorMap.get(ACCUMULATOR_NAME)).getLocalValue();
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Override
public void open(Configuration parameters) throws Exception {
	super.open(parameters);

	getRuntimeContext().addAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR, new IntCounter());
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Override
public IntCounter getIntCounter(String name) {
	return (IntCounter) getAccumulator(name, IntCounter.class);
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Test
public void testHandleRequest() throws Exception {

	// Instance the handler.
	final RestHandlerConfiguration restHandlerConfiguration = RestHandlerConfiguration.fromConfiguration(new Configuration());

	final SubtaskExecutionAttemptAccumulatorsHandler handler = new SubtaskExecutionAttemptAccumulatorsHandler(
		CompletableFuture.completedFuture("127.0.0.1:9527"),
		() -> null,
		Time.milliseconds(100L),
		restHandlerConfiguration.getResponseHeaders(),
		null,
		new ExecutionGraphCache(
			restHandlerConfiguration.getTimeout(),
			Time.milliseconds(restHandlerConfiguration.getRefreshInterval())),
		TestingUtils.defaultExecutor());

	// Instance a empty request.
	final HandlerRequest<EmptyRequestBody, SubtaskAttemptMessageParameters> request = new HandlerRequest<>(
		EmptyRequestBody.getInstance(),
		new SubtaskAttemptMessageParameters()
	);

	final Map<String, Accumulator<?, ?>> userAccumulators = new HashMap<>(2);
	userAccumulators.put("IntCounter", new IntCounter(10));
	userAccumulators.put("LongCounter", new LongCounter(100L));

	// Instance the expected result.
	final StringifiedAccumulatorResult[] accumulatorResults =
		StringifiedAccumulatorResult.stringifyAccumulatorResults(userAccumulators);

	final int attemptNum = 1;
	final int subtaskIndex = 2;

	// Instance the tested execution.
	final ArchivedExecution execution = new ArchivedExecution(
		accumulatorResults,
		null,
		new ExecutionAttemptID(),
		attemptNum,
		ExecutionState.FINISHED,
		null,
		null,
		subtaskIndex,
		new long[ExecutionState.values().length]);

	// Invoke tested method.
	final SubtaskExecutionAttemptAccumulatorsInfo accumulatorsInfo = handler.handleRequest(request, execution);

	final ArrayList<UserAccumulator> userAccumulatorList = new ArrayList<>(userAccumulators.size());
	for (StringifiedAccumulatorResult accumulatorResult : accumulatorResults) {
		userAccumulatorList.add(
			new UserAccumulator(
				accumulatorResult.getName(),
				accumulatorResult.getType(),
				accumulatorResult.getValue()));
	}

	final SubtaskExecutionAttemptAccumulatorsInfo expected = new SubtaskExecutionAttemptAccumulatorsInfo(
		subtaskIndex,
		attemptNum,
		execution.getAttemptId().toString(),
		userAccumulatorList);

	// Verify.
	assertEquals(expected, accumulatorsInfo);
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
/**
 * Verifies that {@link ExecutionGraph#updateState(TaskExecutionState)} updates the accumulators and metrics for an
 * execution that failed or was canceled.
 */
@Test
public void testAccumulatorsAndMetricsForwarding() throws Exception {
	final JobVertexID jid1 = new JobVertexID();
	final JobVertexID jid2 = new JobVertexID();

	JobVertex v1 = new JobVertex("v1", jid1);
	JobVertex v2 = new JobVertex("v2", jid2);

	Tuple2<ExecutionGraph, Map<ExecutionAttemptID, Execution>> graphAndExecutions = setupExecution(v1, 1, v2, 1);
	ExecutionGraph graph = graphAndExecutions.f0;
	
	// verify behavior for canceled executions
	Execution execution1 = graphAndExecutions.f1.values().iterator().next();

	IOMetrics ioMetrics = new IOMetrics(0, 0, 0, 0, 0, 0.0, 0.0, 0.0, 0.0, 0.0);
	Map<String, Accumulator<?, ?>> accumulators = new HashMap<>();
	accumulators.put("acc", new IntCounter(4));
	AccumulatorSnapshot accumulatorSnapshot = new AccumulatorSnapshot(graph.getJobID(), execution1.getAttemptId(), accumulators);
	
	TaskExecutionState state = new TaskExecutionState(graph.getJobID(), execution1.getAttemptId(), ExecutionState.CANCELED, null, accumulatorSnapshot, ioMetrics);
	
	graph.updateState(state);
	
	assertEquals(ioMetrics, execution1.getIOMetrics());
	assertNotNull(execution1.getUserAccumulators());
	assertEquals(4, execution1.getUserAccumulators().get("acc").getLocalValue());
	
	// verify behavior for failed executions
	Execution execution2 = graphAndExecutions.f1.values().iterator().next();

	IOMetrics ioMetrics2 = new IOMetrics(0, 0, 0, 0, 0, 0.0, 0.0, 0.0, 0.0, 0.0);
	Map<String, Accumulator<?, ?>> accumulators2 = new HashMap<>();
	accumulators2.put("acc", new IntCounter(8));
	AccumulatorSnapshot accumulatorSnapshot2 = new AccumulatorSnapshot(graph.getJobID(), execution2.getAttemptId(), accumulators2);

	TaskExecutionState state2 = new TaskExecutionState(graph.getJobID(), execution2.getAttemptId(), ExecutionState.FAILED, null, accumulatorSnapshot2, ioMetrics2);

	graph.updateState(state2);

	assertEquals(ioMetrics2, execution2.getIOMetrics());
	assertNotNull(execution2.getUserAccumulators());
	assertEquals(8, execution2.getUserAccumulators().get("acc").getLocalValue());
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Override
public IntCounter getIntCounter(String name) {
	throw new UnsupportedOperationException("Int counters are not supported in rich async functions.");
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Override
public IntCounter getIntCounter(String name) {
	throw new UnsupportedOperationException();
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
@Override
public IntCounter getIntCounter(String name) {
	return context.getIntCounter(name);
}
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
/**
 * Convenience function to create a counter object for integers.
 */
@PublicEvolving
IntCounter getIntCounter(String name);
 

import org.apache.flink.api.common.accumulators.IntCounter; //导入依赖的package包/类
/**
 * Convenience function to create a counter object for integers.
 */
IntCounter getIntCounter(String name);