s3 分段上传总是在第二部分失败并超时

Question

我正在尝试使用基于the documentation 的亚马逊 s3 客户端在 Kotlin 中进行简单的概念验证多部分上传。第一部分上传成功，我收到带有 etag 的响应。第二部分没有上传任何东西并且超时。在第一部分之后它总是失败。是否有一些我需要以某种方式手动进行的连接清理？

凭据和权限都很好。下面的神奇数字只是为了达到 5MB 的最小部分大小。

我在这里做错了什么？

fun main() {
    val amazonS3 =
        AmazonS3ClientBuilder.standard().withRegion(Regions.EU_WEST_1).withCredentials(ProfileCredentialsProvider())
            .build()


    val bucket = "io.inbot.sandbox"
    val key = "test.txt"
    val multipartUpload =
        amazonS3.initiateMultipartUpload(InitiateMultipartUploadRequest(bucket, key))

    var pn=1
    var off=0L
    val etags = mutableListOf<PartETag>()

    for( i in 0.rangeTo(5)) {

        val buf = ByteArrayOutputStream()
        val writer = buf.writer().buffered()
        for(l in 0.rangeTo(100000)) {
            writer.write("part $i - Hello world for the $l'th time this part.\n")
        }
        writer.flush()
        writer.close()

        val bytes = buf.toByteArray()


        val md = MessageDigest.getInstance("MD5")
        md.update(bytes)
        val md5 = Base64.encodeBytes(md.digest())
        println("going to write ${bytes.size}")
        bytes.inputStream()
        var partRequest = UploadPartRequest().withBucketName(bucket).withKey(key)
            .withUploadId(multipartUpload.uploadId)
            .withFileOffset(off)
            .withPartSize(bytes.size.toLong())
            .withPartNumber(pn++)
            .withMD5Digest(md5)
            .withInputStream(bytes.inputStream())
            .withGeneralProgressListener<UploadPartRequest> { it ->
                println(it.bytesTransferred)
            }
        if(i == 5) {
            partRequest = partRequest.withLastPart(true)
        }

        off+=bytes.size

        val partResponse = amazonS3.uploadPart(partRequest)

        etags.add(partResponse.partETag)
        println("part ${partResponse.partNumber} ${partResponse.eTag} ${bytes.size}")

    }
    val completeMultipartUpload =
        amazonS3.completeMultipartUpload(CompleteMultipartUploadRequest(bucket, key, multipartUpload.uploadId, etags))


}

这总是在第二部分失败

Exception in thread "main" com.amazonaws.services.s3.model.AmazonS3Exception: Your socket connection to the server was not read from or written to within the timeout period. Idle connections will be closed. (Service: Amazon S3; Status Code: 400; Error Code: RequestTimeout; Request ID: F419872A24BB5526; S3 Extended Request ID: 48XWljQNuOH6LJG9Z85NJOGVy4iv/ru44Ai8hxEP+P+nqHECXZwWNwBoMyjiQfxKpr6icGFjxYc=), S3 Extended Request ID: 48XWljQNuOH6LJG9Z85NJOGVy4iv/ru44Ai8hxEP+P+nqHECXZwWNwBoMyjiQfxKpr6icGFjxYc=
    at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1630)

只是为了抢占一些我不想要的答案，我这样做的目的不是上传文件，而是最终能够通过简单地上传部分直到完成然后组合它们来将任意长度的流传输到 s3。所以，我不能真正使用 TransferManager，因为这需要我提前知道大小，而我不会。此外，我不想将其缓冲为文件，因为这将在 dockerized 服务器应用程序中运行。所以我真的很想上传任意数量的部分。我很高兴按顺序进行；虽然我不介意并行性。

我也使用过“com.github.alexmojaki:s3-stream-upload:1.0.1”，但这似乎在内存中保留了很多状态（我已经用完了几次），所以我'想用更简单的东西代替它。

更新。在下面的 cmets 中感谢 ilya。删除 withFileOffset 可以解决问题。

Answer 1

删除 withFileOffset 可以解决问题。感谢@Ilya 指出这一点。

这是我实现的一个简单的输出流，它确实有效。

package io.inbot.aws

import com.amazonaws.auth.profile.ProfileCredentialsProvider
import com.amazonaws.regions.Regions
import com.amazonaws.services.s3.AmazonS3
import com.amazonaws.services.s3.AmazonS3ClientBuilder
import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest
import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest
import com.amazonaws.services.s3.model.InitiateMultipartUploadResult
import com.amazonaws.services.s3.model.PartETag
import com.amazonaws.services.s3.model.UploadPartRequest
import mu.KotlinLogging
import java.io.ByteArrayOutputStream
import java.io.OutputStream
import java.security.MessageDigest
import java.util.Base64

private val logger = KotlinLogging.logger {  }
class S3Writer(
    private val amazonS3: AmazonS3,
    private val bucket: String,
    private val key: String,
    private val threshold: Int = 5*1024*1024
) : OutputStream(), AutoCloseable {

    private val etags: MutableList<PartETag> = mutableListOf()

    private val multipartUpload: InitiateMultipartUploadResult = this.amazonS3.initiateMultipartUpload(InitiateMultipartUploadRequest(bucket, key))

    private val currentPart = ByteArrayOutputStream(threshold)

    private var partNumber = 1

    override fun write(b: Int) {
        currentPart.write(b)
        if(currentPart.size() > threshold) {
            sendPart()
        }
    }

    private fun sendPart(last: Boolean = false) {
        logger.info { "sending part $partNumber" }
        currentPart.flush()

        val bytes = currentPart.toByteArray()

        val md = MessageDigest.getInstance("MD5")
        md.update(bytes)
        val md5 = Base64.getEncoder().encode(md.digest())
        var partRequest = UploadPartRequest().withBucketName(bucket).withKey(key)
            .withUploadId(multipartUpload.uploadId)
            .withPartSize(currentPart.size().toLong())
            .withPartNumber(partNumber++)
            .withMD5Digest(md5.contentToString())
            .withInputStream(bytes.inputStream())

        if(last) {
            logger.info { "final part" }
            partRequest = partRequest.withLastPart(true)
        }

        val partResponse = amazonS3.uploadPart(partRequest)

        etags.add(partResponse.partETag)

        currentPart.reset()

    }


    override fun close() {
        if(currentPart.size() > 0) {
            sendPart(true)
        }
        logger.info { "completing" }
        amazonS3.completeMultipartUpload(CompleteMultipartUploadRequest(bucket, key, multipartUpload.uploadId, etags))
    }

}


fun main() {
    val amazonS3 =
        AmazonS3ClientBuilder.standard().withRegion(Regions.EU_WEST_1).withCredentials(ProfileCredentialsProvider())
            .build()

    val bucket = "io.inbot.sandbox"
    val key = "test.txt"

    try {
        S3Writer(amazonS3, bucket, key).use {
            val w = it.bufferedWriter()
            for (i in 0.rangeTo(1000000)) {
                w.write("Line $i: hello again ...\n")
            }
        }
    } catch (e: Throwable) {
        logger.error(e.message,e)
    }
}