public static RestHighLevelClient getClient(String esAddress, Boolean isAuthMesh, String userName, String password) {
        List<HttpHost> httpHostList = new ArrayList<>();
        String[] address = StringUtils.split(esAddress, ",");
        Arrays.stream(address).forEach( addr -> {
            String[] infoArray = StringUtils.split(addr, ":");
            int port = 9200;
            String host = infoArray[0].trim();
            if (infoArray.length > 1) {
                port = Integer.parseInt(infoArray[1].trim());
            }
            httpHostList.add(new HttpHost(host, port, "http"));
        });

        RestClientBuilder restClientBuilder = RestClient.builder(httpHostList.toArray(new HttpHost[0]));

        if (isAuthMesh) {
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(userName.trim(), password.trim()));
            restClientBuilder.setHttpClientConfigCallback(httpAsyncClientBuilder ->
                    httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
        }

        RestHighLevelClient rhlClient = new RestHighLevelClient(restClientBuilder);

        if (LOG.isInfoEnabled()) {
            LOG.info("Pinging Elasticsearch cluster via hosts {}...", httpHostList);
        }

        try {
            if (!rhlClient.ping(RequestOptions.DEFAULT)) {
                throw new RuntimeException("There are no reachable Elasticsearch nodes!");
            }
        } catch (IOException e) {
            LOG.warn("", e);
        }

        if (LOG.isInfoEnabled()) {
            LOG.info("Create Elasticsearch RestHighLevelClient connected to {}.", httpHostList.toString());
        }

        return rhlClient;
    }
@description:
@program: flink.sql
@author: lany
@create: 2021/01/11 10:23
elasticsearch7/elasticsearch7-side/elasticsearch7-side-core/src/main/java/com/dtstack/flink/sql/side/elasticsearch7/util/Es7Util.java