序

本文主要研究一下flink的KvStateRegistryGateway

KvStateRegistryGateway

flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/KvStateRegistryGateway.java

public interface KvStateRegistryGateway {/*** Notifies that queryable state has been registered.** @param jobId    identifying the job for which to register a key value state* @param jobVertexId JobVertexID the KvState instance belongs to.* @param keyGroupRange Key group range the KvState instance belongs to.* @param registrationName Name under which the KvState has been registered.* @param kvStateId ID of the registered KvState instance.* @param kvStateServerAddress Server address where to find the KvState instance.* @return Future acknowledge if the key-value state has been registered*/CompletableFuture<Acknowledge> notifyKvStateRegistered(final JobID jobId,final JobVertexID jobVertexId,final KeyGroupRange keyGroupRange,final String registrationName,final KvStateID kvStateId,final InetSocketAddress kvStateServerAddress);/*** Notifies that queryable state has been unregistered.** @param jobId    identifying the job for which to unregister a key value state* @param jobVertexId JobVertexID the KvState instance belongs to.* @param keyGroupRange Key group index the KvState instance belongs to.* @param registrationName Name under which the KvState has been registered.* @return Future acknowledge if the key-value state has been unregistered*/CompletableFuture<Acknowledge> notifyKvStateUnregistered(final JobID jobId,final JobVertexID jobVertexId,final KeyGroupRange keyGroupRange,final String registrationName);
}

• KvStateRegistryGateway接口定义了notifyKvStateRegistered、notifyKvStateUnregistered两个方法；JobMaster实现了这两个方法

JobMaster

flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java

public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway {/** Default names for Flink's distributed components. */public static final String JOB_MANAGER_NAME = "jobmanager";public static final String ARCHIVE_NAME = "archive";// ------------------------------------------------------------------------private final JobMasterConfiguration jobMasterConfiguration;private final ResourceID resourceId;private final JobGraph jobGraph;private final Time rpcTimeout;private final HighAvailabilityServices highAvailabilityServices;private final BlobServer blobServer;private final JobManagerJobMetricGroupFactory jobMetricGroupFactory;private final HeartbeatManager<AccumulatorReport, Void> taskManagerHeartbeatManager;private final HeartbeatManager<Void, Void> resourceManagerHeartbeatManager;private final ScheduledExecutorService scheduledExecutorService;private final OnCompletionActions jobCompletionActions;private final FatalErrorHandler fatalErrorHandler;private final ClassLoader userCodeLoader;private final SlotPool slotPool;private final SlotPoolGateway slotPoolGateway;private final RestartStrategy restartStrategy;// --------- BackPressure --------private final BackPressureStatsTracker backPressureStatsTracker;// --------- ResourceManager --------private final LeaderRetrievalService resourceManagerLeaderRetriever;// --------- TaskManagers --------private final Map<ResourceID, Tuple2<TaskManagerLocation, TaskExecutorGateway>> registeredTaskManagers;// -------- Mutable fields ---------private ExecutionGraph executionGraph;@Nullableprivate JobManagerJobStatusListener jobStatusListener;@Nullableprivate JobManagerJobMetricGroup jobManagerJobMetricGroup;@Nullableprivate String lastInternalSavepoint;@Nullableprivate ResourceManagerAddress resourceManagerAddress;@Nullableprivate ResourceManagerConnection resourceManagerConnection;@Nullableprivate EstablishedResourceManagerConnection establishedResourceManagerConnection;//......@Overridepublic CompletableFuture<Acknowledge> notifyKvStateRegistered(final JobID jobId,final JobVertexID jobVertexId,final KeyGroupRange keyGroupRange,final String registrationName,final KvStateID kvStateId,final InetSocketAddress kvStateServerAddress) {if (jobGraph.getJobID().equals(jobId)) {if (log.isDebugEnabled()) {log.debug("Key value state registered for job {} under name {}.",jobGraph.getJobID(), registrationName);}try {executionGraph.getKvStateLocationRegistry().notifyKvStateRegistered(jobVertexId, keyGroupRange, registrationName, kvStateId, kvStateServerAddress);return CompletableFuture.completedFuture(Acknowledge.get());} catch (Exception e) {log.error("Failed to notify KvStateRegistry about registration {}.", registrationName);return FutureUtils.completedExceptionally(e);}} else {if (log.isDebugEnabled()) {log.debug("Notification about key-value state registration for unknown job {} received.", jobId);}return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));}}@Overridepublic CompletableFuture<Acknowledge> notifyKvStateUnregistered(JobID jobId,JobVertexID jobVertexId,KeyGroupRange keyGroupRange,String registrationName) {if (jobGraph.getJobID().equals(jobId)) {if (log.isDebugEnabled()) {log.debug("Key value state unregistered for job {} under name {}.",jobGraph.getJobID(), registrationName);}try {executionGraph.getKvStateLocationRegistry().notifyKvStateUnregistered(jobVertexId, keyGroupRange, registrationName);return CompletableFuture.completedFuture(Acknowledge.get());} catch (Exception e) {log.error("Failed to notify KvStateRegistry about registration {}.", registrationName, e);return FutureUtils.completedExceptionally(e);}} else {if (log.isDebugEnabled()) {log.debug("Notification about key-value state deregistration for unknown job {} received.", jobId);}return FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId));}}//......
}

• JobMaster的notifyKvStateRegistered方法主要是触发executionGraph.getKvStateLocationRegistry().notifyKvStateRegistered；notifyKvStateUnregistered方法主要是触发executionGraph.getKvStateLocationRegistry().notifyKvStateUnregistered

