docs: private void restoreKVStateData() throws IOException, RocksDBException {
  		//for all key-groups in the current state handle...
  		try (RocksDBWriteBatchWrapper writeBatchWrapper = new RocksDBWriteBatchWrapper(db)) {
  			for (Tuple2<Integer, Long> keyGroupOffset : currentKeyGroupsStateHandle.getGroupRangeOffsets()) {
  				int keyGroup = keyGroupOffset.f0;
  
  				// Check that restored key groups all belong to the backend
  				Preconditions.checkState(keyGroupRange.contains(keyGroup),
  					"The key group must belong to the backend");
  
  				long offset = keyGroupOffset.f1;
  				//not empty key-group?
  				if (0L != offset) {
  					currentStateHandleInStream.seek(offset);
  					try (InputStream compressedKgIn = keygroupStreamCompressionDecorator.decorateWithCompression(currentStateHandleInStream)) {
  						DataInputViewStreamWrapper compressedKgInputView = new DataInputViewStreamWrapper(compressedKgIn);
  						//TODO this could be aware of keyGroupPrefixBytes and write only one byte if possible
  						int kvStateId = compressedKgInputView.readShort();
  						ColumnFamilyHandle handle = currentStateHandleKVStateColumnFamilies.get(kvStateId);
  						//insert all k/v pairs into DB
  						boolean keyGroupHasMoreKeys = true;
  						while (keyGroupHasMoreKeys) {
  							byte[] key = BytePrimitiveArraySerializer.INSTANCE.deserialize(compressedKgInputView);
  							byte[] value = BytePrimitiveArraySerializer.INSTANCE.deserialize(compressedKgInputView);
  							if (hasMetaDataFollowsFlag(key)) {
  								//clear the signal bit in the key to make it ready for insertion again
  								clearMetaDataFollowsFlag(key);
  								writeBatchWrapper.put(handle, key, value);
  								//TODO this could be aware of keyGroupPrefixBytes and write only one byte if possible
  								kvStateId = END_OF_KEY_GROUP_MARK
  									& compressedKgInputView.readShort();
  								if (END_OF_KEY_GROUP_MARK == kvStateId) {
  									keyGroupHasMoreKeys = false;
  								} else {
  									handle = currentStateHandleKVStateColumnFamilies.get(kvStateId);
  								}
  							} else {
  								writeBatchWrapper.put(handle, key, value);
  							}
  						}
  					}
  				}
  			}
  		}
  	}