ELK：Elastic异常问题记录

问题1：flink 批量写入es报错 java.io.IOException: request retries exceeded max retry timeout [30000]

基于版本 flink-connector-elasticsearch6_2.11

    org.apache.flink
    flink-connector-elasticsearch6_2.11
    1.10.1

addsink方法

public static  void addSink(List hosts, int bulkFlushMaxActions, int parallelism, int retries, long delay, int flushMaxSizeMb,
                                   DataStream dataStream, ElasticsearchSinkFunction func) {
        ElasticsearchSink.Builder esSinkBuilder = new ElasticsearchSink.Builder<>(hosts, func);
        esSinkBuilder.setBulkFlushMaxActions(bulkFlushMaxActions);
        esSinkBuilder.setBulkFlushBackoff(true);
        esSinkBuilder.setBulkFlushBackoffType(ElasticsearchSinkbase.FlushBackoffType.CONSTANT);
        esSinkBuilder.setBulkFlushBackoffRetries(retries);
        esSinkBuilder.setBulkFlushBackoffDelay(delay);
        esSinkBuilder.setBulkFlushMaxSizeMb(flushMaxSizeMb);
        dataStream.addSink(esSinkBuilder.build()).name("write es").setParallelism(parallelism);
}

错误如下

java.lang.RuntimeException: An error occurred in ElasticsearchSink.
	at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkbase.checkErrorAndRethrow(ElasticsearchSinkbase.java:381)
	at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkbase.checkAsyncErrorsAndRequests(ElasticsearchSinkbase.java:386)
	at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkbase.invoke(ElasticsearchSinkbase.java:307)
	at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56)
	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:173)
	at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:158)
	at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:135)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:69)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:320)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:187)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:496)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:479)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:789)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:594)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.IOException: request retries exceeded max retry timeout [30000]
	at org.elasticsearch.client.RestClient$1.retryIfPossible(RestClient.java:411)
	at org.elasticsearch.client.RestClient$1.failed(RestClient.java:398)
	at org.apache.http.concurrent.BasicFuture.failed(BasicFuture.java:137)
	at org.apache.http.impl.nio.client.AbstractClientExchangeHandler.failed(AbstractClientExchangeHandler.java:419)
	at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.timeout(HttpAsyncRequestExecutor.java:375)
	at org.apache.http.impl.nio.client.InternalIODispatch.onTimeout(InternalIODispatch.java:92)
	at org.apache.http.impl.nio.client.InternalIODispatch.onTimeout(InternalIODispatch.java:39)
	at org.apache.http.impl.nio.reactor.AbstractIODispatch.timeout(AbstractIODispatch.java:175)
	at org.apache.http.impl.nio.reactor.baseIOReactor.sessionTimedOut(baseIOReactor.java:263)
	at org.apache.http.impl.nio.reactor.AbstractIOReactor.timeoutCheck(AbstractIOReactor.java:492)
	at org.apache.http.impl.nio.reactor.baseIOReactor.validate(baseIOReactor.java:213)
	at org.apache.http.impl.nio.reactor.AbstractIOReactor.execute(AbstractIOReactor.java:280)
	at org.apache.http.impl.nio.reactor.baseIOReactor.execute(baseIOReactor.java:104)
	at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor$Worker.run(AbstractMultiworkerIOReactor.java:588)
	... 1 more

解决方式，添加自定义超时时间

esSinkBuilder.setRestClientFactory(new RestClientFactory() {
    @Override
    public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
        restClientBuilder.setMaxRetryTimeoutMillis(5 * 60 * 1000);
    }
});

方法可以扩展添加自定义，自定义链接方法和链接超时时间，根据实际情况采用

esSinkBuilder.setRestClientFactory(new RestClientFactory() {
    @Override
    public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
        restClientBuilder.setMaxRetryTimeoutMillis(5 * 60 * 1000);
        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                return httpAsyncClientBuilder.setDefaultIOReactorConfig(
                        IOReactorConfig.custom()
                        .setIoThreadCount(5)//设置线程数量为5
                        .build());
            }
        });
        restClientBuilder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder builder) {
                builder.setConnectTimeout(5000);//连接超时时间
                builder.setSocketTimeout(60000);
                builder.setConnectionRequestTimeout(10000);
                return builder;
            }
        });
    }
});