KvStateLocationRegistry

flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateLocationRegistry.java

public class KvStateLocationRegistry {/** JobID this coordinator belongs to. */private final JobID jobId;/** Job vertices for determining parallelism per key. */private final Map<JobVertexID, ExecutionJobVertex> jobVertices;/*** Location info keyed by registration name. The name needs to be unique* per JobID, i.e. two operators cannot register KvState with the same* name.*/private final Map<String, KvStateLocation> lookupTable = new HashMap<>();/*** Creates the registry for the job.** @param jobId       JobID this coordinator belongs to.* @param jobVertices Job vertices map of all vertices of this job.*/public KvStateLocationRegistry(JobID jobId, Map<JobVertexID, ExecutionJobVertex> jobVertices) {this.jobId = Preconditions.checkNotNull(jobId, "JobID");this.jobVertices = Preconditions.checkNotNull(jobVertices, "Job vertices");}/*** Returns the {@link KvStateLocation} for the registered KvState instance* or <code>null</code> if no location information is available.** @param registrationName Name under which the KvState instance is registered.* @return Location information or <code>null</code>.*/public KvStateLocation getKvStateLocation(String registrationName) {return lookupTable.get(registrationName);}/*** Notifies the registry about a registered KvState instance.** @param jobVertexId JobVertexID the KvState instance belongs to* @param keyGroupRange Key group range the KvState instance belongs to* @param registrationName Name under which the KvState has been registered* @param kvStateId ID of the registered KvState instance* @param kvStateServerAddress Server address where to find the KvState instance** @throws IllegalArgumentException If JobVertexID does not belong to job* @throws IllegalArgumentException If state has been registered with same* name by another operator.* @throws IndexOutOfBoundsException If key group index is out of bounds.*/public void notifyKvStateRegistered(JobVertexID jobVertexId,KeyGroupRange keyGroupRange,String registrationName,KvStateID kvStateId,InetSocketAddress kvStateServerAddress) {KvStateLocation location = lookupTable.get(registrationName);if (location == null) {// First registration for this operator, create the location infoExecutionJobVertex vertex = jobVertices.get(jobVertexId);if (vertex != null) {int parallelism = vertex.getMaxParallelism();location = new KvStateLocation(jobId, jobVertexId, parallelism, registrationName);lookupTable.put(registrationName, location);} else {throw new IllegalArgumentException("Unknown JobVertexID " + jobVertexId);}}// Duplicated name if vertex IDs don't matchif (!location.getJobVertexId().equals(jobVertexId)) {IllegalStateException duplicate = new IllegalStateException("Registration name clash. KvState with name '" + registrationName +"' has already been registered by another operator (" +location.getJobVertexId() + ").");ExecutionJobVertex vertex = jobVertices.get(jobVertexId);if (vertex != null) {vertex.fail(new SuppressRestartsException(duplicate));}throw duplicate;}location.registerKvState(keyGroupRange, kvStateId, kvStateServerAddress);}/*** Notifies the registry about an unregistered KvState instance.** @param jobVertexId JobVertexID the KvState instance belongs to* @param keyGroupRange Key group index the KvState instance belongs to* @param registrationName Name under which the KvState has been registered* @throws IllegalArgumentException If another operator registered the state instance* @throws IllegalArgumentException If the registration name is not known*/public void notifyKvStateUnregistered(JobVertexID jobVertexId,KeyGroupRange keyGroupRange,String registrationName) {KvStateLocation location = lookupTable.get(registrationName);if (location != null) {// Duplicate name if vertex IDs don't matchif (!location.getJobVertexId().equals(jobVertexId)) {throw new IllegalArgumentException("Another operator (" +location.getJobVertexId() + ") registered the KvState " +"under '" + registrationName + "'.");}location.unregisterKvState(keyGroupRange);if (location.getNumRegisteredKeyGroups() == 0) {lookupTable.remove(registrationName);}} else {throw new IllegalArgumentException("Unknown registration name '" +registrationName + "'. " + "Probably registration/unregistration race.");}}}

• KvStateLocationRegistry的构造器要求传入jobId及jobVertices；它有一个属性为lookupTable，存储了registrationName与KvStateLocation的映射关系
• notifyKvStateRegistered方法在lookupTable查找不到对应的KvStateLocation的时候会创建一个KvStateLocation并存放入lookupTable，最后调用location.registerKvState方法
• notifyKvStateUnregistere方法在lookupTable查找对应KvStateLocation的时候会触发location.unregisterKvState，然后将该KvStateLocation从lookupTable中移除

小结

• KvStateRegistryGateway接口定义了notifyKvStateRegistered、notifyKvStateUnregistered两个方法；JobMaster实现了这两个方法
• JobMaster的notifyKvStateRegistered方法主要是触发executionGraph.getKvStateLocationRegistry().notifyKvStateRegistered；notifyKvStateUnregistered方法主要是触发executionGraph.getKvStateLocationRegistry().notifyKvStateUnregistered
• KvStateLocationRegistry的构造器要求传入jobId及jobVertices；它有一个属性为lookupTable，存储了registrationName与KvStateLocation的映射关系；notifyKvStateRegistered方法在lookupTable查找不到对应的KvStateLocation的时候会创建一个KvStateLocation并存放入lookupTable，最后调用location.registerKvState方法；notifyKvStateUnregistere方法在lookupTable查找对应KvStateLocation的时候会触发location.unregisterKvState，然后将该KvStateLocation从lookupTable中移除

doc

• KvStateRegistryGateway