Spring Webflux 和 Amazon SDK 2.x：S3AsyncClient 超时答案

【问题标题】：Spring Webflux and Amazon SDK 2.x: S3AsyncClient timeoutSpring Webflux 和 Amazon SDK 2.x：S3AsyncClient 超时
【发布时间】：2020-09-02 01:24:26
【问题描述】：

我正在使用 Spring boot 2.3.1、Webflux、带有响应式 mongodb 驱动程序和 Amazon SDk 2.14.6 的 Spring Data 实施一个响应式项目。

我有一个 CRUD，它在 MongoDB 上保留一个实体，并且必须将一个文件上传到 S3。我正在使用 SDK 反应方法s3AsyncClient.putObject，但我遇到了一些问题。 CompletableFuture 抛出以下异常：

java.util.concurrent.CompletionException: software.amazon.awssdk.core.exception.ApiCallTimeoutException: Client execution did not complete before the specified timeout configuration: 60000 millis
    at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314) ~[na:na]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoMapFuseable] :
    reactor.core.publisher.Mono.map(Mono.java:3054)
    br.com.wareline.waredrive.service.S3Service.uploadFile(S3Service.java:94)

我尝试上传的文件大约有 34kb，它是一个简单的文本文件。

上传方法在我的S3Service.java 类中，该类在DocumentoService.java

中自动装配

@Component
public class S3Service {

    @Autowired
    private final ConfiguracaoService configuracaoService;

    public Mono<PutObjectResponse> uploadFile(final HttpHeaders headers, final Flux<ByteBuffer> body, final String fileKey, final String cliente) {
        return configuracaoService.findByClienteId(cliente)
                .switchIfEmpty(Mono.error(new ResponseStatusException(HttpStatus.NOT_FOUND, String.format("Configuração com id %s não encontrada", cliente))))
                .map(configuracao -> uploadFileToS3(headers, body, fileKey, configuracao))
                .doOnSuccess(response -> {
                    checkResult(response);
                });
    }

    private PutObjectResponse uploadFileToS3(final HttpHeaders headers, final Flux<ByteBuffer> body, final String fileKey, final Configuracao configuracao) {

        final long length = headers.getContentLength();
        if (length < 0) {
            throw new UploadFailedException(HttpStatus.BAD_REQUEST.value(), Optional.of("required header missing: Content-Length"));
        }
        final Map<String, String> metadata = new HashMap<>();
        final MediaType mediaType = headers.getContentType() != null ? headers.getContentType() : MediaType.APPLICATION_OCTET_STREAM;

        final S3AsyncClient s3AsyncClient = getS3AsyncClient(configuracao);

        return s3AsyncClient.putObject(
                PutObjectRequest.builder()
                        .bucket(configuracao.getBucket())
                        .contentLength(length)
                        .key(fileKey)
                        .contentType(mediaType)
                        .metadata(metadata)
                        .build(),
                AsyncRequestBody.fromPublisher(body))
                .whenComplete((resp, err) -> s3AsyncClient.close())
                .join();
    }

    public S3AsyncClient getS3AsyncClient(final Configuracao s3Props) {

        final SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder()
            .readTimeout(Duration.ofMinutes(1))
            .writeTimeout(Duration.ofMinutes(1))
            .connectionTimeout(Duration.ofMinutes(1))
            .maxConcurrency(64)
            .build();

        final S3Configuration serviceConfiguration = S3Configuration.builder().checksumValidationEnabled(false).chunkedEncodingEnabled(true).build();

        return S3AsyncClient.builder()
            .httpClient(httpClient)
            .region(Region.of(s3Props.getRegion()))
            .credentialsProvider(() -> AwsBasicCredentials.create(s3Props.getAccessKey(), s3Props.getSecretKey()))
            .serviceConfiguration(serviceConfiguration)
            .overrideConfiguration(builder -> builder.apiCallTimeout(Duration.ofMinutes(1)).apiCallAttemptTimeout(Duration.ofMinutes(1)))
            .build();

    }

我的实现基于 Amazon SDK 文档和 https://github.com/awsdocs/aws-doc-sdk-examples/blob/master/javav2/example_code/s3/src/main/java/com/example/s3/S3AsyncOps.java 的代码示例

我无法弄清楚异步客户端超时问题的原因。奇怪的是，当我使用相同的 S3AsyncClient 从存储桶下载文件时，它可以工作。我试图将 S3AsyncClient 中的超时时间增加到大约 5 分钟，但没有成功。我不知道我做错了什么。

【问题讨论】：

不确定这是否是问题所在，但您不会被动地使用 AWS sdk。当您调用 join 时，您实际上是在阻塞一个线程。相反，您应该使用 Mono.fromFuturereturn 包装 CompletableFuture 并从 flatMap 运算符调用 uploadFileToS3 方法。
我已经尝试按照您的建议将 completableFuture 包装在 Mono.fromFuture 中，但我遇到了同样的错误。
然后，作为下一步，我将检查 Flux<ByteBuffer> body 是否实际被 aws sdk 消耗或只是挂在那里。还要检查您在上传之前/之后是否订阅了相同的通量，这可能会导致类似的问题。

标签： java spring-boot spring-webflux aws-java-sdk aws-java-sdk-2.x

【解决方案1】：

我发现了错误。当我在PutObjectRequest.builder().contentLength(length) 中定义contentLength 时，我使用headers.getContentLength()，这是整个请求的大小。在我的请求中，其他信息一起传递，使得内容长度大于实际文件长度。

我在亚马逊文档中找到了这个：

“Content-Length”标头中设置的字节数大于实际文件大小

当您向 Amazon S3 发送 HTTP 请求时，Amazon S3 期望接收 Content-Length 标头中指定的数据量。如果 Amazon S3 未收到预期的数据量，并且连接空闲 20 秒或更长时间，则连接关闭。请务必验证您的实际文件大小发送到 Amazon S3 与中指定的文件大小一致 Content-Length 标头。

https://aws.amazon.com/pt/premiumsupport/knowledge-center/s3-socket-connection-timeout-error/

发生超时错误是因为 S3 等待发送的内容长度达到客户端通知的大小，文件在达到通知的内容长度之前结束传输。然后连接保持空闲，S3 关闭套接字。

我把内容长度改成真实文件大小，上传成功。

【讨论】：