写给大忙人看的 Flink Operator State 的存储方式

	@Override
	public abstract <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
		Environment env,
		JobID jobID,
		String operatorIdentifier,
		TypeSerializer<K> keySerializer,
		int numberOfKeyGroups,
		KeyGroupRange keyGroupRange,
		TaskKvStateRegistry kvStateRegistry,
		TtlTimeProvider ttlTimeProvider,
		MetricGroup metricGroup,
		@Nonnull Collection<KeyedStateHandle> stateHandles,
		CloseableRegistry cancelStreamRegistry) throws IOException;

	@Override
	public abstract OperatorStateBackend createOperatorStateBackend(
		Environment env,
		String operatorIdentifier,
		@Nonnull Collection<OperatorStateHandle> stateHandles,
		CloseableRegistry cancelStreamRegistry) throws Exception;

@Override
	public DefaultOperatorStateBackend build() throws BackendBuildingException {
		Map<String, PartitionableListState<?>> registeredOperatorStates = new HashMap<>();
		Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates = new HashMap<>();
		CloseableRegistry cancelStreamRegistryForBackend = new CloseableRegistry();
		AbstractSnapshotStrategy<OperatorStateHandle> snapshotStrategy =
			new DefaultOperatorStateBackendSnapshotStrategy(
				userClassloader,
				asynchronousSnapshots,
				registeredOperatorStates,
				registeredBroadcastStates,
				cancelStreamRegistryForBackend);
		OperatorStateRestoreOperation restoreOperation = new OperatorStateRestoreOperation(
			cancelStreamRegistry,
			userClassloader,
			registeredOperatorStates,
			registeredBroadcastStates,
			restoreStateHandles
		);
		try {
			// OperatorState restore
			restoreOperation.restore();
		} catch (Exception e) {
			IOUtils.closeQuietly(cancelStreamRegistryForBackend);
			throw new BackendBuildingException("Failed when trying to restore operator state backend", e);
		}
		// 当执行完 restoreOperation.restore(); 之后将得到的 registeredOperatorStates 传入 DefaultOperatorStateBackend
		return new DefaultOperatorStateBackend(
			executionConfig,
			cancelStreamRegistryForBackend,
			registeredOperatorStates,
			registeredBroadcastStates,
			new HashMap<>(),
			new HashMap<>(),
			snapshotStrategy
		);
	}

