@Override
    public Map<String, String> tableOptions() {
        Map<String, String> options = super.tableOptions();
        options.put("connector", "elasticsearch-7-inlong");
        if (version == 5 || version == 6) {
            options.put("connector", "elasticsearch-6-inlong");
            options.put("document-type", documentType);
        }
        options.put("hosts", hosts);
        options.put("index", index);
        options.put("password", password);
        options.put("username", username);
        return options;
    }
if you want to set field routing, set the routing.field-name
The elasticsearch-6-inlong Connector is compatible with elasticsearch versions 5 and 6
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/ElasticsearchLoadNode.java