flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateTable.java
/**
	 * @param key            the key of the mapping to remove. Not null.
	 * @param namespace      the namespace of the mapping to remove. Not null.
	 * @param value          the value that is the second input for the transformation.
	 * @param transformation the transformation function to apply on the old state and the given value.
	 * @param <T>            type of the value that is the second input to the {@link StateTransformationFunction}.
	 * @throws Exception exception that happen on applying the function.
	 * @see #transform(Object, Object, StateTransformationFunction).
	 */
@param key            the key of the mapping to remove. Not null.
@param namespace      the namespace of the mapping to remove. Not null.
@param value          the value that is the second input for the transformation.
@param transformation the transformation function to apply on the old state and the given value.
@param             type of the value that is the second input to the  StateTransformationFunction.
@throws Exception exception that happen on applying the function.
@param key the key of the mapping to remove . Not null . @param namespace the namespace of the mapping to remove . Not null . @param value the value that is the second input for the transformation . @param transformation the transformation function to apply on the old state and the given value . @param type of the value that is the second input to the StateTransformationFunction . @throws Exception exception that happen on applying the function .
<T> void transform(
			K key,
			N namespace,
			T value,
			StateTransformationFunction<S, T> transformation) throws Exception {
		final StateTableEntry<K, N, S> entry = putEntry(key, namespace);
		entry.state = transformation.apply(
				(entry.stateVersion < highestRequiredSnapshotVersion) ?
						getStateSerializer().copy(entry.state) :
						entry.state,
				value);
		entry.stateVersion = stateTableVersion;
	}
< T > void transform ( K key , N namespace , T value , StateTransformationFunction < S , T > transformation ) throws Exception { final StateTableEntry < K , N , S > entry = putEntry ( key , namespace ) ; entry . state = transformation . apply ( ( entry . stateVersion < highestRequiredSnapshotVersion ) ? getStateSerializer ( ) . copy ( entry . state ) : entry . state , value ) ; entry . stateVersion = stateTableVersion ; }
@param key            the key of the mapping to remove.
@param key the key of the mapping to remove .
// copy-on-write check for state