flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
/**
	 * Takes the different {@link OperatorStateHandles} created by calling {@link #snapshot(long, long)}
	 * on different instances of {@link AbstractStreamOperatorTestHarness} (each one representing one subtask)
	 * and repacks them into a single {@link OperatorStateHandles} so that the parallelism of the test
	 * can change arbitrarily (i.e. be able to scale both up and down).
	 *
	 * <p>After repacking the partial states, use {@link #initializeState(OperatorStateHandles)} to initialize
	 * a new instance with the resulting state. Bear in mind that for parallelism greater than one, you
	 * have to use the constructor {@link #AbstractStreamOperatorTestHarness(StreamOperator, int, int, int)}.
	 *
	 * <p><b>NOTE: </b> each of the {@code handles} in the argument list is assumed to be from a single task of a single
	 * operator (i.e. chain length of one).
	 *
	 * <p>For an example of how to use it, have a look at
	 * {@link AbstractStreamOperatorTest#testStateAndTimerStateShufflingScalingDown()}.
	 *
	 * @param handles the different states to be merged.
	 * @return the resulting state, or {@code null} if no partial states are specified.
	 */
Takes the different  OperatorStateHandles created by calling  #snapshot(long, long)
on different instances of  AbstractStreamOperatorTestHarness (each one representing one subtask)
and repacks them into a single  OperatorStateHandles so that the parallelism of the test
can change arbitrarily .

After repacking the partial states, use  #initializeState(OperatorStateHandles) to initialize
a new instance with the resulting state. Bear in mind that for parallelism greater than one, you
have to use the constructor  #AbstractStreamOperatorTestHarness(StreamOperator, int, int, int).

NOTE:  each of the  handles in the argument list is assumed to be from a single task of a single
operator .

For an example of how to use it, have a look at
AbstractStreamOperatorTest#testStateAndTimerStateShufflingScalingDown().

@param handles the different states to be merged.
@return the resulting state, or  null if no partial states are specified.
Takes the different OperatorStateHandles created by calling #snapshot ( long long ) on different instances of AbstractStreamOperatorTestHarness ( each one representing one subtask ) and repacks them into a single OperatorStateHandles so that the parallelism of the test can change arbitrarily . After repacking the partial states use #initializeState ( OperatorStateHandles ) to initialize a new instance with the resulting state . Bear in mind that for parallelism greater than one you have to use the constructor #AbstractStreamOperatorTestHarness ( StreamOperator int int int ) . NOTE : each of the handles in the argument list is assumed to be from a single task of a single operator . For an example of how to use it have a look at AbstractStreamOperatorTest#testStateAndTimerStateShufflingScalingDown () . @param handles the different states to be merged . @return the resulting state or null if no partial states are specified .
public static OperatorStateHandles repackageState(OperatorStateHandles... handles) throws Exception {
		if (handles.length < 1) {
			return null;
		} else if (handles.length == 1) {
			return handles[0];
		}
		List<OperatorStateHandle> mergedManagedOperatorState = new ArrayList<>(handles.length);
		List<OperatorStateHandle> mergedRawOperatorState = new ArrayList<>(handles.length);
		List<KeyedStateHandle> mergedManagedKeyedState = new ArrayList<>(handles.length);
		List<KeyedStateHandle> mergedRawKeyedState = new ArrayList<>(handles.length);
		for (OperatorStateHandles handle: handles) {
			Collection<OperatorStateHandle> managedOperatorState = handle.getManagedOperatorState();
			Collection<OperatorStateHandle> rawOperatorState = handle.getRawOperatorState();
			Collection<KeyedStateHandle> managedKeyedState = handle.getManagedKeyedState();
			Collection<KeyedStateHandle> rawKeyedState = handle.getRawKeyedState();
			if (managedOperatorState != null) {
				mergedManagedOperatorState.addAll(managedOperatorState);
			}
			if (rawOperatorState != null) {
				mergedRawOperatorState.addAll(rawOperatorState);
			}
			if (managedKeyedState != null) {
				mergedManagedKeyedState.addAll(managedKeyedState);
			}
			if (rawKeyedState != null) {
				mergedRawKeyedState.addAll(rawKeyedState);
			}
		}
		return new OperatorStateHandles(
			0,
			mergedManagedKeyedState,
			mergedRawKeyedState,
			mergedManagedOperatorState,
			mergedRawOperatorState);
	}
public static OperatorStateHandles repackageState ( OperatorStateHandles ... handles ) throws Exception { if ( handles . length < 1 ) { return null ; } else if ( handles . length == 1 ) { return handles [ 0 ] ; } List < OperatorStateHandle > mergedManagedOperatorState = new ArrayList < > ( handles . length ) ; List < OperatorStateHandle > mergedRawOperatorState = new ArrayList < > ( handles . length ) ; List < KeyedStateHandle > mergedManagedKeyedState = new ArrayList < > ( handles . length ) ; List < KeyedStateHandle > mergedRawKeyedState = new ArrayList < > ( handles . length ) ; for ( OperatorStateHandles handle : handles ) { Collection < OperatorStateHandle > managedOperatorState = handle . getManagedOperatorState ( ) ; Collection < OperatorStateHandle > rawOperatorState = handle . getRawOperatorState ( ) ; Collection < KeyedStateHandle > managedKeyedState = handle . getManagedKeyedState ( ) ; Collection < KeyedStateHandle > rawKeyedState = handle . getRawKeyedState ( ) ; if ( managedOperatorState != null ) { mergedManagedOperatorState . addAll ( managedOperatorState ) ; } if ( rawOperatorState != null ) { mergedRawOperatorState . addAll ( rawOperatorState ) ; } if ( managedKeyedState != null ) { mergedManagedKeyedState . addAll ( managedKeyedState ) ; } if ( rawKeyedState != null ) { mergedRawKeyedState . addAll ( rawKeyedState ) ; } } return new OperatorStateHandles ( 0 , mergedManagedKeyedState , mergedRawKeyedState , mergedManagedOperatorState , mergedRawOperatorState ) ; }
Takes the different {@link OperatorStateHandles} created by calling {@link #snapshot(long, long)}
 on different instances of {@link AbstractStreamOperatorTestHarness} (each one representing one subtask)
 and repacks them into a single {@link OperatorStateHandles} so that the parallelism of the test
 can change arbitrarily (i.e.
Takes the different { @link OperatorStateHandles } created by calling { @link #snapshot ( long long ) } on different instances of { @link AbstractStreamOperatorTestHarness } ( each one representing one subtask ) and repacks them into a single { @link OperatorStateHandles } so that the parallelism of the test can change arbitrarily ( i . e .