flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/CheckpointTestUtils.java
/**
	 * Creates a random collection of OperatorState objects containing various types of state handles.
	 */
Creates a random collection of OperatorState objects containing various types of state handles.
Creates a random collection of OperatorState objects containing various types of state handles .
public static Collection<OperatorState> createOperatorStates(
			Random random,
			int numTaskStates,
			int numSubtasksPerTask) {
		List<OperatorState> taskStates = new ArrayList<>(numTaskStates);
		for (int stateIdx = 0; stateIdx < numTaskStates; ++stateIdx) {
			OperatorState taskState = new OperatorState(new OperatorID(), numSubtasksPerTask, 128);
			boolean hasOperatorStateBackend = random.nextBoolean();
			boolean hasOperatorStateStream = random.nextBoolean();
			boolean hasKeyedBackend = random.nextInt(4) != 0;
			boolean hasKeyedStream = random.nextInt(4) != 0;
			boolean isIncremental = random.nextInt(3) == 0;
			for (int subtaskIdx = 0; subtaskIdx < numSubtasksPerTask; subtaskIdx++) {
				StreamStateHandle operatorStateBackend =
					new TestByteStreamStateHandleDeepCompare("b", ("Beautiful").getBytes(ConfigConstants.DEFAULT_CHARSET));
				StreamStateHandle operatorStateStream =
					new TestByteStreamStateHandleDeepCompare("b", ("Beautiful").getBytes(ConfigConstants.DEFAULT_CHARSET));
				OperatorStateHandle operatorStateHandleBackend = null;
				OperatorStateHandle operatorStateHandleStream = null;
				Map<String, StateMetaInfo> offsetsMap = new HashMap<>();
				offsetsMap.put("A", new OperatorStateHandle.StateMetaInfo(new long[]{0, 10, 20}, OperatorStateHandle.Mode.SPLIT_DISTRIBUTE));
				offsetsMap.put("B", new OperatorStateHandle.StateMetaInfo(new long[]{30, 40, 50}, OperatorStateHandle.Mode.SPLIT_DISTRIBUTE));
				offsetsMap.put("C", new OperatorStateHandle.StateMetaInfo(new long[]{60, 70, 80}, OperatorStateHandle.Mode.BROADCAST));
				if (hasOperatorStateBackend) {
					operatorStateHandleBackend = new OperatorStateHandle(offsetsMap, operatorStateBackend);
				}
				if (hasOperatorStateStream) {
					operatorStateHandleStream = new OperatorStateHandle(offsetsMap, operatorStateStream);
				}
				KeyedStateHandle keyedStateBackend = null;
				KeyedStateHandle keyedStateStream = null;
				if (hasKeyedBackend) {
					if (isIncremental) {
						keyedStateBackend = createDummyIncrementalKeyedStateHandle(random);
					} else {
						keyedStateBackend = createDummyKeyGroupStateHandle(random);
					}
				}
				if (hasKeyedStream) {
					keyedStateStream = createDummyKeyGroupStateHandle(random);
				}
				taskState.putState(subtaskIdx, new OperatorSubtaskState(
						operatorStateHandleBackend,
						operatorStateHandleStream,
						keyedStateStream,
						keyedStateBackend));
			}
			taskStates.add(taskState);
		}
		return taskStates;
	}
public static Collection < OperatorState > createOperatorStates ( Random random , int numTaskStates , int numSubtasksPerTask ) { List < OperatorState > taskStates = new ArrayList < > ( numTaskStates ) ; for ( int stateIdx = 0 ; stateIdx < numTaskStates ; ++ stateIdx ) { OperatorState taskState = new OperatorState ( new OperatorID ( ) , numSubtasksPerTask , 128 ) ; boolean hasOperatorStateBackend = random . nextBoolean ( ) ; boolean hasOperatorStateStream = random . nextBoolean ( ) ; boolean hasKeyedBackend = random . nextInt ( 4 ) != 0 ; boolean hasKeyedStream = random . nextInt ( 4 ) != 0 ; boolean isIncremental = random . nextInt ( 3 ) == 0 ; for ( int subtaskIdx = 0 ; subtaskIdx < numSubtasksPerTask ; subtaskIdx ++ ) { StreamStateHandle operatorStateBackend = new TestByteStreamStateHandleDeepCompare ( "b" , ( "Beautiful" ) . getBytes ( ConfigConstants . DEFAULT_CHARSET ) ) ; StreamStateHandle operatorStateStream = new TestByteStreamStateHandleDeepCompare ( "b" , ( "Beautiful" ) . getBytes ( ConfigConstants . DEFAULT_CHARSET ) ) ; OperatorStateHandle operatorStateHandleBackend = null ; OperatorStateHandle operatorStateHandleStream = null ; Map < String , StateMetaInfo > offsetsMap = new HashMap < > ( ) ; offsetsMap . put ( "A" , new OperatorStateHandle . StateMetaInfo ( new long [ ] { 0 , 10 , 20 } , OperatorStateHandle . Mode . SPLIT_DISTRIBUTE ) ) ; offsetsMap . put ( "B" , new OperatorStateHandle . StateMetaInfo ( new long [ ] { 30 , 40 , 50 } , OperatorStateHandle . Mode . SPLIT_DISTRIBUTE ) ) ; offsetsMap . put ( "C" , new OperatorStateHandle . StateMetaInfo ( new long [ ] { 60 , 70 , 80 } , OperatorStateHandle . Mode . BROADCAST ) ) ; if ( hasOperatorStateBackend ) { operatorStateHandleBackend = new OperatorStateHandle ( offsetsMap , operatorStateBackend ) ; } if ( hasOperatorStateStream ) { operatorStateHandleStream = new OperatorStateHandle ( offsetsMap , operatorStateStream ) ; } KeyedStateHandle keyedStateBackend = null ; KeyedStateHandle keyedStateStream = null ; if ( hasKeyedBackend ) { if ( isIncremental ) { keyedStateBackend = createDummyIncrementalKeyedStateHandle ( random ) ; } else { keyedStateBackend = createDummyKeyGroupStateHandle ( random ) ; } } if ( hasKeyedStream ) { keyedStateStream = createDummyKeyGroupStateHandle ( random ) ; } taskState . putState ( subtaskIdx , new OperatorSubtaskState ( operatorStateHandleBackend , operatorStateHandleStream , keyedStateStream , keyedStateBackend ) ) ; } taskStates . add ( taskState ) ; } return taskStates ; }
Creates a random collection of OperatorState objects containing various types of state handles.
Creates a random collection of OperatorState objects containing various types of state handles .