一文搞懂 Flink rocksdb 中的数据恢复

// rockdb restore 入口方法
	private AbstractRocksDBRestoreOperation<K> getRocksDBRestoreOperation(
		int keyGroupPrefixBytes,
		CloseableRegistry cancelStreamRegistry,
		LinkedHashMap<String, RocksDBKeyedStateBackend.RocksDbKvStateInfo> kvStateInformation,
		RocksDbTtlCompactFiltersManager ttlCompactFiltersManager) {
		if (restoreStateHandles.isEmpty()) {
			return new RocksDBNoneRestoreOperation<>(
				keyGroupRange,
				keyGroupPrefixBytes,
				numberOfTransferingThreads,
				cancelStreamRegistry,
				userCodeClassLoader,
				kvStateInformation,
				keySerializerProvider,
				instanceBasePath,
				instanceRocksDBPath,
				dbOptions,
				columnFamilyOptionsFactory,
				nativeMetricOptions,
				metricGroup,
				restoreStateHandles,
				ttlCompactFiltersManager);
		}
		KeyedStateHandle firstStateHandle = restoreStateHandles.iterator().next();
		if (firstStateHandle instanceof IncrementalKeyedStateHandle) {
			return new RocksDBIncrementalRestoreOperation<>(
				operatorIdentifier,
				keyGroupRange,
				keyGroupPrefixBytes,
				numberOfTransferingThreads,
				cancelStreamRegistry,
				userCodeClassLoader,
				kvStateInformation,
				keySerializerProvider,
				instanceBasePath,
				instanceRocksDBPath,
				dbOptions,
				columnFamilyOptionsFactory,
				nativeMetricOptions,
				metricGroup,
				restoreStateHandles,
				ttlCompactFiltersManager);
		} else {
			return new RocksDBFullRestoreOperation<>(
				keyGroupRange,
				keyGroupPrefixBytes,
				numberOfTransferingThreads,
				cancelStreamRegistry,
				userCodeClassLoader,
				kvStateInformation,
				keySerializerProvider,
				instanceBasePath,
				instanceRocksDBPath,
				dbOptions,
				columnFamilyOptionsFactory,
				nativeMetricOptions,
				metricGroup,
				restoreStateHandles,
				ttlCompactFiltersManager);
		}
	}

@Override
	public RocksDBRestoreResult restore() throws Exception {
		if (restoreStateHandles == null || restoreStateHandles.isEmpty()) {
			return null;
		}
		final KeyedStateHandle theFirstStateHandle = restoreStateHandles.iterator().next();
		boolean isRescaling = (restoreStateHandles.size() > 1 ||
			!Objects.equals(theFirstStateHandle.getKeyGroupRange(), keyGroupRange));
		if (isRescaling) {
			restoreWithRescaling(restoreStateHandles);
		} else {
			restoreWithoutRescaling(theFirstStateHandle);
		}
		return new RocksDBRestoreResult(this.db, defaultColumnFamilyHandle,
			nativeMetricMonitor, lastCompletedCheckpointId, backendUID, restoredSstFiles);
	}

private void restoreWithRescaling(Collection<KeyedStateHandle> restoreStateHandles) throws Exception {
		// Prepare for restore with rescaling
		KeyedStateHandle initialHandle = RocksDBIncrementalCheckpointUtils.chooseTheBestStateHandleForInitial(
			restoreStateHandles, keyGroupRange);
		// Init base DB instance
		if (initialHandle != null) {
			restoreStateHandles.remove(initialHandle);
			initDBWithRescaling(initialHandle);
		} else {
			openDB();
		}
		// Transfer remaining key-groups from temporary instance into base DB
		byte[] startKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
		RocksDBKeySerializationUtils.serializeKeyGroup(keyGroupRange.getStartKeyGroup(), startKeyGroupPrefixBytes);
		byte[] stopKeyGroupPrefixBytes = new byte[keyGroupPrefixBytes];
		RocksDBKeySerializationUtils.serializeKeyGroup(keyGroupRange.getEndKeyGroup() + 1, stopKeyGroupPrefixBytes);
		for (KeyedStateHandle rawStateHandle : restoreStateHandles) {
			if (!(rawStateHandle instanceof IncrementalRemoteKeyedStateHandle)) {
				throw new IllegalStateException("Unexpected state handle type, " +
					"expected " + IncrementalRemoteKeyedStateHandle.class +
					", but found " + rawStateHandle.getClass());
			}
			//本地的
			Path temporaryRestoreInstancePath = new Path(instanceBasePath.getAbsolutePath() + UUID.randomUUID().toString());
			//首先呢会把 rawStateHandle 对应的 state 数据下载到 temporaryRestoreInstancePath 并且作为一个临时的 RocksDB 实例的数据目录
			try (RestoredDBInstance tmpRestoreDBInfo = restoreDBInstanceFromStateHandle(
				(IncrementalRemoteKeyedStateHandle) rawStateHandle,
				temporaryRestoreInstancePath);
				RocksDBWriteBatchWrapper writeBatchWrapper = new RocksDBWriteBatchWrapper(this.db)) {
				List<ColumnFamilyDescriptor> tmpColumnFamilyDescriptors = tmpRestoreDBInfo.columnFamilyDescriptors;
				List<ColumnFamilyHandle> tmpColumnFamilyHandles = tmpRestoreDBInfo.columnFamilyHandles;
				// iterating only the requested descriptors automatically skips the default column family handle
				for (int i = 0; i < tmpColumnFamilyDescriptors.size(); ++i) {
					ColumnFamilyHandle tmpColumnFamilyHandle = tmpColumnFamilyHandles.get(i);
					ColumnFamilyHandle targetColumnFamilyHandle = getOrRegisterStateColumnFamilyHandle(
						null, tmpRestoreDBInfo.stateMetaInfoSnapshots.get(i))
						.columnFamilyHandle;
					//会把临时的 rockdb 实例的数据写入到 rocksdb 中
					try (RocksIteratorWrapper iterator = RocksDBOperationUtils.getRocksIterator(tmpRestoreDBInfo.db, tmpColumnFamilyHandle)) {
						iterator.seek(startKeyGroupPrefixBytes);
						while (iterator.isValid()) {
							if (RocksDBIncrementalCheckpointUtils.beforeThePrefixBytes(iterator.key(), stopKeyGroupPrefixBytes)) {
								// insert data to rocksdb
								writeBatchWrapper.put(targetColumnFamilyHandle, iterator.key(), iterator.value());
							} else {
								// Since the iterator will visit the record according to the sorted order,
								// we can just break here.
								break;
							}
							iterator.next();
						}
					} // releases native iterator resources
				}
			} finally {
				cleanUpPathQuietly(temporaryRestoreInstancePath);
			}
		}
	}