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
/**
	 * Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#initializeState(OperatorSubtaskState)}.
	 * Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#setup(StreamTask, StreamConfig, Output)}
	 * if it was not called before.
	 *
	 * <p>This will reshape the state handles to include only those key-group states
	 * in the local key-group range and the operator states that would be assigned to the local
	 * subtask.
	 */


This will reshape the state handles to include only those key-group states
in the local key-group range and the operator states that would be assigned to the local
subtask.
This will reshape the state handles to include only those key - group states in the local key - group range and the operator states that would be assigned to the local subtask .
public void initializeState(OperatorStateHandles operatorStateHandles) throws Exception {
		if (!setupCalled) {
			setup();
		}
		if (operatorStateHandles != null) {
			int numKeyGroups = getEnvironment().getTaskInfo().getMaxNumberOfParallelSubtasks();
			int numSubtasks = getEnvironment().getTaskInfo().getNumberOfParallelSubtasks();
			int subtaskIndex = getEnvironment().getTaskInfo().getIndexOfThisSubtask();
			List<KeyGroupRange> keyGroupPartitions = StateAssignmentOperation.createKeyGroupPartitions(
					numKeyGroups,
					numSubtasks);
			KeyGroupRange localKeyGroupRange =
					keyGroupPartitions.get(subtaskIndex);
			List<KeyedStateHandle> localManagedKeyGroupState = null;
			if (operatorStateHandles.getManagedKeyedState() != null) {
				localManagedKeyGroupState = StateAssignmentOperation.getKeyedStateHandles(
						operatorStateHandles.getManagedKeyedState(),
						localKeyGroupRange);
			}
			List<KeyedStateHandle> localRawKeyGroupState = null;
			if (operatorStateHandles.getRawKeyedState() != null) {
				localRawKeyGroupState = StateAssignmentOperation.getKeyedStateHandles(
						operatorStateHandles.getRawKeyedState(),
						localKeyGroupRange);
			}
			List<OperatorStateHandle> managedOperatorState = new ArrayList<>();
			if (operatorStateHandles.getManagedOperatorState() != null) {
				managedOperatorState.addAll(operatorStateHandles.getManagedOperatorState());
			}
			Collection<OperatorStateHandle> localManagedOperatorState = operatorStateRepartitioner.repartitionState(
					managedOperatorState,
					numSubtasks).get(subtaskIndex);
			List<OperatorStateHandle> rawOperatorState = new ArrayList<>();
			if (operatorStateHandles.getRawOperatorState() != null) {
				rawOperatorState.addAll(operatorStateHandles.getRawOperatorState());
			}
			Collection<OperatorStateHandle> localRawOperatorState = operatorStateRepartitioner.repartitionState(
					rawOperatorState,
					numSubtasks).get(subtaskIndex);
			OperatorSubtaskState massagedOperatorStateHandles = new OperatorSubtaskState(
				nullToEmptyCollection(localManagedOperatorState),
				nullToEmptyCollection(localRawOperatorState),
				nullToEmptyCollection(localManagedKeyGroupState),
				nullToEmptyCollection(localRawKeyGroupState));
			operator.initializeState(massagedOperatorStateHandles);
		} else {
			operator.initializeState(null);
		}
		initializeCalled = true;
	}
public void initializeState ( OperatorStateHandles operatorStateHandles ) throws Exception { if ( ! setupCalled ) { setup ( ) ; } if ( operatorStateHandles != null ) { int numKeyGroups = getEnvironment ( ) . getTaskInfo ( ) . getMaxNumberOfParallelSubtasks ( ) ; int numSubtasks = getEnvironment ( ) . getTaskInfo ( ) . getNumberOfParallelSubtasks ( ) ; int subtaskIndex = getEnvironment ( ) . getTaskInfo ( ) . getIndexOfThisSubtask ( ) ; List < KeyGroupRange > keyGroupPartitions = StateAssignmentOperation . createKeyGroupPartitions ( numKeyGroups , numSubtasks ) ; KeyGroupRange localKeyGroupRange = keyGroupPartitions . get ( subtaskIndex ) ; List < KeyedStateHandle > localManagedKeyGroupState = null ; if ( operatorStateHandles . getManagedKeyedState ( ) != null ) { localManagedKeyGroupState = StateAssignmentOperation . getKeyedStateHandles ( operatorStateHandles . getManagedKeyedState ( ) , localKeyGroupRange ) ; } List < KeyedStateHandle > localRawKeyGroupState = null ; if ( operatorStateHandles . getRawKeyedState ( ) != null ) { localRawKeyGroupState = StateAssignmentOperation . getKeyedStateHandles ( operatorStateHandles . getRawKeyedState ( ) , localKeyGroupRange ) ; } List < OperatorStateHandle > managedOperatorState = new ArrayList < > ( ) ; if ( operatorStateHandles . getManagedOperatorState ( ) != null ) { managedOperatorState . addAll ( operatorStateHandles . getManagedOperatorState ( ) ) ; } Collection < OperatorStateHandle > localManagedOperatorState = operatorStateRepartitioner . repartitionState ( managedOperatorState , numSubtasks ) . get ( subtaskIndex ) ; List < OperatorStateHandle > rawOperatorState = new ArrayList < > ( ) ; if ( operatorStateHandles . getRawOperatorState ( ) != null ) { rawOperatorState . addAll ( operatorStateHandles . getRawOperatorState ( ) ) ; } Collection < OperatorStateHandle > localRawOperatorState = operatorStateRepartitioner . repartitionState ( rawOperatorState , numSubtasks ) . get ( subtaskIndex ) ; OperatorSubtaskState massagedOperatorStateHandles = new OperatorSubtaskState ( nullToEmptyCollection ( localManagedOperatorState ) , nullToEmptyCollection ( localRawOperatorState ) , nullToEmptyCollection ( localManagedKeyGroupState ) , nullToEmptyCollection ( localRawKeyGroupState ) ) ; operator . initializeState ( massagedOperatorStateHandles ) ; } else { operator . initializeState ( null ) ; } initializeCalled = true ; }
Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#initializeState(OperatorSubtaskState)}.
Calls { @link org . apache . flink . streaming . api . operators . StreamOperator#initializeState ( OperatorSubtaskState ) } .
// create a new OperatorStateHandles that only contains the state for our key-groups