import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Override
public void update(V value) throws IOException {
	if (value == null) {
		clear();
		return;
	}
	DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(keySerializationStream);
	try {
		writeCurrentKeyWithGroupAndNamespace();
		byte[] key = keySerializationStream.toByteArray();
		keySerializationStream.reset();
		valueSerializer.serialize(value, out);
		backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
	} catch (Exception e) {
		throw new RuntimeException("Error while adding data to RocksDB", e);
	}
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Override
public void add(V value) throws IOException {
	try {
		writeCurrentKeyWithGroupAndNamespace();
		byte[] key = keySerializationStream.toByteArray();
		byte[] valueBytes = backend.db.get(columnFamily, key);

		DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(keySerializationStream);
		if (valueBytes == null) {
			keySerializationStream.reset();
			valueSerializer.serialize(value, out);
			backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
		} else {
			V oldValue = valueSerializer.deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStream(valueBytes)));
			V newValue = reduceFunction.reduce(oldValue, value);
			keySerializationStream.reset();
			valueSerializer.serialize(newValue, out);
			backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
		}
	} catch (Exception e) {
		throw new RuntimeException("Error while adding data to RocksDB", e);
	}
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Override
public void add(T value) throws IOException {
	try {
		writeCurrentKeyWithGroupAndNamespace();
		byte[] key = keySerializationStream.toByteArray();
		byte[] valueBytes = backend.db.get(columnFamily, key);
		DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(keySerializationStream);
		if (valueBytes == null) {
			keySerializationStream.reset();
			valueSerializer.serialize(foldFunction.fold(stateDesc.getDefaultValue(), value), out);
			backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
		} else {
			ACC oldValue = valueSerializer.deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStreamWithPos(valueBytes)));
			ACC newValue = foldFunction.fold(oldValue, value);
			keySerializationStream.reset();
			valueSerializer.serialize(newValue, out);
			backend.db.put(columnFamily, writeOptions, key, keySerializationStream.toByteArray());
		}
	} catch (Exception e) {
		throw new RuntimeException("Error while adding data to RocksDB", e);
	}
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Creates a new RocksDB backed state.
 *  @param namespaceSerializer The serializer for the namespace.
 */
