我正在使用Apache Flink v1.6.0，并且尝试写到托管在Elastic Cloud中的 Elasticsearch v6.4.0 。向弹性云集群进行身份验证时出现问题。
我已经能够让Flink写入本地Elasticsearch v6.4.0节点，该节点没有使用以下代码进行加密：
/* Elasticsearch Configuration */ List<HttpHost> httpHosts = new ArrayList<>(); httpHosts.add(new HttpHost("127.0.0.1", 9200, "http")); // use a ElasticsearchSink.Builder to create an ElasticsearchSink ElasticsearchSink.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 Elasticsearch esSinkBuilder.setBulkFlushMaxActions(1); // finally, build and add the sink to the job's pipeline stream.addSink(esSinkBuilder.build());
然而，当我尝试添加验证到代码库中，作为记录在这里的弗林克文档和这里对应Elasticsearch Java文档上。看起来像这样：
// provide a RestClientFactory for custom configuration on the internally created REST client Header[] defaultHeaders = new Header[]{new BasicHeader("username", "password")}; esSinkBuilder.setRestClientFactory( restClientBuilder -> { restClientBuilder.setDefaultHeaders(defaultHeaders); } );
执行作业时出现以下错误：
14:49:54,700 INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService - Stopped Akka RPC service. Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: org.elasticsearch.ElasticsearchStatusException: method [HEAD], host [https://XXXXXXXXXXXXXX.europe-west1.gcp.cloud.es.io:9243], URI [/], status line [HTTP/1.1 401 Unauthorized] at org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:623) at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:123) at com.downuk.AverageStockSalePrice.main(AverageStockSalePrice.java:146) Caused by: org.elasticsearch.ElasticsearchStatusException: method [HEAD], host [https://XXXXXXXXXXXXXX.europe-west1.gcp.cloud.es.io:9243], URI [/], status line [HTTP/1.1 401 Unauthorized] at org.elasticsearch.client.RestHighLevelClient.parseResponseException(RestHighLevelClient.java:625)
谁能帮我指出我要去哪里了？
我能看弗林克例子后去解决它在这里和Elasticsearch文档这里。
原来，我试图在上面设置错误的配置：
restClientBuilder.setDefaultHeaders(...);
不是实际需要设置的是：
restClientBuilder.setHttpClientConfigCallback(...);
一旦使用了正确的自定义配置，其余的就非常简单了。所以我缺少的那部分是：
// 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); } }); } );
最后，这里是Elasticsearch Sink的完整片段：
/* Elasticsearch Configuration */ List<HttpHost> httpHosts = new ArrayList<>(); httpHosts.add(new HttpHost("127.0.0.1", 9200, "http")); // use a ElasticsearchSink.Builder to create an ElasticsearchSink ElasticsearchSink.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 Elasticsearch 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 stream.addSink(esSinkBuilder.build());
希望这对卡在同一地方的其他人有所帮助！
