Apache Flink (v1.6.0) аутентифицирует Elasticsearch Sink (v6.4)

Question

Я использую Apache Flink v1.6.0 и пытаюсь записать в Elasticsearch v6.4.0, который размещен в Elastic Cloud . У меня проблема при аутентификации в кластере Elastic Cloud.

Мне удалось заставить 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());

Однако, когда я пытаюсь добавить аутентификацию в базу кода, как указано здесь в документации Flink и здесь в соответствующей документации 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)

Может кто-нибудь помочь указать, где я иду не так?

Answer 1

Я смог разобраться с этим после просмотра примера Flink здесь и документации 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());

Надеюсь, это поможет любому, кто застрял в том же месте!

Answer 2

override def configureRestClientBuilder(restClientBuilder: RestClientBuilder): Unit = {
        // TODO Additional rest client args go here - authentication headers for secure connections etc...
      }
    })

Надеюсь, это поможет вам.