protected AbstractRocksDBState(
		ColumnFamilyHandle columnFamily,
		TypeSerializer<N> namespaceSerializer,
		SD stateDesc,
		RocksDBKeyedStateBackend<K> backend) {

	this.namespaceSerializer = namespaceSerializer;
	this.backend = backend;

	this.columnFamily = columnFamily;

	writeOptions = new WriteOptions();
	writeOptions.setDisableWAL(true);
	this.stateDesc = Preconditions.checkNotNull(stateDesc, "State Descriptor");

	this.keySerializationStream = new ByteArrayOutputStreamWithPos(128);
	this.keySerializationDataOutputView = new DataOutputViewStreamWrapper(keySerializationStream);
	this.ambiguousKeyPossible = (backend.getKeySerializer().getLength() < 0)
			&& (namespaceSerializer.getLength() < 0);
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public byte[] getSerializedValue(byte[] serializedKeyAndNamespace) throws Exception {
	Preconditions.checkNotNull(serializedKeyAndNamespace, "Serialized key and namespace");

	//TODO make KvStateSerializer key-group aware to save this round trip and key-group computation
	Tuple2<K, N> des = KvStateSerializer.<K, N>deserializeKeyAndNamespace(
			serializedKeyAndNamespace,
			backend.getKeySerializer(),
			namespaceSerializer);

	int keyGroup = KeyGroupRangeAssignment.assignToKeyGroup(des.f0, backend.getNumberOfKeyGroups());

	// we cannot reuse the keySerializationStream member since this method
	// is called concurrently to the other ones and it may thus contain garbage
	ByteArrayOutputStreamWithPos tmpKeySerializationStream = new ByteArrayOutputStreamWithPos(128);
	DataOutputViewStreamWrapper tmpKeySerializationDateDataOutputView = new DataOutputViewStreamWrapper(tmpKeySerializationStream);

	writeKeyWithGroupAndNamespace(keyGroup, des.f0, des.f1,
		tmpKeySerializationStream, tmpKeySerializationDateDataOutputView);

	return backend.db.get(columnFamily, tmpKeySerializationStream.toByteArray());
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Write a list of serializers and their corresponding config snapshots to the provided
 * data output view. This method writes in a fault tolerant way, so that when read again
 * using {@link #readSerializersAndConfigsWithResilience(DataInputView, ClassLoader)}, if
 * deserialization of the serializer fails, its configuration snapshot will remain intact.
 *
 * <p>Specifically, all written serializers and their config snapshots are indexed by their
 * offset positions within the serialized bytes. The serialization format is as follows:
 * <ul>
 *     <li>1. number of serializer and configuration snapshot pairs.</li>
 *     <li>2. offsets of each serializer and configuration snapshot, in order.</li>
 *     <li>3. total number of bytes for the serialized serializers and the config snapshots.</li>
 *     <li>4. serialized serializers and the config snapshots.</li>
 * </ul>
 *
 * @param out the data output view.
 * @param serializersAndConfigs serializer and configuration snapshot pairs
 *
 * @throws IOException
 */
public static void writeSerializersAndConfigsWithResilience(
		DataOutputView out,
		List<Tuple2<TypeSerializer<?>, TypeSerializerConfigSnapshot>> serializersAndConfigs) throws IOException {

	try (
		ByteArrayOutputStreamWithPos bufferWithPos = new ByteArrayOutputStreamWithPos();
		DataOutputViewStreamWrapper bufferWrapper = new DataOutputViewStreamWrapper(bufferWithPos)) {

		out.writeInt(serializersAndConfigs.size());
		for (Tuple2<TypeSerializer<?>, TypeSerializerConfigSnapshot> serAndConfSnapshot : serializersAndConfigs) {
			out.writeInt(bufferWithPos.getPosition());
			writeSerializer(bufferWrapper, serAndConfSnapshot.f0);

			out.writeInt(bufferWithPos.getPosition());
			writeSerializerConfigSnapshot(bufferWrapper, serAndConfSnapshot.f1);
		}

		out.writeInt(bufferWithPos.getPosition());
		out.write(bufferWithPos.getBuf(), 0, bufferWithPos.getPosition());
	}
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Clones the given writable using the {@link IOReadableWritable serialization}.
 *
 * @param original Object to clone
 * @param <T> Type of the object to clone
 * @return Cloned object
 * @throws IOException Thrown is the serialization fails.
 */
public static <T extends IOReadableWritable> T createCopyWritable(T original) throws IOException {
	if (original == null) {
		return null;
	}

	final ByteArrayOutputStream baos = new ByteArrayOutputStream();
	try (DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(baos)) {
		original.write(out);
	}

	final ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
	try (DataInputViewStreamWrapper in = new DataInputViewStreamWrapper(bais)) {

		@SuppressWarnings("unchecked")
		T copy = (T) instantiate(original.getClass());
		copy.read(in);
		return copy;
	}
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that reconfiguring with a config snapshot of a preceding POJO serializer
 * with different POJO type will result in INCOMPATIBLE.
 */
@Test
public void testReconfigureWithDifferentPojoType() throws Exception {
	PojoSerializer<SubTestUserClassB> pojoSerializer1 = (PojoSerializer<SubTestUserClassB>)
		TypeExtractor.getForClass(SubTestUserClassB.class).createSerializer(new ExecutionConfig());

	// snapshot configuration and serialize to bytes
	TypeSerializerConfigSnapshot pojoSerializerConfigSnapshot = pojoSerializer1.snapshotConfiguration();
	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerSerializationUtil.writeSerializerConfigSnapshot(new DataOutputViewStreamWrapper(out), pojoSerializerConfigSnapshot);
		serializedConfig = out.toByteArray();
	}

	PojoSerializer<SubTestUserClassA> pojoSerializer2 = (PojoSerializer<SubTestUserClassA>)
		TypeExtractor.getForClass(SubTestUserClassA.class).createSerializer(new ExecutionConfig());

	// read configuration again from bytes
	try(ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		pojoSerializerConfigSnapshot = TypeSerializerSerializationUtil.readSerializerConfigSnapshot(
			new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
	}

	CompatibilityResult<SubTestUserClassA> compatResult = pojoSerializer2.ensureCompatibility(pojoSerializerConfigSnapshot);
	assertTrue(compatResult.isRequiresMigration());
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that reconfiguration result is INCOMPATIBLE if data type has changed.
 */
@Test
public void testMigrationStrategyWithDifferentKryoType() throws Exception {
	KryoSerializer<TestClassA> kryoSerializerForA = new KryoSerializer<>(TestClassA.class, new ExecutionConfig());

	// snapshot configuration and serialize to bytes
	TypeSerializerConfigSnapshot kryoSerializerConfigSnapshot = kryoSerializerForA.snapshotConfiguration();
	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerSerializationUtil.writeSerializerConfigSnapshot(new DataOutputViewStreamWrapper(out), kryoSerializerConfigSnapshot);
		serializedConfig = out.toByteArray();
	}

	KryoSerializer<TestClassB> kryoSerializerForB = new KryoSerializer<>(TestClassB.class, new ExecutionConfig());

	// read configuration again from bytes
	try (ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		kryoSerializerConfigSnapshot = TypeSerializerSerializationUtil.readSerializerConfigSnapshot(
			new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
	}

	CompatibilityResult<TestClassB> compatResult = kryoSerializerForB.ensureCompatibility(kryoSerializerConfigSnapshot);
	assertTrue(compatResult.isRequiresMigration());
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Test
public void testSnapshotConfigurationAndReconfigure() throws Exception {
	final TypeSerializerConfigSnapshot configSnapshot = getSerializer().snapshotConfiguration();

	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerSerializationUtil.writeSerializerConfigSnapshot(
			new DataOutputViewStreamWrapper(out), configSnapshot);
		serializedConfig = out.toByteArray();
	}

	TypeSerializerConfigSnapshot restoredConfig;
	try (ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		restoredConfig = TypeSerializerSerializationUtil.readSerializerConfigSnapshot(
			new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
	}

	CompatibilityResult strategy = getSerializer().ensureCompatibility(restoredConfig);
	assertFalse(strategy.isRequiresMigration());

	// also verify that the serializer's reconfigure implementation detects incompatibility
	strategy = getSerializer().ensureCompatibility(new TestIncompatibleSerializerConfigSnapshot());
	assertTrue(strategy.isRequiresMigration());
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that reading and writing configuration snapshots work correctly.
 */
@Test
public void testSerializeConfigurationSnapshots() throws Exception {
	TestConfigSnapshot configSnapshot1 = new TestConfigSnapshot(1, "foo");
	TestConfigSnapshot configSnapshot2 = new TestConfigSnapshot(2, "bar");

	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerUtil.writeSerializerConfigSnapshots(
			new DataOutputViewStreamWrapper(out),
			configSnapshot1,
			configSnapshot2);

		serializedConfig = out.toByteArray();
	}

	TypeSerializerConfigSnapshot[] restoredConfigs;
	try (ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		restoredConfigs = TypeSerializerUtil.readSerializerConfigSnapshots(
			new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
	}

	assertEquals(2, restoredConfigs.length);
	assertEquals(configSnapshot1, restoredConfigs[0]);
	assertEquals(configSnapshot2, restoredConfigs[1]);
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that deserializing config snapshots fail if the config class could not be found.
 */
@Test
public void testFailsWhenConfigurationSnapshotClassNotFound() throws Exception {
	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerUtil.writeSerializerConfigSnapshot(
			new DataOutputViewStreamWrapper(out), new TestConfigSnapshot(123, "foobar"));
		serializedConfig = out.toByteArray();
	}

	try (ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		// read using a dummy classloader
		TypeSerializerUtil.readSerializerConfigSnapshot(
			new DataInputViewStreamWrapper(in), new URLClassLoader(new URL[0], null));
		fail("Expected a ClassNotFoundException wrapped in IOException");
	} catch (IOException expected) {
		// test passes
	}
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Test
public void testStateSerializerSerializationProxy() throws Exception {

	TypeSerializer<?> serializer = IntSerializer.INSTANCE;

	TypeSerializerSerializationProxy<?> proxy = new TypeSerializerSerializationProxy<>(serializer);

	byte[] serialized;
	try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) {
		proxy.write(new DataOutputViewStreamWrapper(out));
		serialized = out.toByteArray();
	}

	proxy = new TypeSerializerSerializationProxy<>(Thread.currentThread().getContextClassLoader());

	try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
		proxy.read(new DataInputViewStreamWrapper(in));
	}

	Assert.assertEquals(serializer, proxy.getTypeSerializer());
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that reading and writing serializers work correctly.
 */
@Test
public void testSerializerSerialization() throws Exception {

	TypeSerializer<?> serializer = IntSerializer.INSTANCE;

	byte[] serialized;
	try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) {
		TypeSerializerSerializationUtil.writeSerializer(new DataOutputViewStreamWrapper(out), serializer);
		serialized = out.toByteArray();
	}

	TypeSerializer<?> deserializedSerializer;
	try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
		deserializedSerializer = TypeSerializerSerializationUtil.tryReadSerializer(
			new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
	}

	Assert.assertEquals(serializer, deserializedSerializer);
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies deserialization failure cases when reading a serializer from bytes, in the
 * case of a {@link InvalidClassException}.
 */
@Test
public void testSerializerSerializationWithInvalidClass() throws Exception {

	TypeSerializer<?> serializer = IntSerializer.INSTANCE;

	byte[] serialized;
	try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) {
		TypeSerializerSerializationUtil.writeSerializer(new DataOutputViewStreamWrapper(out), serializer);
		serialized = out.toByteArray();
	}

	TypeSerializer<?> deserializedSerializer;

	// mock failure when deserializing serializers
	TypeSerializerSerializationUtil.TypeSerializerSerializationProxy<?> mockProxy =
			mock(TypeSerializerSerializationUtil.TypeSerializerSerializationProxy.class);
	doThrow(new IOException()).when(mockProxy).read(any(DataInputViewStreamWrapper.class));
	PowerMockito.whenNew(TypeSerializerSerializationUtil.TypeSerializerSerializationProxy.class).withAnyArguments().thenReturn(mockProxy);

	try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
		deserializedSerializer = TypeSerializerSerializationUtil.tryReadSerializer(
			new DataInputViewStreamWrapper(in), Thread.currentThread().getContextClassLoader());
	}
	Assert.assertEquals(null, deserializedSerializer);
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
/**
 * Verifies that deserializing config snapshots fail if the config class could not be found.
 */
@Test
public void testFailsWhenConfigurationSnapshotClassNotFound() throws Exception {
	byte[] serializedConfig;
	try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
		TypeSerializerSerializationUtil.writeSerializerConfigSnapshot(
			new DataOutputViewStreamWrapper(out), new TypeSerializerSerializationUtilTest.TestConfigSnapshot(123, "foobar"));
		serializedConfig = out.toByteArray();
	}

	try (ByteArrayInputStream in = new ByteArrayInputStream(serializedConfig)) {
		// read using a dummy classloader
		TypeSerializerSerializationUtil.readSerializerConfigSnapshot(
			new DataInputViewStreamWrapper(in), new URLClassLoader(new URL[0], null));
		fail("Expected a ClassNotFoundException wrapped in IOException");
	} catch (IOException expected) {
		// test passes
	}
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
public static void testSerialization(String[] values) throws IOException {
	ByteArrayOutputStream baos = new ByteArrayOutputStream(4096);
	DataOutputViewStreamWrapper serializer = new DataOutputViewStreamWrapper(baos);
	
	for (String value : values) {
		StringValue sv = new StringValue(value);
		sv.write(serializer);
	}
	
	serializer.close();
	baos.close();
	
	ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
	DataInputViewStreamWrapper deserializer = new DataInputViewStreamWrapper(bais);
	
	int num = 0;
	while (bais.available() > 0) {
		StringValue deser = new StringValue();
		deser.read(deserializer);
		
		assertEquals("DeserializedString differs from original string.", values[num], deser.getValue());
		num++;
	}
	
	assertEquals("Wrong number of deserialized values", values.length, num);
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Test
public void testReadSameVersion() throws Exception {

	String payload = "test";

	TestWriteable testWriteable = new TestWriteable(1, payload);
	byte[] serialized;
	try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) {
		testWriteable.write(new DataOutputViewStreamWrapper(out));
		serialized = out.toByteArray();
	}

	testWriteable = new TestWriteable(1);
	try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
		testWriteable.read(new DataInputViewStreamWrapper(in));
	}

	Assert.assertEquals(payload, testWriteable.getData());
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Test
public void testReadCompatibleVersion() throws Exception {

	String payload = "test";

	TestWriteable testWriteable = new TestWriteable(1, payload);
	byte[] serialized;
	try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) {
		testWriteable.write(new DataOutputViewStreamWrapper(out));
		serialized = out.toByteArray();
	}

	testWriteable = new TestWriteable(2) {
		@Override
		public int[] getCompatibleVersions() {
			return new int[] {1, 2};
		}
	};
	try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
		testWriteable.read(new DataInputViewStreamWrapper(in));
	}

	Assert.assertEquals(payload, testWriteable.getData());
}
 

import org.apache.flink.core.memory.DataOutputViewStreamWrapper; //导入依赖的package包/类
@Test
public void testReadMismatchVersion() throws Exception {

	String payload = "test";

	TestWriteable testWriteable = new TestWriteable(1, payload);
	byte[] serialized;
	try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) {
		testWriteable.write(new DataOutputViewStreamWrapper(out));
		serialized = out.toByteArray();
	}

	testWriteable = new TestWriteable(2);
	try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
		testWriteable.read(new DataInputViewStreamWrapper(in));
		Assert.fail("Version mismatch expected.");
	} catch (VersionMismatchException ignored) {

	}

	Assert.assertEquals(null, testWriteable.getData());
}