import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
public FlinkMetricContainer(RuntimeContext runtimeContext) {
  this.runtimeContext = runtimeContext;
  this.flinkCounterCache = new HashMap<>();
  this.flinkDistributionGaugeCache = new HashMap<>();
  this.flinkGaugeCache = new HashMap<>();

  Accumulator<MetricsContainerStepMap, MetricsContainerStepMap> metricsAccumulator =
      runtimeContext.getAccumulator(ACCUMULATOR_NAME);
  if (metricsAccumulator == null) {
    metricsAccumulator = new MetricsAccumulator();
    try {
      runtimeContext.addAccumulator(ACCUMULATOR_NAME, metricsAccumulator);
    } catch (Exception e) {
      LOG.error("Failed to create metrics accumulator.", e);
    }
  }
  this.metricsAccumulator = (MetricsAccumulator) metricsAccumulator;
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private <V, A extends Serializable> Accumulator<V, A> getAccumulator(String name,
		Class<? extends Accumulator<V, A>> accumulatorClass)
{
	Accumulator<?, ?> accumulator = accumulators.get(name);

	if (accumulator != null) {
		AccumulatorHelper.compareAccumulatorTypes(name, accumulator.getClass(), accumulatorClass);
	} else {
		// Create new accumulator
		try {
			accumulator = accumulatorClass.newInstance();
		}
		catch (Exception e) {
			throw new RuntimeException("Cannot create accumulator " + accumulatorClass.getName());
		}
		accumulators.put(name, accumulator);
	}
	return (Accumulator<V, A>) accumulator;
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
private void testExecuteOnCollection(FlatMapFunction<String, String> udf, List<String> input, boolean mutableSafe) throws Exception {
	ExecutionConfig executionConfig = new ExecutionConfig();
	if (mutableSafe) {
		executionConfig.disableObjectReuse();
	} else {
		executionConfig.enableObjectReuse();
	}
	final TaskInfo taskInfo = new TaskInfo("Test UDF", 4, 0, 4, 0);
	// run on collections
	final List<String> result = getTestFlatMapOperator(udf)
			.executeOnCollections(input,
					new RuntimeUDFContext(
						taskInfo,  null, executionConfig, new HashMap<String, Future<Path>>(),
						new HashMap<String, Accumulator<?, ?>>(), new UnregisteredMetricsGroup()),
					executionConfig);

	Assert.assertEquals(input.size(), result.size());
	Assert.assertEquals(input, result);
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
@SuppressWarnings({"rawtypes", "unchecked"})
@Before
public void setup() {
	joiner = new MockRichFlatJoinFunction();

	baseOperator =
		new OuterJoinOperatorBase(joiner,
			new BinaryOperatorInformation(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO,
				BasicTypeInfo.STRING_TYPE_INFO), new int[0], new int[0], "TestJoiner", null);

	executionConfig = new ExecutionConfig();

	String taskName = "Test rich outer join function";
	TaskInfo taskInfo = new TaskInfo(taskName, 1, 0, 1, 0);
	HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<>();
	HashMap<String, Future<Path>> cpTasks = new HashMap<>();

	runtimeContext = new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks,
		accumulatorMap, new UnregisteredMetricsGroup());
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
@Test
public void testResetBroadcastVariableWithInitializer() {
	try {
		RuntimeUDFContext ctx = new RuntimeUDFContext(
				taskInfo, getClass().getClassLoader(), new ExecutionConfig(),
				new HashMap<String, Future<Path>>(),
				new HashMap<String, Accumulator<?, ?>>(),
				new UnregisteredMetricsGroup());
		
		ctx.setBroadcastVariable("name", Arrays.asList(1, 2, 3, 4));
		
		// access it the first time with an initializer
		List<Double> list = ctx.getBroadcastVariableWithInitializer("name", new ConvertingInitializer());
		assertEquals(Arrays.asList(1.0, 2.0, 3.0, 4.0), list);
		
		// set it again to something different
		ctx.setBroadcastVariable("name", Arrays.asList(2, 3, 4, 5));
		
		List<Double> list2 = ctx.getBroadcastVariableWithInitializer("name", new ConvertingInitializer());
		assertEquals(Arrays.asList(2.0, 3.0, 4.0, 5.0), list2);
	}
	catch (Exception e) {
		e.printStackTrace();
		fail(e.getMessage());
	}
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
/**
 * Deserializes accumulators from a task state update.
 *
 * <p>This method never throws an exception!
 *
 * @param state The task execution state from which to deserialize the accumulators.
 * @return The deserialized accumulators, of null, if there are no accumulators or an error occurred.
 */
private Map<String, Accumulator<?, ?>> deserializeAccumulators(TaskExecutionState state) {
	AccumulatorSnapshot serializedAccumulators = state.getAccumulators();

	if (serializedAccumulators != null) {
		try {
			return serializedAccumulators.deserializeUserAccumulators(userClassLoader);
		}
		catch (Throwable t) {
			// we catch Throwable here to include all form of linking errors that may
			// occur if user classes are missing in the classpath
			LOG.error("Failed to deserialize final accumulator results.", t);
		}
	}
	return null;
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
/**
 * Updates the accumulators during the runtime of a job. Final accumulator results are transferred
 * through the UpdateTaskExecutionState message.
 * @param accumulatorSnapshot The serialized flink and user-defined accumulators
 */
public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot) {
	Map<String, Accumulator<?, ?>> userAccumulators;
	try {
		userAccumulators = accumulatorSnapshot.deserializeUserAccumulators(userClassLoader);

		ExecutionAttemptID execID = accumulatorSnapshot.getExecutionAttemptID();
		Execution execution = currentExecutions.get(execID);
		if (execution != null) {
			execution.setAccumulators(userAccumulators);
		} else {
			LOG.debug("Received accumulator result for unknown execution {}.", execID);
		}
	} catch (Exception e) {
		LOG.error("Cannot update accumulators for job {}.", getJobID(), e);
	}
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的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.Accumulator; //导入依赖的package包/类
@Test
public void stringifyingResultsShouldReportNullLocalValueAsNonnullValueString() {
	final String name = "a";
	final NullBearingAccumulator acc = new NullBearingAccumulator();
	final Map<String, Accumulator<?, ?>> accumulatorMap = new HashMap<>();
	accumulatorMap.put(name, acc);

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

	assertEquals(1, results.length);

	// Note the use of a String with a content of "null" rather than a null value
	final StringifiedAccumulatorResult firstResult = results[0];
	assertEquals(name, firstResult.getName());
	assertEquals("NullBearingAccumulator", firstResult.getType());
	assertEquals("null", firstResult.getValue());
}
 

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

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

	assertEquals(1, results.length);

	// Note the use of String values with content of "null" rather than null values
	final StringifiedAccumulatorResult firstResult = results[0];
	assertEquals(name, firstResult.getName());
	assertEquals("null", firstResult.getType());
	assertEquals("null", firstResult.getValue());
}
 

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

	final ExecutionConfig config = new ExecutionConfig();
	config.registerKryoType(Path.class);

	final AtomicReference<Object> descriptorCapture = new AtomicReference<>();

	StreamingRuntimeContext context = new StreamingRuntimeContext(
			createDescriptorCapturingMockOp(descriptorCapture, config),
			createMockEnvironment(),
			Collections.<String, Accumulator<?, ?>>emptyMap());

	ValueStateDescriptor<TaskInfo> descr = new ValueStateDescriptor<>("name", TaskInfo.class);
	context.getState(descr);

	StateDescriptor<?, ?> descrIntercepted = (StateDescriptor<?, ?>) descriptorCapture.get();
	TypeSerializer<?> serializer = descrIntercepted.getSerializer();

	// check that the Path class is really registered, i.e., the execution config was applied
	assertTrue(serializer instanceof KryoSerializer);
	assertTrue(((KryoSerializer<?>) serializer).getKryo().getRegistration(Path.class).getId() > 0);
}
 

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

	final ExecutionConfig config = new ExecutionConfig();
	config.registerKryoType(Path.class);

	final AtomicReference<Object> descriptorCapture = new AtomicReference<>();

	StreamingRuntimeContext context = new StreamingRuntimeContext(
			createDescriptorCapturingMockOp(descriptorCapture, config),
			createMockEnvironment(),
			Collections.<String, Accumulator<?, ?>>emptyMap());

	ListStateDescriptor<TaskInfo> descr = new ListStateDescriptor<>("name", TaskInfo.class);
	context.getListState(descr);

	ListStateDescriptor<?> descrIntercepted = (ListStateDescriptor<?>) descriptorCapture.get();
	TypeSerializer<?> serializer = descrIntercepted.getSerializer();

	// check that the Path class is really registered, i.e., the execution config was applied
	assertTrue(serializer instanceof ListSerializer);

	TypeSerializer<?> elementSerializer = descrIntercepted.getElementSerializer();
	assertTrue(elementSerializer instanceof KryoSerializer);
	assertTrue(((KryoSerializer<?>) elementSerializer).getKryo().getRegistration(Path.class).getId() > 0);
}
 

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

	final ExecutionConfig config = new ExecutionConfig();
	config.registerKryoType(Path.class);

	final AtomicReference<Object> descriptorCapture = new AtomicReference<>();

	StreamingRuntimeContext context = new StreamingRuntimeContext(
			createDescriptorCapturingMockOp(descriptorCapture, config),
			createMockEnvironment(),
			Collections.<String, Accumulator<?, ?>>emptyMap());

	MapStateDescriptor<String, TaskInfo> descr =
			new MapStateDescriptor<>("name", String.class, TaskInfo.class);

	context.getMapState(descr);

	MapStateDescriptor<?, ?> descrIntercepted = (MapStateDescriptor<?, ?>) descriptorCapture.get();
	TypeSerializer<?> valueSerializer = descrIntercepted.getValueSerializer();

	// check that the Path class is really registered, i.e., the execution config was applied
	assertTrue(valueSerializer instanceof KryoSerializer);
	assertTrue(((KryoSerializer<?>) valueSerializer).getKryo().getRegistration(Path.class).getId() > 0);
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
private static StreamTask<?, ?> createMockTask() {
	Configuration configuration = new Configuration();
	configuration.setString(CoreOptions.STATE_BACKEND, "jobmanager");

	StreamTask<?, ?> task = mock(StreamTask.class);
	when(task.getAccumulatorMap()).thenReturn(new HashMap<String, Accumulator<?, ?>>());
	when(task.getName()).thenReturn("Test task name");
	when(task.getExecutionConfig()).thenReturn(new ExecutionConfig());

	final TaskManagerRuntimeInfo mockTaskManagerRuntimeInfo = mock(TaskManagerRuntimeInfo.class);
	when(mockTaskManagerRuntimeInfo.getConfiguration()).thenReturn(configuration);

	final Environment env = mock(Environment.class);
	when(env.getTaskInfo()).thenReturn(new TaskInfo("Test task name", 1, 0, 1, 0));
	when(env.getUserClassLoader()).thenReturn(AggregatingAlignedProcessingTimeWindowOperatorTest.class.getClassLoader());
	when(env.getMetricGroup()).thenReturn(new UnregisteredTaskMetricsGroup());
	when(env.getTaskManagerInfo()).thenReturn(new TestingTaskManagerRuntimeInfo());

	when(task.getEnvironment()).thenReturn(env);
	return task;
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
@Override
public void merge(Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> other) {
    other.getLocalValue().forEach((k, v) -> {
        if (values.containsKey(k)) {
            values.get(k).add(v.longValue());
        } else {
            values.put(k, v);
        }
    });
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
@Override
public Accumulator<MetricsContainerStepMap, MetricsContainerStepMap> clone() {
  try {
    super.clone();
  } catch (CloneNotSupportedException ignored) {
  }
  MetricsAccumulator metricsAccumulator = new MetricsAccumulator();
  metricsAccumulator.getLocalValue().updateAll(this.getLocalValue());
  return metricsAccumulator;
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
public CollectionExecutor(ExecutionConfig executionConfig) {
	this.executionConfig = executionConfig;
	
	this.intermediateResults = new HashMap<Operator<?>, List<?>>();
	this.accumulators = new HashMap<String, Accumulator<?,?>>();
	this.previousAggregates = new HashMap<String, Value>();
	this.aggregators = new HashMap<String, Aggregator<?>>();
	this.cachedFiles = new HashMap<String, Future<Path>>();
	this.classLoader = getClass().getClassLoader();
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
public AbstractRuntimeUDFContext(TaskInfo taskInfo,
									ClassLoader userCodeClassLoader,
									ExecutionConfig executionConfig,
									Map<String, Accumulator<?,?>> accumulators,
									Map<String, Future<Path>> cpTasks,
									MetricGroup metrics) {
	this.taskInfo = checkNotNull(taskInfo);
	this.userCodeClassLoader = userCodeClassLoader;
	this.executionConfig = executionConfig;
	this.distributedCache = new DistributedCache(checkNotNull(cpTasks));
	this.accumulators = checkNotNull(accumulators);
	this.metrics = metrics;
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
@Override
public <V, A extends Serializable> void addAccumulator(String name, Accumulator<V, A> accumulator) {
	if (accumulators.containsKey(name)) {
		throw new UnsupportedOperationException("The accumulator '" + name
				+ "' already exists and cannot be added.");
	}
	accumulators.put(name, accumulator);
}
 

import org.apache.flink.api.common.accumulators.Accumulator; //导入依赖的package包/类
@Test
public void testDataSourceWithRuntimeContext() {
	try {
		TestRichOutputFormat out = new TestRichOutputFormat();
		GenericDataSinkBase<String> sink = new GenericDataSinkBase<String>(
				out,
				new UnaryOperatorInformation<String, Nothing>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.getInfoFor(Nothing.class)),
				"test_sink");
		sink.setInput(source);

		ExecutionConfig executionConfig = new ExecutionConfig();
		final HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<String, Accumulator<?, ?>>();
		final HashMap<String, Future<Path>> cpTasks = new HashMap<>();
		final TaskInfo taskInfo = new TaskInfo("test_sink", 1, 0, 1, 0);
		executionConfig.disableObjectReuse();
		in.reset();
		
		sink.executeOnCollections(asList(TestIOData.NAMES), new RuntimeUDFContext(
				taskInfo, null, executionConfig, cpTasks, accumulatorMap, new UnregisteredMetricsGroup()),
				executionConfig);
	
			assertEquals(out.output, asList(TestIOData.RICH_NAMES));

		executionConfig.enableObjectReuse();
		out.clear();
		in.reset();
		
		sink.executeOnCollections(asList(TestIOData.NAMES), new RuntimeUDFContext(
				taskInfo, null, executionConfig, cpTasks, accumulatorMap, new UnregisteredMetricsGroup()),
				executionConfig);
		assertEquals(out.output, asList(TestIOData.RICH_NAMES));
	} catch(Exception e){
		e.printStackTrace();
		fail(e.getMessage());
	}
}