protected TypeInformation<Row> getRowTypeInformation(KafkaSourceTableInfo kafkaSourceTableInfo) {
        String[] fieldTypes = kafkaSourceTableInfo.getFieldTypes();
        Class<?>[] fieldClasses = kafkaSourceTableInfo.getFieldClasses();
        TypeInformation[] types =
                IntStream.range(0, fieldClasses.length)
                        .mapToObj(i -> {
                            if (fieldClasses[i].isArray()) {
                                return DataTypeUtils.convertToArray(fieldTypes[i]);
                            }
                            return TypeInformation.of(fieldClasses[i]);
                        })
                        .toArray(TypeInformation[]::new);


        return new RowTypeInfo(types, kafkaSourceTableInfo.getFields());
    }
Date: 2020/3/20
Company: www.dtstack.com
@author maqi
kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/AbstractKafkaSource.java