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 TaskState objects containing various types of state handles.
	 */
Creates a random collection of TaskState objects containing various types of state handles.
Creates a random collection of TaskState objects containing various types of state handles .
public static Collection<TaskState> createTaskStates(
			Random random,
			int numTaskStates,
			int numSubtasksPerTask) {
		List<TaskState> taskStates = new ArrayList<>(numTaskStates);
		for (int stateIdx = 0; stateIdx < numTaskStates; ++stateIdx) {
			int chainLength = 1 + random.nextInt(8);
			TaskState taskState = new TaskState(new JobVertexID(), numSubtasksPerTask, 128, chainLength);
			int noNonPartitionableStateAtIndex = random.nextInt(chainLength);
			int noOperatorStateBackendAtIndex = random.nextInt(chainLength);
			int noOperatorStateStreamAtIndex = random.nextInt(chainLength);
			boolean hasKeyedBackend = random.nextInt(4) != 0;
			boolean hasKeyedStream = random.nextInt(4) != 0;
			for (int subtaskIdx = 0; subtaskIdx < numSubtasksPerTask; subtaskIdx++) {
				List<OperatorStateHandle> operatorStatesBackend = new ArrayList<>(chainLength);
				List<OperatorStateHandle> operatorStatesStream = new ArrayList<>(chainLength);
				for (int chainIdx = 0; chainIdx < chainLength; ++chainIdx) {
					StreamStateHandle operatorStateBackend =
							new TestByteStreamStateHandleDeepCompare("b-" + chainIdx, ("Beautiful-" + chainIdx).getBytes(ConfigConstants.DEFAULT_CHARSET));
					StreamStateHandle operatorStateStream =
							new TestByteStreamStateHandleDeepCompare("b-" + chainIdx, ("Beautiful-" + chainIdx).getBytes(ConfigConstants.DEFAULT_CHARSET));
					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 (chainIdx != noOperatorStateBackendAtIndex) {
						OperatorStateHandle operatorStateHandleBackend =
								new OperatorStateHandle(offsetsMap, operatorStateBackend);
						operatorStatesBackend.add(operatorStateHandleBackend);
					}
					if (chainIdx != noOperatorStateStreamAtIndex) {
						OperatorStateHandle operatorStateHandleStream =
								new OperatorStateHandle(offsetsMap, operatorStateStream);
						operatorStatesStream.add(operatorStateHandleStream);
					}
				}
				KeyGroupsStateHandle keyedStateBackend = null;
				KeyGroupsStateHandle keyedStateStream = null;
				if (hasKeyedBackend) {
					keyedStateBackend = createDummyKeyGroupStateHandle(random);
				}
				if (hasKeyedStream) {
					keyedStateStream = createDummyKeyGroupStateHandle(random);
				}
				taskState.putState(subtaskIdx, new SubtaskState(
						new ChainedStateHandle<>(operatorStatesBackend),
						new ChainedStateHandle<>(operatorStatesStream),
						keyedStateStream,
						keyedStateBackend));
			}
			taskStates.add(taskState);
		}
		return taskStates;
	}
public static Collection < TaskState > createTaskStates ( Random random , int numTaskStates , int numSubtasksPerTask ) { List < TaskState > taskStates = new ArrayList < > ( numTaskStates ) ; for ( int stateIdx = 0 ; stateIdx < numTaskStates ; ++ stateIdx ) { int chainLength = 1 + random . nextInt ( 8 ) ; TaskState taskState = new TaskState ( new JobVertexID ( ) , numSubtasksPerTask , 128 , chainLength ) ; int noNonPartitionableStateAtIndex = random . nextInt ( chainLength ) ; int noOperatorStateBackendAtIndex = random . nextInt ( chainLength ) ; int noOperatorStateStreamAtIndex = random . nextInt ( chainLength ) ; boolean hasKeyedBackend = random . nextInt ( 4 ) != 0 ; boolean hasKeyedStream = random . nextInt ( 4 ) != 0 ; for ( int subtaskIdx = 0 ; subtaskIdx < numSubtasksPerTask ; subtaskIdx ++ ) { List < OperatorStateHandle > operatorStatesBackend = new ArrayList < > ( chainLength ) ; List < OperatorStateHandle > operatorStatesStream = new ArrayList < > ( chainLength ) ; for ( int chainIdx = 0 ; chainIdx < chainLength ; ++ chainIdx ) { StreamStateHandle operatorStateBackend = new TestByteStreamStateHandleDeepCompare ( "b-" + chainIdx , ( "Beautiful-" + chainIdx ) . getBytes ( ConfigConstants . DEFAULT_CHARSET ) ) ; StreamStateHandle operatorStateStream = new TestByteStreamStateHandleDeepCompare ( "b-" + chainIdx , ( "Beautiful-" + chainIdx ) . getBytes ( ConfigConstants . DEFAULT_CHARSET ) ) ; 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 ( chainIdx != noOperatorStateBackendAtIndex ) { OperatorStateHandle operatorStateHandleBackend = new OperatorStateHandle ( offsetsMap , operatorStateBackend ) ; operatorStatesBackend . add ( operatorStateHandleBackend ) ; } if ( chainIdx != noOperatorStateStreamAtIndex ) { OperatorStateHandle operatorStateHandleStream = new OperatorStateHandle ( offsetsMap , operatorStateStream ) ; operatorStatesStream . add ( operatorStateHandleStream ) ; } } KeyGroupsStateHandle keyedStateBackend = null ; KeyGroupsStateHandle keyedStateStream = null ; if ( hasKeyedBackend ) { keyedStateBackend = createDummyKeyGroupStateHandle ( random ) ; } if ( hasKeyedStream ) { keyedStateStream = createDummyKeyGroupStateHandle ( random ) ; } taskState . putState ( subtaskIdx , new SubtaskState ( new ChainedStateHandle < > ( operatorStatesBackend ) , new ChainedStateHandle < > ( operatorStatesStream ) , keyedStateStream , keyedStateBackend ) ) ; } taskStates . add ( taskState ) ; } return taskStates ; }
Creates a random collection of TaskState objects containing various types of state handles.
Creates a random collection of TaskState objects containing various types of state handles .