Flink (v1.6) Elasticsearch (v6.x) Sink -- with auth

es-sink-flink.java

// -- ** ELASTICSEARCH

List<HttpHost> httpHost = new ArrayList<>();
httpHost.add(new HttpHost("127.0.0.1", 9200, "http"));

ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
        httpHost,
        new ElasticsearchSinkFunction<String>() {
            public IndexRequest createIndexRequest(String element) {
//                        Map<String, String> json = new HashMap<>();
//                        json.put("data", element);

//                        System.out.println(json);

                return Requests.indexRequest()
                        .index("tweets")
                        .type("tweet")
                        .source(element, XContentType.JSON);
            }

            @Override
            public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
                indexer.add(createIndexRequest(element));
            }
        }
);

// configuration for the bulk requests; this instructs the sink to emit after every element, otherwise they would be buffered
esSinkBuilder.setBulkFlushMaxActions(1);


// provide a RestClientFactory for custom configuration on the internally created REST client
esSinkBuilder.setRestClientFactory(
        restClientBuilder -> {
            restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {

                    // elasticsearch username and password
                    CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("$USERNAME", "$PASSWORD"));

                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
            });
        }
);

// finally, build and add the sink to the job's pipeline
jsonTweets.addSink(esSinkBuilder.build());

Additional info can be found on StackOverflow here