org.apache.flink.runtime.checkpoint

Class StateAssignmentOperation

java.lang.Object

org.apache.flink.runtime.checkpoint.StateAssignmentOperation

@Internal
public class StateAssignmentOperation
extends Object

This class encapsulates the operation of assigning restored state when restoring from a checkpoint.

Constructor Summary

Constructors

Constructor and Description
StateAssignmentOperation(long restoreCheckpointId, Set<ExecutionJobVertex> tasks, Map<OperatorID,OperatorState> operatorStates, boolean allowNonRestoredState)

Method Summary

Modifier and Type	Method and Description
static <T> Map<OperatorInstanceID,List<T>>	applyRepartitioner(OperatorID operatorID, OperatorStateRepartitioner<T> opStateRepartitioner, List<List<T>> chainOpParallelStates, int oldParallelism, int newParallelism)
static <T> List<List<T>>	applyRepartitioner(OperatorStateRepartitioner<T> opStateRepartitioner, List<List<T>> chainOpParallelStates, int oldParallelism, int newParallelism) Repartitions the given operator state using the given OperatorStateRepartitioner with respect to the new parallelism.
void	assignStates()
void	checkParallelismPreconditions(org.apache.flink.runtime.checkpoint.TaskStateAssignment taskStateAssignment)
static List<KeyGroupRange>	createKeyGroupPartitions(int numberKeyGroups, int parallelism) Groups the available set of key groups into key group partitions.
static void	extractIntersectingState(Collection<? extends KeyedStateHandle> originalSubtaskStateHandles, KeyGroupRange rangeToExtract, List<KeyedStateHandle> extractedStateCollector) Extracts certain key group ranges from the given state handles and adds them to the collector.
static List<KeyedStateHandle>	getManagedKeyedStateHandles(OperatorState operatorState, KeyGroupRange subtaskKeyGroupRange) Collect managedKeyedStateHandles which have intersection with given KeyGroupRange from operatorState.
static List<KeyedStateHandle>	getRawKeyedStateHandles(OperatorState operatorState, KeyGroupRange subtaskKeyGroupRange) Collect rawKeyedStateHandles which have intersection with given KeyGroupRange from operatorState.
void	reDistributeInputChannelStates(org.apache.flink.runtime.checkpoint.TaskStateAssignment stateAssignment)
static <T extends StateObject> void	reDistributePartitionableStates(Map<OperatorID,OperatorState> oldOperatorStates, int newParallelism, Function<OperatorSubtaskState,StateObjectCollection<T>> extractHandle, OperatorStateRepartitioner<T> stateRepartitioner, Map<OperatorInstanceID,List<T>> result)
void	reDistributeResultSubpartitionStates(org.apache.flink.runtime.checkpoint.TaskStateAssignment assignment)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

StateAssignmentOperation

public StateAssignmentOperation(long restoreCheckpointId,
                                Set<ExecutionJobVertex> tasks,
                                Map<OperatorID,OperatorState> operatorStates,
                                boolean allowNonRestoredState)

Method Detail

assignStates

public void assignStates()

checkParallelismPreconditions

public void checkParallelismPreconditions(org.apache.flink.runtime.checkpoint.TaskStateAssignment taskStateAssignment)

reDistributePartitionableStates

public static <T extends StateObject> void reDistributePartitionableStates(Map<OperatorID,OperatorState> oldOperatorStates,
                                                                           int newParallelism,
                                                                           Function<OperatorSubtaskState,StateObjectCollection<T>> extractHandle,
                                                                           OperatorStateRepartitioner<T> stateRepartitioner,
                                                                           Map<OperatorInstanceID,List<T>> result)

reDistributeResultSubpartitionStates

public void reDistributeResultSubpartitionStates(org.apache.flink.runtime.checkpoint.TaskStateAssignment assignment)

reDistributeInputChannelStates

public void reDistributeInputChannelStates(org.apache.flink.runtime.checkpoint.TaskStateAssignment stateAssignment)

getManagedKeyedStateHandles

public static List<KeyedStateHandle> getManagedKeyedStateHandles(OperatorState operatorState,
                                                                 KeyGroupRange subtaskKeyGroupRange)

Collect managedKeyedStateHandles which have intersection with given KeyGroupRange from operatorState.

Parameters:

operatorState - all state handles of a operator

subtaskKeyGroupRange - the KeyGroupRange of a subtask

Returns:

all managedKeyedStateHandles which have intersection with given KeyGroupRange

getRawKeyedStateHandles

public static List<KeyedStateHandle> getRawKeyedStateHandles(OperatorState operatorState,
                                                             KeyGroupRange subtaskKeyGroupRange)

Collect rawKeyedStateHandles which have intersection with given KeyGroupRange from operatorState.

Parameters:

operatorState - all state handles of a operator

subtaskKeyGroupRange - the KeyGroupRange of a subtask

Returns:

all rawKeyedStateHandles which have intersection with given KeyGroupRange

extractIntersectingState

@VisibleForTesting
public static void extractIntersectingState(Collection<? extends KeyedStateHandle> originalSubtaskStateHandles,
                                                               KeyGroupRange rangeToExtract,
                                                               List<KeyedStateHandle> extractedStateCollector)

Extracts certain key group ranges from the given state handles and adds them to the collector.

createKeyGroupPartitions

public static List<KeyGroupRange> createKeyGroupPartitions(int numberKeyGroups,
                                                           int parallelism)

Groups the available set of key groups into key group partitions. A key group partition is the set of key groups which is assigned to the same task. Each set of the returned list constitutes a key group partition.

IMPORTANT: The assignment of key groups to partitions has to be in sync with the KeyGroupStreamPartitioner.

Parameters:

numberKeyGroups - Number of available key groups (indexed from 0 to numberKeyGroups - 1)

parallelism - Parallelism to generate the key group partitioning for

Returns:

List of key group partitions

applyRepartitioner

public static <T> Map<OperatorInstanceID,List<T>> applyRepartitioner(OperatorID operatorID,
                                                                     OperatorStateRepartitioner<T> opStateRepartitioner,
                                                                     List<List<T>> chainOpParallelStates,
                                                                     int oldParallelism,
                                                                     int newParallelism)

applyRepartitioner

public static <T> List<List<T>> applyRepartitioner(OperatorStateRepartitioner<T> opStateRepartitioner,
                                                   List<List<T>> chainOpParallelStates,
                                                   int oldParallelism,
                                                   int newParallelism)

Repartitions the given operator state using the given OperatorStateRepartitioner with respect to the new parallelism.

Parameters:

opStateRepartitioner - partitioner to use

chainOpParallelStates - state to repartition

oldParallelism - parallelism with which the state is currently partitioned

newParallelism - parallelism with which the state should be partitioned

Returns:

repartitioned state