@Override
    public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
        ElasticsearchTableInfo elasticsearchTableInfo = new ElasticsearchTableInfo();
        elasticsearchTableInfo.setName(tableName);
        parseFieldsInfo(fieldsInfo, elasticsearchTableInfo);
        elasticsearchTableInfo.setAddress((String) props.get(KEY_ES7_ADDRESS.toLowerCase()));
        elasticsearchTableInfo.setId((String) props.get(KEY_ES7_ID_FIELD_INDEX_LIST.toLowerCase()));
        elasticsearchTableInfo.setIndex((String) props.get(KEY_ES7_INDEX.toLowerCase()));
        elasticsearchTableInfo.setIndex_definition((String) props.get(KEY_ES7_INDEX_DEFINITION));
        elasticsearchTableInfo.setParallelism(MathUtil.getIntegerVal(props.getOrDefault(KEY_ES7_PARALLELISM.toLowerCase(), 1)));

        String authMeshStr = (String) props.get(KEY_ES7_AUTHMESH.toLowerCase());
        if (authMeshStr != null && StringUtils.equalsIgnoreCase(KEY_TRUE, authMeshStr)) {
            elasticsearchTableInfo.setAuthMesh(MathUtil.getBoolean(authMeshStr));
            elasticsearchTableInfo.setUserName(MathUtil.getString(props.get(KEY_ES7_USERNAME.toLowerCase())));
            elasticsearchTableInfo.setPassword(MathUtil.getString(props.get(KEY_ES7_PASSWORD.toLowerCase())));
        }
        elasticsearchTableInfo.check();
        return elasticsearchTableInfo;
    }
@description:
@program: flink.sql
@author: lany
@create: 2021/01/04 17:19
elasticsearch7/elasticsearch7-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/table/ElasticsearchSinkParser.java