public void avroParamCheck() {
        Preconditions.checkNotNull(schemaString, "avro type schemaInfo is required");
        if (StringUtils.equalsIgnoreCase(updateMode, EUpdateMode.UPSERT.name())) {
            Schema schema = new Schema.Parser().parse(schemaString);
            schema.getFields()
                    .stream()
                    .filter(field -> StringUtils.equalsIgnoreCase(field.name(), RETRACT_FIELD_KEY))
                    .findFirst()
                    .orElseThrow(() ->
                            new NullPointerException(String.valueOf("arvo upsert mode the retract attribute must be contained in schemaInfo field ")));
        }
    }
Date: 2018/12/18
Company: www.dtstack.com

@author DocLi
@modifyer maqi
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkTableInfo.java