protected TypeInformation[] getTypeInformations(KafkaSinkTableInfo kafka11SinkTableInfo) {
        String[] fieldTypes = kafka11SinkTableInfo.getFieldTypes();
        Class<?>[] fieldClasses = kafka11SinkTableInfo.getFieldClasses();
        TypeInformation[] types = IntStream.range(0, fieldClasses.length)
                .mapToObj(
                    i -> {
                        if (fieldClasses[i].isArray()) {
                            return DataTypeUtils.convertToArray(fieldTypes[i]);
                        }
                        if (fieldClasses[i] == new HashMap().getClass()) {
                            return DataTypeUtils.convertToMap(fieldTypes[i]);
                        }
                        return TypeInformation.of(fieldClasses[i]);
                    })
                .toArray(TypeInformation[]::new);
        return types;
    }
Date: 2020/3/30
Company: www.dtstack.com
@author maqi
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/AbstractKafkaSink.java