import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@Test(expected = InvalidTypesException.class)
public void testUnboundedPojoSourceButReturnInvalidTupleType() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    DataStream<Event> input = env.addSource(new RandomEventSource(5).closeDelay(1500));

    DataStream<Tuple5<Long, Integer, String, Double, Long>> output = SiddhiCEP
        .define("inputStream", input, "id", "name", "price", "timestamp")
        .cql("from inputStream select timestamp, id, name, price insert into  outputStream")
        .returns("outputStream");

    DataStream<Long> following = output.map(new MapFunction<Tuple5<Long, Integer, String, Double, Long>, Long>() {
        @Override
        public Long map(Tuple5<Long, Integer, String, Double, Long> value) throws Exception {
            return value.f0;
        }
    });

    String resultPath = tempFolder.newFile().toURI().toString();
    following.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
    env.execute();
    assertEquals(5, getLineCount(resultPath));
    env.execute();
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private <OUT> TypeInformation<OUT> getTypeInformation(EsperSelectFunction<OUT> esperSelectFunction) {
    try {
        TypeExtractionUtils.LambdaExecutable lambdaExecutable = TypeExtractionUtils.checkAndExtractLambda(esperSelectFunction);
        if (esperSelectFunction instanceof ResultTypeQueryable) {
            return ((ResultTypeQueryable<OUT>) esperSelectFunction).getProducedType();
        }
        if (lambdaExecutable != null) {
            Type type = lambdaExecutable.getReturnType();
            return (TypeInformation<OUT>) TypeExtractor.createTypeInfo(type);
        }
        else {
            return TypeExtractor.createTypeInfo(esperSelectFunction, EsperSelectFunction.class, esperSelectFunction.getClass(), 0);
        }
    } catch (TypeExtractionException e) {
        throw new InvalidTypesException("Could not extract types.", e);
    }
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
/**
 * Creates a new PravegaDeserializationSchema using the given Pravega serializer, and the
 * type described by the type class.
 *
 * <p>Use this constructor if the produced type is not generic and can be fully described by
 * a class. If the type is generic, use the {@link #PravegaDeserializationSchema(TypeHint, Serializer)}
 * constructor instead.
 * 
 * @param typeClass  The class describing the deserialized type.
 * @param serializer The serializer to deserialize the byte messages.
 */
public PravegaDeserializationSchema(Class<T> typeClass, Serializer<T> serializer) {
    checkNotNull(typeClass);
    checkSerializer(serializer);

    this.serializer = serializer;

    try {
        this.typeInfo = TypeInformation.of(typeClass);
    } catch (InvalidTypesException e) {
        throw new IllegalArgumentException(
                "Due to Java's type erasure, the generic type information cannot be properly inferred. " + 
                "Please pass a 'TypeHint' instead of a class to describe the type. " +
                "For example, to describe 'Tuple2<String, String>' as a generic type, use " +
                "'new PravegaDeserializationSchema<>(new TypeHint<Tuple2<String, String>>(){}, serializer);'"
        );
    }
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@Test(expected = InvalidTypesException.class)
public void testUnboundedPojoSourceButReturnInvalidTupleType() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	DataStream<Event> input = env.addSource(new RandomEventSource(5).closeDelay(1500));

	DataStream<Tuple5<Long, Integer, String, Double, Long>> output = SiddhiCEP
		.define("inputStream", input, "id", "name", "price", "timestamp")
		.cql("from inputStream select timestamp, id, name, price insert into  outputStream")
		.returns("outputStream");

	DataStream<Long> following = output.map(new MapFunction<Tuple5<Long, Integer, String, Double, Long>, Long>() {
		@Override
		public Long map(Tuple5<Long, Integer, String, Double, Long> value) throws Exception {
			return value.f0;
		}
	});

	String resultPath = tempFolder.newFile().toURI().toString();
	following.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
	env.execute();
	assertEquals(5, getLineCount(resultPath));
	env.execute();
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@Test
public void testInputValidationError() {

	RichMapFunction<Writable, String> function = new RichMapFunction<Writable, String>() {
		@Override
		public String map(Writable value) throws Exception {
			return null;
		}
	};

	@SuppressWarnings("unchecked")
	TypeInformation<Writable> inType =
			(TypeInformation<Writable>) (TypeInformation<?>) new WritableTypeInfo<>(DirectWritable.class);

	try {
		TypeExtractor.getMapReturnTypes(function, inType);
		fail("exception expected");
	}
	catch (InvalidTypesException e) {
		// right
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的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.api.common.functions.InvalidTypesException; //导入依赖的package包/类
/**
 * Recursively determine all declared fields
 * This is required because class.getFields() is not returning fields defined
 * in parent classes.
 *
 * @param clazz class to be analyzed
 * @param ignoreDuplicates if true, in case of duplicate field names only the lowest one
 *                            in a hierarchy will be returned; throws an exception otherwise
 * @return list of fields
 */
@PublicEvolving
public static List<Field> getAllDeclaredFields(Class<?> clazz, boolean ignoreDuplicates) {
	List<Field> result = new ArrayList<Field>();
	while (clazz != null) {
		Field[] fields = clazz.getDeclaredFields();
		for (Field field : fields) {
			if(Modifier.isTransient(field.getModifiers()) || Modifier.isStatic(field.getModifiers())) {
				continue; // we have no use for transient or static fields
			}
			if(hasFieldWithSameName(field.getName(), result)) {
				if (ignoreDuplicates) {
					continue;
				} else {
					throw new InvalidTypesException("The field "+field+" is already contained in the hierarchy of the "+clazz+"."
						+ "Please use unique field names through your classes hierarchy");
				}
			}
			result.add(field);
		}
		clazz = clazz.getSuperclass();
	}
	return result;
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@Override
public TypeInformation<Either<L, R>> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParameters) {
	TypeInformation<?> leftType = genericParameters.get("L");
	TypeInformation<?> rightType = genericParameters.get("R");

	if (leftType == null) {
		throw new InvalidTypesException("Type extraction is not possible on Either" +
			" type as it does not contain information about the 'left' type.");
	}

	if (rightType == null) {
		throw new InvalidTypesException("Type extraction is not possible on Either" +
			" type as it does not contain information about the 'right' type.");
	}

	return new EitherTypeInfo(leftType, rightType);
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
/**
 * This method extracts the n-th type argument from the given type. An InvalidTypesException
 * is thrown if the type does not have any type arguments or if the index exceeds the number
 * of type arguments.
 *
 * @param t Type to extract the type arguments from
 * @param index Index of the type argument to extract
 * @return The extracted type argument
 * @throws InvalidTypesException if the given type does not have any type arguments or if the
 * index exceeds the number of type arguments.
 */
public static Type extractTypeArgument(Type t, int index) throws InvalidTypesException {
	if (t instanceof ParameterizedType) {
		Type[] actualTypeArguments = ((ParameterizedType) t).getActualTypeArguments();

		if (index < 0 || index >= actualTypeArguments.length) {
			throw new InvalidTypesException("Cannot extract the type argument with index " +
											index + " because the type has only " + actualTypeArguments.length +
											" type arguments.");
		} else {
			return actualTypeArguments[index];
		}
	} else {
		throw new InvalidTypesException("The given type " + t + " is not a parameterized type.");
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
/**
 * Extracts a Single Abstract Method (SAM) as defined in Java Specification (4.3.2. The Class Object,
 * 9.8 Functional Interfaces, 9.4.3 Interface Method Body) from given class.
 *
 * @param baseClass a class that is a FunctionalInterface to retrieve a SAM from
 * @throws InvalidTypesException if the given class does not implement FunctionalInterface
 * @return single abstract method of the given class
 */
public static Method getSingleAbstractMethod(Class<?> baseClass) {

	if (!baseClass.isInterface()) {
		throw new InvalidTypesException("Given class: " + baseClass + "is not a FunctionalInterface.");
	}

	Method sam = null;
	for (Method method : baseClass.getMethods()) {
		if (Modifier.isAbstract(method.getModifiers())) {
			if (sam == null) {
				sam = method;
			} else {
				throw new InvalidTypesException("Given class: " + baseClass +
					" is not a FunctionalInterface. It has more than one abstract method.");
			}
		}
	}

	if (sam == null) {
		throw new InvalidTypesException(
			"Given class: " + baseClass + " is not a FunctionalInterface. It does not have any abstract methods.");
	}

	return sam;
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testMissingTupleGenerics() {
	RichMapFunction<?, ?> function = new RichMapFunction<String, Tuple2>() {
		private static final long serialVersionUID = 1L;

		@Override
		public Tuple2 map(String value) throws Exception {
			return null;
		}
	};

	TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation) TypeInfoParser.parse("String"), "name", true);
	Assert.assertTrue(ti instanceof MissingTypeInfo);
	
	try {
		TypeExtractor.getMapReturnTypes(function, (TypeInformation) TypeInfoParser.parse("String"));
		Assert.fail("Expected an exception");
	}
	catch (InvalidTypesException e) {
		// expected
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testTupleSupertype() {
	RichMapFunction<?, ?> function = new RichMapFunction<String, Tuple>() {
		private static final long serialVersionUID = 1L;

		@Override
		public Tuple map(String value) throws Exception {
			return null;
		}
	};

	TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation) TypeInfoParser.parse("String"), "name", true);
	Assert.assertTrue(ti instanceof MissingTypeInfo);
	
	try {
		TypeExtractor.getMapReturnTypes(function, (TypeInformation) TypeInfoParser.parse("String"));
		Assert.fail("Expected an exception");
	}
	catch (InvalidTypesException e) {
		// expected
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testFunctionWithMissingGenerics() {
	RichMapFunction function = new RichMapFunction() {
		private static final long serialVersionUID = 1L;

		@Override
		public String map(Object value) throws Exception {
			return null;
		}
	};

	TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, TypeInfoParser.parse("String"), "name", true);
	Assert.assertTrue(ti instanceof MissingTypeInfo);
	
	try {
		TypeExtractor.getMapReturnTypes(function, TypeInfoParser.parse("String"));
		Assert.fail("Expected an exception");
	}
	catch (InvalidTypesException e) {
		// expected
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testValueSupertypeException() {
	RichMapFunction<?, ?> function = new RichMapFunction<StringValue, Value>() {
		private static final long serialVersionUID = 1L;

		@Override
		public Value map(StringValue value) throws Exception {
			return null;
		}
	};

	TypeInformation<?> ti =TypeExtractor.getMapReturnTypes(function, (TypeInformation)TypeInfoParser.parse("StringValue"), "name", true);
	Assert.assertTrue(ti instanceof MissingTypeInfo);
	
	try {
		TypeExtractor.getMapReturnTypes(function, (TypeInformation)TypeInfoParser.parse("StringValue"));
		Assert.fail("Expected an exception");
	}
	catch (InvalidTypesException e) {
		// expected
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@Test
public void testFunctionDependingOnInputWithTupleInputWithTypeMismatch() {
	IdentityMapper2<Boolean> function = new IdentityMapper2<Boolean>();

	TypeInformation<Tuple2<Boolean, String>> inputType = new TupleTypeInfo<Tuple2<Boolean, String>>(BasicTypeInfo.BOOLEAN_TYPE_INFO,
			BasicTypeInfo.INT_TYPE_INFO);
	
	// input is: Tuple2<Boolean, Integer>
	// allowed: Tuple2<?, String>

	try {
		TypeExtractor.getMapReturnTypes(function, inputType);
		Assert.fail("exception expected");
	} catch (InvalidTypesException e) {
		// right
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testTypeErasure() {
	TypeInformation<?> ti = TypeExtractor.getFlatMapReturnTypes(new DummyFlatMapFunction<String, Integer, String, Boolean>(), 
				(TypeInformation) TypeInfoParser.parse("Tuple2<String, Integer>"), "name", true);
	Assert.assertTrue(ti instanceof MissingTypeInfo);
	
	try {
		TypeExtractor.getFlatMapReturnTypes(new DummyFlatMapFunction<String, Integer, String, Boolean>(), 
				(TypeInformation) TypeInfoParser.parse("Tuple2<String, Integer>"));
		
		Assert.fail("Expected an exception");
	}
	catch (InvalidTypesException e) {
		// expected
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
@Test
public void testInputValidationError() {

	RichMapFunction<Writable, String> function = new RichMapFunction<Writable, String>() {
		@Override
		public String map(Writable value) throws Exception {
			return null;
		}
	};

	@SuppressWarnings("unchecked")
	TypeInformation<Writable> inType = 
			(TypeInformation<Writable>) (TypeInformation<?>) new WritableTypeInfo<>(DirectWritable.class);
	
	try {
		TypeExtractor.getMapReturnTypes(function, inType);
		fail("exception expected");
	}
	catch (InvalidTypesException e) {
		// right
	}
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
/**
 * Returns the output type of this {@code StreamTransformation} as a {@link TypeInformation}. Once
 * this is used once the output type cannot be changed anymore using {@link #setOutputType}.
 *
 * @return The output type of this {@code StreamTransformation}
 */
public TypeInformation<T> getOutputType() {
	if (outputType instanceof MissingTypeInfo) {
		MissingTypeInfo typeInfo = (MissingTypeInfo) this.outputType;
		throw new InvalidTypesException(
				"The return type of function '"
						+ typeInfo.getFunctionName()
						+ "' could not be determined automatically, due to type erasure. "
						+ "You can give type information hints by using the returns(...) "
						+ "method on the result of the transformation call, or by letting "
						+ "your function implement the 'ResultTypeQueryable' "
						+ "interface.", typeInfo.getTypeException());
	}
	typeUsed = true;
	return this.outputType;
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
public static <X> KeySelector<X, Tuple> getSelectorForKeys(Keys<X> keys, TypeInformation<X> typeInfo, ExecutionConfig executionConfig) {
	if (!(typeInfo instanceof CompositeType)) {
		throw new InvalidTypesException(
				"This key operation requires a composite type such as Tuples, POJOs, or Case Classes.");
	}

	CompositeType<X> compositeType = (CompositeType<X>) typeInfo;

	int[] logicalKeyPositions = keys.computeLogicalKeyPositions();
	int numKeyFields = logicalKeyPositions.length;

	TypeInformation<?>[] typeInfos = keys.getKeyFieldTypes();
	// use ascending order here, the code paths for that are usually a slight bit faster
	boolean[] orders = new boolean[numKeyFields];
	for (int i = 0; i < numKeyFields; i++) {
		orders[i] = true;
	}

	TypeComparator<X> comparator = compositeType.createComparator(logicalKeyPositions, orders, 0, executionConfig);
	return new ComparableKeySelector<>(comparator, numKeyFields, new TupleTypeInfo<>(typeInfos));
}
 

import org.apache.flink.api.common.functions.InvalidTypesException; //导入依赖的package包/类
public static <X, K> KeySelector<X, K> getSelectorForOneKey(
		Keys<X> keys, Partitioner<K> partitioner, TypeInformation<X> typeInfo, ExecutionConfig executionConfig) {
	if (!(typeInfo instanceof CompositeType)) {
		throw new InvalidTypesException(
				"This key operation requires a composite type such as Tuples, POJOs, case classes, etc");
	}
	if (partitioner != null) {
		keys.validateCustomPartitioner(partitioner, null);
	}

	CompositeType<X> compositeType = (CompositeType<X>) typeInfo;
	int[] logicalKeyPositions = keys.computeLogicalKeyPositions();
	if (logicalKeyPositions.length != 1) {
		throw new IllegalArgumentException("There must be exactly 1 key specified");
	}

	TypeComparator<X> comparator = compositeType.createComparator(
			logicalKeyPositions, new boolean[] { true }, 0, executionConfig);
	return new OneKeySelector<>(comparator);
}