@Override
    @SuppressWarnings("unchecked")
    public void configure(Map<String, ?> configs, boolean isKey) {
        String deserializerClassName;
        if(isKey){
            deserializerClassName = (String)configs.get(KafkaSourceTableInfo.DT_KEY_DESERIALIZER);
        }else{
            deserializerClassName = (String)configs.get(KafkaSourceTableInfo.DT_VALUE_DESERIALIZER);
        }
        try {
            this.deserializer = (Deserializer<T>)Class.forName(deserializerClassName).newInstance();
        } catch (Exception e) {
            throw new RuntimeException("Can't create instance: " + deserializerClassName, e);
        }
        this.deserializer.configure(configs, isKey);
    }
Date: 2021/04/20
Company: www.dtstack.com

@author tudou
kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserializer/DtKafkaDeserializer.java