我能看弗林克例子后去解决它在这里和Elasticsearch文档这里。
原来，我试图在上面设置错误的配置：
restClientBuilder.setDefaultHeaders(...);
不是实际需要设置的是：
restClientBuilder.setHttpClientConfigCallback(...);
一旦使用了正确的自定义配置，其余的就非常简单了。所以我缺少的那部分是：
// provide a RestClientFactory for custom configuration on the internally created REST clientesSinkBuilder.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); } }); });最后，这里是Elasticsearch Sink的完整片段：
List<HttpHost> httpHosts = new ArrayList<>();httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));// use a ElasticsearchSink.Builder to create an ElasticsearchSinkElasticsearchSink.Builder<ObjectNode> esSinkBuilder = new ElasticsearchSink.Builder<>( httpHosts, new ElasticsearchSinkFunction<ObjectNode>() { private IndexRequest createIndexRequest(ObjectNode payload) { // remove the value node so the fields are at the base of the json payload JsonNode jsonOutput = payload.get("value"); return Requests.indexRequest() .index("raw-payload") .type("payload") .source(jsonOutput.toString(), XContentType.JSON); } @Override public void process(ObjectNode payload, RuntimeContext ctx, RequestIndexer indexer) { indexer.add(createIndexRequest(payload)); } });// set number of events to be seen before writing to ElasticsearchesSinkBuilder.setBulkFlushMaxActions(1);// provide a RestClientFactory for custom configuration on the internally created REST clientesSinkBuilder.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 pipelinestream.addSink(esSinkBuilder.build());希望这对卡在同一地方的其他人有所帮助！
