使用 Spring Reactive Elasticsearch Client 获取错误响应正文

【问题标题】：Get body of error response with Spring Reactive Elasticsearch Client使用 Spring Reactive Elasticsearch Client 获取错误响应正文
【发布时间】：2020-06-09 17:43:47
【问题描述】：

我正在使用 Spring Data Elasticsearch 来获取一些聚合，并且我正在尝试使用 ReactiveElasticsearchClient。我从 Elasticsearch 收到 500 错误，但不知道如何获取响应正文，以便调试我的请求有什么问题。到目前为止，这是我所拥有的：

    final Flux<Aggregation> resp = client.aggregate(request -> {
        final SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.queryStringQuery(query)).size(0)
                .aggregation(AggregationBuilders.composite("report", compositeValueBuilders));

        request.indices(index).source(sourceBuilder);

    });

    resp.doOnError(throwable -> {
        throwable.printStackTrace();
    }).log().doOnEach(signal -> {
        if (signal.hasError()) {
            signal.getThrowable().printStackTrace();
        } else {
            final Aggregation agg = signal.get();
            System.out.println(agg.getType());
            System.out.println(agg.getClass());
            System.out.println(agg.getName());
        }
    }).blockLast();

为 org.springframework.data.elasticsearch.client.WIRE 启用跟踪日志记录后，会记录请求正文，但响应的日志行是这样的：

2020-06-09 14:12:46.625 TRACE 19999 --- [or-http-epoll-1] os.data.elasticsearch.client.WIRE : [4ed5a037] 收到的原始响应：500 INTERNAL_SERVER_ERROR

我得到以下堆栈跟踪。我可以看到ElasticsearchStatusException，但我没有看到我可以使用Signal 对象来获取响应正文。

org.elasticsearch.ElasticsearchStatusException: POST request to /<index name>/_search returned error code 500.
    at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.handleServerError(DefaultReactiveElasticsearchClient.java:809) ~[spring-data-elasticsearch-4.0.0.RELEASE.jar:4.0.0.RELEASE]
    at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.readResponseBody(DefaultReactiveElasticsearchClient.java:760) ~[spring-data-elasticsearch-4.0.0.RELEASE.jar:4.0.0.RELEASE]
    at org.springframework.data.elasticsearch.client.reactive.DefaultReactiveElasticsearchClient.lambda$sendRequest$22(DefaultReactiveElasticsearchClient.java:680) ~[spring-data-elasticsearch-4.0.0.RELEASE.jar:4.0.0.RELEASE]
    at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onNext(MonoFlatMapMany.java:156) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1755) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.onNext(FluxOnAssembly.java:385) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:173) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:213) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:845) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:845) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2317) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onSubscribeInner(MonoFlatMapMany.java:143) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onNext(MonoFlatMapMany.java:182) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.FluxRetryPredicate$RetryPredicateSubscriber.onNext(FluxRetryPredicate.java:82) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:156) ~[reactor-core-3.3.5.RELEASE.jar:3.3.5.RELEASE]
    at reactor.netty.http.client.HttpClientConnect$HttpIOHandlerObserver.onStateChange(HttpClientConnect.java:428) ~[reactor-netty-0.9.7.RELEASE.jar:0.9.7.RELEASE]
    at reactor.netty.ReactorNetty$CompositeConnectionObserver.onStateChange(ReactorNetty.java:514) ~[reactor-netty-0.9.7.RELEASE.jar:0.9.7.RELEASE]
    at reactor.netty.resources.PooledConnectionProvider$DisposableAcquire.onStateChange(PooledConnectionProvider.java:536) ~[reactor-netty-0.9.7.RELEASE.jar:0.9.7.RELEASE]
    at reactor.netty.resources.PooledConnectionProvider$PooledConnection.onStateChange(PooledConnectionProvider.java:427) ~[reactor-netty-0.9.7.RELEASE.jar:0.9.7.RELEASE]
    at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:562) ~[reactor-netty-0.9.7.RELEASE.jar:0.9.7.RELEASE]
    at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:96) ~[reactor-netty-0.9.7.RELEASE.jar:0.9.7.RELEASE]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286) ~[netty-handler-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324) ~[netty-codec-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:311) ~[netty-codec-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:425) ~[netty-codec-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276) ~[netty-codec-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) ~[netty-transport-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792) ~[netty-transport-native-epoll-4.1.49.Final-linux-x86_64.jar:4.1.49.Final]
    at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:475) ~[netty-transport-native-epoll-4.1.49.Final-linux-x86_64.jar:4.1.49.Final]
    at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378) ~[netty-transport-native-epoll-4.1.49.Final-linux-x86_64.jar:4.1.49.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) ~[netty-common-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.49.Final.jar:4.1.49.Final]
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.49.Final.jar:4.1.49.Final]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

Tl;dr：使用 Spring ReactiveElasticsearchClient 时如何获取错误响应的响应正文？

【问题讨论】：

    标签： spring-data-elasticsearch


    【解决方案1】：

    你应该activate the logging通过设置日志级别

    org.springframework.data.elasticsearch.client.WIRE

    到跟踪级别。这会记录发送和接收的内容，尤其是在出现错误的情况下（摘自实现）：

    private <T> Publisher<? extends T> readResponseBody(String logId, Request request, ClientResponse response,
        Class<T> responseType) {

    // ...

    if (response.statusCode().is5xxServerError()) {

        ClientLogger.logRawResponse(logId, response.statusCode());
        return handleServerError(request, response);
    }

    if (response.statusCode().is4xxClientError()) {

        ClientLogger.logRawResponse(logId, response.statusCode());
        return handleClientError(logId, request, response, responseType);
    }

    // ...
}

    编辑：

    对不起，我的错，应该看到响应正文没有被记录。

    您可以做的是使用拦截代理来监控您的应用程序和 Elasticsearch 之间的流量。有OWASP ZAP proxy 和Burp Suite Community edition。两个应用程序都作为端口 8080 上的代理启动，因此您需要将应用程序配置为侦听不同的端口（应用程序属性中的server.port，例如我的测试应用程序使用 9090）或更改代理端口。除了您需要在您的应用程序中配置代理之外，我还有以下配置：

    @Configuration
public class ReactiveRestClientConfig extends AbstractReactiveElasticsearchConfiguration {
    @Override
    public ReactiveElasticsearchClient reactiveElasticsearchClient() {
        ClientConfiguration clientConfiguration = ClientConfiguration.builder() //
            .connectedTo("localhost:9200") //
            .withProxy("localhost:8080")
            .build();
        return ReactiveRestClients.create(clientConfiguration);

    }
}

    访问您的应用程序时，Elasticsearch 的流量会通过代理，您可以在其中看到请求和响应。

    【讨论】：

    • 我可以看到我的请求正文，但对于响应它只是说“[4ed5a037] Received raw response: 500 INTERNAL_SERVER_ERROR”
    • 查看我的更新答案，希望对您有所帮助，如有必要，请随时询问更多信息
    • 不幸的是，这是唯一的选择 - 对我来说有太多的障碍，特别是对于生产系统，所以我将切换到 RestHighLevelClient，但这对于其他尝试的人来说绝对有用使响应式客户端工作。
    • 在非反应性堆栈上使用相同的查询应该会给你同样的错误，因为 500 来自 Elasticsearch
    • 我将添加一张票，以便在错误情况下也记录响应正文
    猜你喜欢
    • 2019-03-23
    • 1970-01-01
    • 2018-09-04
    • 2012-05-10
    • 2016-04-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode