@Override
    public boolean verifyClientConnection(TransportClient client) throws IOException {


        boolean verifyResult = false;

        if (esTableInfo.isEnableKrb()) {
            //1. login kdc with keytab and krb5 conf
            UserGroupInformation ugi  = KrbUtils.loginAndReturnUgi(
                    esTableInfo.getPrincipal(),
                    esTableInfo.getKeytab(),
                    esTableInfo.getKrb5conf());

            //2. refresh availableNodes.
            verifyResult = ugi.doAs((PrivilegedAction<Boolean>) () -> {
                LOG.info("Refresh client available nodes.");
                client.refreshAvailableNodes();
                return client.connectedNodes().isEmpty();
            });
        } else {
            client.refreshAvailableNodes();
            verifyResult = client.connectedNodes().isEmpty();
        }


        if (!verifyResult) {
            return true;
        }

        // close the transportClient here
        IOUtils.closeQuietly(client);
        throw new RuntimeException("Elasticsearch client is not connected to any Elasticsearch nodes!");
    }
@date 2019/11/16
@author xiuzhu
@Company: www.dtstack.com
elasticsearch5-xh/elasticsearch5-xh-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ExtendES5ApiCallBridge.java