@Override
	public Void restore() throws Exception {
		if (stateHandles.isEmpty()) {
			return null;
		}

		for (OperatorStateHandle stateHandle : stateHandles) {

			if (stateHandle == null) {
				continue;
			}
			//读取 file stream
			FSDataInputStream in = stateHandle.openInputStream();
			closeStreamOnCancelRegistry.registerCloseable(in);

			ClassLoader restoreClassLoader = Thread.currentThread().getContextClassLoader();

			try {
				Thread.currentThread().setContextClassLoader(userClassloader);
				OperatorBackendSerializationProxy backendSerializationProxy =
					new OperatorBackendSerializationProxy(userClassloader);

				backendSerializationProxy.read(new DataInputViewStreamWrapper(in));

				List<StateMetaInfoSnapshot> restoredOperatorMetaInfoSnapshots =
					backendSerializationProxy.getOperatorStateMetaInfoSnapshots();

				// Recreate all PartitionableListStates from the meta info
				for (StateMetaInfoSnapshot restoredSnapshot : restoredOperatorMetaInfoSnapshots) {

					final RegisteredOperatorStateBackendMetaInfo<?> restoredMetaInfo =
						new RegisteredOperatorStateBackendMetaInfo<>(restoredSnapshot);

					if (restoredMetaInfo.getPartitionStateSerializer() instanceof UnloadableDummyTypeSerializer) {

						// must fail now if the previous typeSerializer cannot be restored because there is no typeSerializer
						// capable of reading previous state
						// TODO when eager state registration is in place, we can try to get a convert deserializer
						// TODO from the newly registered typeSerializer instead of simply failing here

						throw new IOException("Unable to restore operator state [" + restoredSnapshot.getName() + "]." +
							" The previous typeSerializer of the operator state must be present; the typeSerializer could" +
							" have been removed from the classpath, or its implementation have changed and could" +
							" not be loaded. This is a temporary restriction that will be fixed in future versions.");
					}

					PartitionableListState<?> listState = registeredOperatorStates.get(restoredSnapshot.getName());

					if (null == listState) {
						listState = new PartitionableListState<>(restoredMetaInfo);

						registeredOperatorStates.put(listState.getStateMetaInfo().getName(), listState);
					} else {
						// TODO with eager state registration in place, check here for typeSerializer migration strategies
					}
				}

				// ... and then get back the broadcast state.
				List<StateMetaInfoSnapshot> restoredBroadcastMetaInfoSnapshots =
					backendSerializationProxy.getBroadcastStateMetaInfoSnapshots();

				for (StateMetaInfoSnapshot restoredSnapshot : restoredBroadcastMetaInfoSnapshots) {

					final RegisteredBroadcastStateBackendMetaInfo<?, ?> restoredMetaInfo =
						new RegisteredBroadcastStateBackendMetaInfo<>(restoredSnapshot);

					if (restoredMetaInfo.getKeySerializer() instanceof UnloadableDummyTypeSerializer ||
						restoredMetaInfo.getValueSerializer() instanceof UnloadableDummyTypeSerializer) {

						// must fail now if the previous typeSerializer cannot be restored because there is no typeSerializer
						// capable of reading previous state
						// TODO when eager state registration is in place, we can try to get a convert deserializer
						// TODO from the newly registered typeSerializer instead of simply failing here

						throw new IOException("Unable to restore broadcast state [" + restoredSnapshot.getName() + "]." +
							" The previous key and value serializers of the state must be present; the serializers could" +
							" have been removed from the classpath, or their implementations have changed and could" +
							" not be loaded. This is a temporary restriction that will be fixed in future versions.");
					}

					BackendWritableBroadcastState<?, ?> broadcastState = registeredBroadcastStates.get(restoredSnapshot.getName());

					if (broadcastState == null) {
						broadcastState = new HeapBroadcastState<>(restoredMetaInfo);

						registeredBroadcastStates.put(broadcastState.getStateMetaInfo().getName(), broadcastState);
					} else {
						// TODO with eager state registration in place, check here for typeSerializer migration strategies
					}
				}

				// Restore all the states
				for (Map.Entry<String, OperatorStateHandle.StateMetaInfo> nameToOffsets :
					stateHandle.getStateNameToPartitionOffsets().entrySet()) {

					final String stateName = nameToOffsets.getKey();

					PartitionableListState<?> listStateForName = registeredOperatorStates.get(stateName);
					if (listStateForName == null) {
						BackendWritableBroadcastState<?, ?> broadcastStateForName = registeredBroadcastStates.get(stateName);
						Preconditions.checkState(broadcastStateForName != null, "Found state without " +
							"corresponding meta info: " + stateName);
						deserializeBroadcastStateValues(broadcastStateForName, in, nameToOffsets.getValue());
					} else {
						deserializeOperatorStateValues(listStateForName, in, nameToOffsets.getValue());
					}
				}

			} finally {
				Thread.currentThread().setContextClassLoader(restoreClassLoader);
				if (closeStreamOnCancelRegistry.unregisterCloseable(in)) {
					IOUtils.closeQuietly(in);
				}
			}
		}
		return null;
	}

public final void initializeState(FunctionInitializationContext context) throws Exception {
.....
		this.unionOffsetStates = stateStore.getUnionListState(new ListStateDescriptor<>(
				OFFSETS_STATE_NAME,
				TypeInformation.of(new TypeHint<Tuple2<KafkaTopicPartition, Long>>() {})));
.....
	}

public final void snapshotState(FunctionSnapshotContext context) throws Exception {
		...
					unionOffsetStates.add(
							Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
				...
	}

@Override
	public void add(S value) {
		Preconditions.checkNotNull(value, "You cannot add null to a ListState.");
		internalList.add(value);
	}

...
PartitionableListState<?> value = entry.getValue();
						//operator state 写入 hdfs
						long[] partitionOffsets = value.write(localOut);
						...

public long[] write(FSDataOutputStream out) throws IOException {

		long[] partitionOffsets = new long[internalList.size()];

		DataOutputView dov = new DataOutputViewStreamWrapper(out);

		for (int i = 0; i < internalList.size(); ++i) {
			S element = internalList.get(i);
			partitionOffsets[i] = out.getPos();
			getStateMetaInfo().getPartitionStateSerializer().serialize(element, dov);
		}

		return partitionOffsets;
	}