import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
/**
 * Snapshots the state {@code (stateName -> (valueCoder && (namespace -> value)))} for a given
 * {@code keyGroupIdx}.
 *
 * @param keyGroupIdx the id of the key-group to be put in the snapshot.
 * @param out the stream to write to.
 */
public void snapshotKeyGroupState(int keyGroupIdx, DataOutputStream out) throws Exception {
  int localIdx = getIndexForKeyGroup(keyGroupIdx);
  Map<String, Tuple2<Coder<?>, Map<String, ?>>> stateTable = stateTables[localIdx];
  Preconditions.checkState(stateTable.size() <= Short.MAX_VALUE,
      "Too many States: " + stateTable.size() + ". Currently at most "
          + Short.MAX_VALUE + " states are supported");
  out.writeShort(stateTable.size());
  for (Map.Entry<String, Tuple2<Coder<?>, Map<String, ?>>> entry : stateTable.entrySet()) {
    out.writeUTF(entry.getKey());
    Coder coder = entry.getValue().f0;
    InstantiationUtil.serializeObject(out, coder);
    Map<String, ?> map = entry.getValue().f1;
    out.writeInt(map.size());
    for (Map.Entry<String, ?> entry1 : map.entrySet()) {
      StringUtf8Coder.of().encode(entry1.getKey(), out);
      coder.encode(entry1.getValue(), out);
    }
  }
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
/**
 * Restore the state {@code (stateName -> (valueCoder && (namespace -> value)))}
 * for a given {@code keyGroupIdx}.
 *
 * @param keyGroupIdx the id of the key-group to be put in the snapshot.
 * @param in the stream to read from.
 * @param userCodeClassLoader the class loader that will be used to deserialize
 *                            the valueCoder.
 */
public void restoreKeyGroupState(int keyGroupIdx, DataInputStream in,
                                 ClassLoader userCodeClassLoader) throws Exception {
  int localIdx = getIndexForKeyGroup(keyGroupIdx);
  Map<String, Tuple2<Coder<?>, Map<String, ?>>> stateTable = stateTables[localIdx];
  int numStates = in.readShort();
  for (int i = 0; i < numStates; ++i) {
    String stateName = in.readUTF();
    Coder coder = InstantiationUtil.deserializeObject(in, userCodeClassLoader);
    Tuple2<Coder<?>, Map<String, ?>> tuple2 = stateTable.get(stateName);
    if (tuple2 == null) {
      tuple2 = new Tuple2<>();
      tuple2.f0 = coder;
      tuple2.f1 = new HashMap<>();
      stateTable.put(stateName, tuple2);
    }
    Map<String, Object> map = (Map<String, Object>) tuple2.f1;
    int mapSize = in.readInt();
    for (int j = 0; j < mapSize; j++) {
      String namespace = StringUtf8Coder.of().decode(in);
      Object value = coder.decode(in);
      map.put(namespace, value);
    }
  }
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
private SimpleConsumerThread<T> createAndStartSimpleConsumerThread(
		List<KafkaTopicPartitionState<TopicAndPartition>> seedPartitions,
		Node leader,
		ExceptionProxy errorHandler) throws IOException, ClassNotFoundException {
	// each thread needs its own copy of the deserializer, because the deserializer is
	// not necessarily thread safe
	final KeyedDeserializationSchema<T> clonedDeserializer =
			InstantiationUtil.clone(deserializer, runtimeContext.getUserCodeClassLoader());

	// seed thread with list of fetch partitions (otherwise it would shut down immediately again
	SimpleConsumerThread<T> brokerThread = new SimpleConsumerThread<>(
			this, errorHandler, kafkaConfig, leader, seedPartitions, unassignedPartitionsQueue,
			clonedDeserializer, invalidOffsetBehavior);

	brokerThread.setName(String.format("SimpleConsumer - %s - broker-%s (%s:%d)",
			runtimeContext.getTaskName(), leader.id(), leader.host(), leader.port()));
	brokerThread.setDaemon(true);
	brokerThread.start();

	LOG.info("Starting thread {}", brokerThread.getName());
	return brokerThread;
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
public RecordPairComparator(int[] keyFieldsReference, int[] keyFieldsCandidate, Class<? extends Value>[] keyTypes) {
	if (keyFieldsReference.length != keyFieldsCandidate.length || keyFieldsCandidate.length != keyTypes.length) {
		throw new IllegalArgumentException(
			"The arrays describing the key positions and types must be of the same length.");
	}
	this.keyFields1 = keyFieldsReference;
	this.keyFields2 = keyFieldsCandidate;
	
	// instantiate fields to extract keys into
	this.keyHolders1 = new Value[keyTypes.length];
	this.keyHolders2 = new Value[keyTypes.length];
	
	for (int i = 0; i < keyTypes.length; i++) {
		if (keyTypes[i] == null) {
			throw new NullPointerException("Key type " + i + " is null.");
		}
		this.keyHolders1[i] = InstantiationUtil.instantiate(keyTypes[i], Value.class);
		this.keyHolders2[i] = InstantiationUtil.instantiate(keyTypes[i], Value.class);
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
@Override
public E nextRecord(E reuseValue) throws IOException {
	if (reachedEnd()) {
		return null;
	}

	// if we start a new block, then register the event, and
	// restart the counter.
	if (dataFileReader.previousSync() != lastSync) {
		lastSync = dataFileReader.previousSync();
		recordsReadSinceLastSync = 0;
	}
	recordsReadSinceLastSync++;

	if (reuseAvroValue) {
		return dataFileReader.next(reuseValue);
	} else {
		if (GenericRecord.class == avroValueType) {
			return dataFileReader.next();
		} else {
			return dataFileReader.next(InstantiationUtil.instantiate(avroValueType, Object.class));
		}
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
@Test
public void testSerializability() throws IOException, ClassNotFoundException {
	final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();

	final AvroRowSerializationSchema serOrig = new AvroRowSerializationSchema(testData.f0);
	final AvroRowDeserializationSchema deserOrig = new AvroRowDeserializationSchema(testData.f0);

	byte[] serBytes = InstantiationUtil.serializeObject(serOrig);
	byte[] deserBytes = InstantiationUtil.serializeObject(deserOrig);

	AvroRowSerializationSchema serCopy =
		InstantiationUtil.deserializeObject(serBytes, Thread.currentThread().getContextClassLoader());
	AvroRowDeserializationSchema deserCopy =
		InstantiationUtil.deserializeObject(deserBytes, Thread.currentThread().getContextClassLoader());

	final byte[] bytes = serCopy.serialize(testData.f2);
	deserCopy.deserialize(bytes);
	deserCopy.deserialize(bytes);
	final Row actual = deserCopy.deserialize(bytes);

	assertEquals(testData.f2, actual);
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
public <T1, T2> TypePairComparatorFactory<T1, T2> getPairComparatorFactory(ClassLoader cl) {
	final String className = this.config.getString(DRIVER_PAIR_COMPARATOR_FACTORY, null);
	if (className == null) {
		return null;
	}
	
	@SuppressWarnings("unchecked")
	final Class<TypePairComparatorFactory<T1, T2>> superClass = (Class<TypePairComparatorFactory<T1, T2>>) (Class<?>) TypePairComparatorFactory.class;
	try {
		final Class<? extends TypePairComparatorFactory<T1, T2>> clazz = Class.forName(className, true, cl).asSubclass(superClass);
		return InstantiationUtil.instantiate(clazz, superClass);
	}
	catch (ClassNotFoundException cnfex) {
		throw new RuntimeException("The class '" + className + "', noted in the configuration as " +
			"pair comparator factory, could not be found. It is not part of the user code's class loader resources.");
	}
	catch (ClassCastException ccex) {
		throw new CorruptConfigurationException("The class noted in the configuration as the pair comparator factory " +
			"is no subclass of TypePairComparatorFactory.");
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
/**
 * Returns the type information factory for a type using the factory registry or annotations.
 */
@Internal
public static <OUT> TypeInfoFactory<OUT> getTypeInfoFactory(Type t) {
	final Class<?> factoryClass;
	if (registeredTypeInfoFactories.containsKey(t)) {
		factoryClass = registeredTypeInfoFactories.get(t);
	}
	else {
		if (!isClassType(t) || !typeToClass(t).isAnnotationPresent(TypeInfo.class)) {
			return null;
		}
		final TypeInfo typeInfoAnnotation = typeToClass(t).getAnnotation(TypeInfo.class);
		factoryClass = typeInfoAnnotation.value();
		// check for valid factory class
		if (!TypeInfoFactory.class.isAssignableFrom(factoryClass)) {
			throw new InvalidTypesException("TypeInfo annotation does not specify a valid TypeInfoFactory.");
		}
	}

	// instantiate
	return (TypeInfoFactory<OUT>) InstantiationUtil.instantiate(factoryClass);
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
public void restoreKeysForKeyGroup(DataInputViewStreamWrapper stream, int keyGroupIdx,
								ClassLoader userCodeClassLoader) throws IOException, ClassNotFoundException {

	checkArgument(localKeyGroupRange.contains(keyGroupIdx),
		"Key Group " + keyGroupIdx + " does not belong to the local range.");

	int numKeys = stream.readInt();
	if (numKeys > 0) {

		TypeSerializer<K> tmpKeyDeserializer = InstantiationUtil.deserializeObject(stream, userCodeClassLoader);

		if (keySerializer != null && !keySerializer.equals(tmpKeyDeserializer)) {
			throw new IllegalArgumentException("Tried to restore keys " +
				"for the watermark callback service with mismatching serializers.");
		}

		this.keySerializer = tmpKeyDeserializer;

		Set<K> keys = getRegisteredKeysForKeyGroup(keyGroupIdx);
		for (int i = 0; i < numKeys; i++) {
			keys.add(keySerializer.deserialize(stream));
		}
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private PojoComparator(PojoComparator<T> toClone) {
	this.keyFields = toClone.keyFields;
	this.comparators = new TypeComparator[toClone.comparators.length];

	for (int i = 0; i < toClone.comparators.length; i++) {
		this.comparators[i] = toClone.comparators[i].duplicate();
	}

	this.normalizedKeyLengths = toClone.normalizedKeyLengths;
	this.numLeadingNormalizableKeys = toClone.numLeadingNormalizableKeys;
	this.normalizableKeyPrefixLen = toClone.normalizableKeyPrefixLen;
	this.invertNormKey = toClone.invertNormKey;

	this.type = toClone.type;

	try {
		this.serializer = (TypeSerializer<T>) InstantiationUtil.deserializeObject(
				InstantiationUtil.serializeObject(toClone.serializer), Thread.currentThread().getContextClassLoader());
	} catch (IOException | ClassNotFoundException e) {
		throw new RuntimeException("Cannot copy serializer", e);
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
/**
 * This test checks the serialization/deserialization of {@link IntegerTaskEvent} objects.
 */
@Test
public void testIntegerTaskEvent() {

	try {
		final IntegerTaskEvent orig = new IntegerTaskEvent(11);
		final IntegerTaskEvent copy = InstantiationUtil.createCopyWritable(orig);

		assertEquals(orig.getInteger(), copy.getInteger());
		assertEquals(orig.hashCode(), copy.hashCode());
		assertTrue(orig.equals(copy));

	} catch (IOException ioe) {
		fail(ioe.getMessage());
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
/**
 * Tries to copy the given record from using the provided Kryo instance. If this fails, then
 * the record from is copied by serializing it into a byte buffer and deserializing it from
 * there.
 *
 * @param from Element to copy
 * @param reuse Reuse element for the deserialization
 * @param kryo Kryo instance to use
 * @param serializer TypeSerializer which is used in case of a Kryo failure
 * @param <T> Type of the element to be copied
 * @return Copied element
 */
public static <T> T copy(T from, T reuse, Kryo kryo, TypeSerializer<T> serializer) {
	try {
		return kryo.copy(from);
	} catch (KryoException ke) {
		// Kryo could not copy the object --> try to serialize/deserialize the object
		try {
			byte[] byteArray = InstantiationUtil.serializeToByteArray(serializer, from);

			return InstantiationUtil.deserializeFromByteArray(serializer, reuse, byteArray);
		} catch (IOException ioe) {
			throw new RuntimeException("Could not copy object by serializing/deserializing" +
				" it.", ioe);
		}
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void read(DataInputView in) throws IOException {
	super.read(in);

	String serializerConfigClassname = in.readUTF();
	Class<? extends TypeSerializerConfigSnapshot> serializerConfigSnapshotClass;
	try {
		serializerConfigSnapshotClass = (Class<? extends TypeSerializerConfigSnapshot>)
			Class.forName(serializerConfigClassname, true, userCodeClassLoader);
	} catch (ClassNotFoundException e) {
		throw new IOException(
			"Could not find requested TypeSerializerConfigSnapshot class "
				+ serializerConfigClassname +  " in classpath.", e);
	}

	serializerConfigSnapshot = InstantiationUtil.instantiate(serializerConfigSnapshotClass);
	serializerConfigSnapshot.setUserCodeClassLoader(userCodeClassLoader);
	serializerConfigSnapshot.read(in);
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
@Override
public void write(DataOutputView out) throws IOException {
	super.write(out);

	if (typeSerializer instanceof UnloadableDummyTypeSerializer) {
		UnloadableDummyTypeSerializer<T> dummyTypeSerializer =
			(UnloadableDummyTypeSerializer<T>) this.typeSerializer;

		byte[] serializerBytes = dummyTypeSerializer.getActualBytes();
		out.write(serializerBytes.length);
		out.write(serializerBytes);
	} else {
		// write in a way that allows the stream to recover from exceptions
		try (ByteArrayOutputStreamWithPos streamWithPos = new ByteArrayOutputStreamWithPos()) {
			InstantiationUtil.serializeObject(streamWithPos, typeSerializer);
			out.writeInt(streamWithPos.getPosition());
			out.write(streamWithPos.getBuf(), 0, streamWithPos.getPosition());
		}
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
public Throwable deserializeError(ClassLoader classloader) {
	if (serializedException == null) {
		// failed to serialize the original exception
		// return this SerializedThrowable as a stand in
		return this;
	}

	Throwable cached = cachedException == null ? null : cachedException.get();
	if (cached == null) {
		try {
			cached = InstantiationUtil.deserializeObject(serializedException, classloader);
			cachedException = new WeakReference<Throwable>(cached);
		}
		catch (Throwable t) {
			// something went wrong
			// return this SerializedThrowable as a stand in
			return this;
		}
	}
	return cached;
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
public void snapshotKeysForKeyGroup(DataOutputViewStreamWrapper stream, int keyGroupIdx) throws Exception {

		// we cleanup also here to avoid checkpointing the deletion set
		cleanupRegisteredKeys();

		Set<K> keySet = getRegisteredKeysForKeyGroup(keyGroupIdx);
		if (keySet != null) {
			stream.writeInt(keySet.size());

			InstantiationUtil.serializeObject(stream, keySerializer);
			for (K key : keySet) {
				keySerializer.serialize(key, stream);
			}
		} else {
			stream.writeInt(0);
		}
	}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
@Override
public Iterator<T> iterator() {
	return new Iterator<T>() {
		private int index;

		@Override
		public boolean hasNext() {
			return index < classes.size();
		}

		@Override
		public T next() {
			return InstantiationUtil.instantiate(classes.get(index++));
		}

		@Override
		public void remove() {
			throw new UnsupportedOperationException();
		}
	};
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
private void reinstantiateDriver() throws Exception {
	if (this.driver instanceof ResettableDriver) {
		final ResettableDriver<?, ?> resDriver = (ResettableDriver<?, ?>) this.driver;
		resDriver.reset();
	} else {
		Class<? extends Driver<S, OT>> driverClass = this.config.getDriver();
		this.driver = InstantiationUtil.instantiate(driverClass, Driver.class);

		try {
			this.driver.setup(this);
		}
		catch (Throwable t) {
			throw new Exception("The pact driver setup for '" + this.getEnvironment().getTaskInfo().getTaskName() +
					"' , caused an error: " + t.getMessage(), t);
		}
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
@Override
public InputSplit getNextInputSplit(ClassLoader userCodeClassLoader) throws InputSplitProviderException {
	Preconditions.checkNotNull(userCodeClassLoader);

	CompletableFuture<SerializedInputSplit> futureInputSplit = jobMasterGateway.requestNextInputSplit(
		jobVertexID,
		executionAttemptID);

	try {
		SerializedInputSplit serializedInputSplit = futureInputSplit.get(timeout.getSize(), timeout.getUnit());

		if (serializedInputSplit.isEmpty()) {
			return null;
		} else {
			return InstantiationUtil.deserializeObject(serializedInputSplit.getInputSplitData(), userCodeClassLoader);
		}
	} catch (Exception e) {
		throw new InputSplitProviderException("Requesting the next input split failed.", e);
	}
}
 

import org.apache.flink.util.InstantiationUtil; //导入依赖的package包/类
private org.apache.hadoop.fs.FileSystem instantiateFileSystem(Class<? extends org.apache.hadoop.fs.FileSystem> fsClass)
	throws IOException {
	try {
		return fsClass.newInstance();
	}
	catch (ExceptionInInitializerError e) {
		throw new IOException("The filesystem class '" + fsClass.getName() + "' throw an exception upon initialization.", e.getException());
	}
	catch (Throwable t) {
		String errorMessage = InstantiationUtil.checkForInstantiationError(fsClass);
		if (errorMessage != null) {
			throw new IOException("The filesystem class '" + fsClass.getName() + "' cannot be instantiated: " + errorMessage);
		} else {
			throw new IOException("An error occurred while instantiating the filesystem class '" +
					fsClass.getName() + "'.", t);
		}
	}
}