@Override
    public TransportClient createClient(Map<String, String> clientConfig) throws IOException{

        TransportClient transportClient;

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

            //2. set transwarp attributes
            Settings settings = Settings.builder().put(clientConfig)
                    .put("client.transport.sniff", true)
                    .put("security.enable", true)
                    .put(NetworkModule.TRANSPORT_TYPE_KEY, "security-netty3")
                    .build();

            //3. build transport client with transwarp plugins
            transportClient = ugi.doAs((PrivilegedAction<TransportClient>) () -> {
                TransportClient tmpClient = new PreBuiltTransportClient(settings,
                        Collections.singletonList(DoorKeeperClientPlugin.class));
                for (TransportAddress transport : ElasticsearchUtils.convertInetSocketAddresses(transportAddresses)) {
                    tmpClient.addTransportAddress(transport);
                }
                return tmpClient;
            });
        } else {
            Settings settings = Settings.builder().put(clientConfig)
                    .put("client.transport.sniff", true)
                    .build();

            transportClient = new PreBuiltTransportClient(settings);
            for (TransportAddress transport : ElasticsearchUtils.convertInetSocketAddresses(transportAddresses)) {
                transportClient.addTransportAddress(transport);
            }
        }

        return transportClient;
    }
@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