flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java
/**
	 * Recursively converts extracted AvroTypeInfo into a RowTypeInfo nested structure with deterministic field order.
	 * Replaces generic Utf8 with basic String type information.
	 */
Recursively converts extracted AvroTypeInfo into a RowTypeInfo nested structure with deterministic field order.
Replaces generic Utf8 with basic String type information.
Recursively converts extracted AvroTypeInfo into a RowTypeInfo nested structure with deterministic field order . Replaces generic Utf8 with basic String type information .
private static TypeInformation<?> convertToTypeInformation(TypeInformation<?> extracted, Schema schema) {
		if (schema.getType() == Schema.Type.RECORD) {
			final List<Schema.Field> fields = schema.getFields();
			final AvroTypeInfo<?> avroTypeInfo = (AvroTypeInfo<?>) extracted;
			final TypeInformation<?>[] types = new TypeInformation<?>[fields.size()];
			final String[] names = new String[fields.size()];
			for (int i = 0; i < fields.size(); i++) {
				final Schema.Field field = fields.get(i);
				types[i] = convertToTypeInformation(avroTypeInfo.getTypeAt(field.name()), field.schema());
				names[i] = field.name();
			}
			return new RowTypeInfo(types, names);
		} else if (extracted instanceof GenericTypeInfo<?>) {
			final GenericTypeInfo<?> genericTypeInfo = (GenericTypeInfo<?>) extracted;
			if (genericTypeInfo.getTypeClass() == Utf8.class) {
				return BasicTypeInfo.STRING_TYPE_INFO;
			}
		}
		return extracted;
	}
private static TypeInformation < ? > convertToTypeInformation ( TypeInformation < ? > extracted , Schema schema ) { if ( schema . getType ( ) == Schema . Type . RECORD ) { final List < Schema . Field > fields = schema . getFields ( ) ; final AvroTypeInfo < ? > avroTypeInfo = ( AvroTypeInfo < ? > ) extracted ; final TypeInformation < ? > [ ] types = new TypeInformation < ? > [ fields . size ( ) ] ; final String [ ] names = new String [ fields . size ( ) ] ; for ( int i = 0 ; i < fields . size ( ) ; i ++ ) { final Schema . Field field = fields . get ( i ) ; types [ i ] = convertToTypeInformation ( avroTypeInfo . getTypeAt ( field . name ( ) ) , field . schema ( ) ) ; names [ i ] = field . name ( ) ; } return new RowTypeInfo ( types , names ) ; } else if ( extracted instanceof GenericTypeInfo < ? > ) { final GenericTypeInfo < ? > genericTypeInfo = ( GenericTypeInfo < ? > ) extracted ; if ( genericTypeInfo . getTypeClass ( ) == Utf8 . class ) { return BasicTypeInfo . STRING_TYPE_INFO ; } } return extracted ; }
Recursively converts extracted AvroTypeInfo into a RowTypeInfo nested structure with deterministic field order.
Recursively converts extracted AvroTypeInfo into a RowTypeInfo nested structure with deterministic field order .