TransferManager parallel uploads send TRANSFER_FAILURE but continue transfering after the failure.

[jamesceddy]

When TransferManager multipart upload has part transfer failures, TRANSFER_FAILED_EVENT is sent back to the clients on the first failure. However, if later parts are successful then additional events are send back to the client. This might be alright, because at least "TRANSFER_FAILED_EVENT" was notified to the client. However I suspect here that TRANSFER_FAILED_EVENT should be a terminal event and upon this event, a client might be taking actions to correct the failure, which could be interfered with by outstanding transfers.

Steps:
(1) Start N multipart xfers.
(2) iptables DROP the connection (see code below)
(3) TRANSFER_FAILED_EVENT is received by the client

Expected: TRANSFER_FAILED_EVENT indicates the transfer is completed, and in a terminal state. No further events are expected to be received.

Actual: Further successful events are received by listeners well after the failure event.

This might be expected behavior, but it's unclear due to TRANSFER_FAILED_EVENT not being documented.

`
val uploads = mutableMapOf<String, Upload>()
val uploadLastCallbackEvents = ConcurrentHashMap<String, ProgressEventType>()
val isNetworkBlockerStarted = AtomicBoolean(false)
val fileKeys = mutableListOf(
"file42mb",
"file22mb",
"file18mb",
"file10mb"
)
fun main() {
netUp()
val accessKey = "zzzz"
val secretKey = "xyz"

val s3Client = AmazonS3ClientBuilder.standard().withRegion("us-east-1").withCredentials(object :AWSCredentialsProvider{
    override fun getCredentials(): AWSCredentials {
        return BasicAWSCredentials(accessKey, secretKey)
    }

    override fun refresh() {
    }
}).build()
ClientConfiguration().withRetryMode(RetryMode.STANDARD).withRetryPolicy(PredefinedRetryPolicies.getDefaultRetryPolicyWithCustomMaxRetries(20))
val transferManager = TransferManagerBuilder.standard()
    .withS3Client(s3Client)
    
    .withExecutorFactory { Executors.newFixedThreadPool(3) }
    .build()
println("start")

fileKeys.forEach{ file ->
    val upload = transferManager.upload(
                    new PutObjectRequest("fried-chicken",
                        File(file).name,
                        File(file)
                    ))

    upload[file] = upload
    upload.addProgressListener(ProgressListener {
        blockNetwork()
        uploadLastCallbackEvents[file] = it.eventType
        println("Progress event $it --> $file")
    })
}


while (true){
    println("\n---------------------------------------------------------------------------")
    uploads.forEach{ (key, upload) ->
        blockNetwork()
        println(" State is : ${upload.state} ==> $key  ==> ${uploadLastCallbackEvents[key]}")
    }
    println("---------------------------------------------------------------------------\n")
    TimeUnit.SECONDS.sleep(5)
}

}

fun blockNetwork() {
if (!isNetworkBlockerStarted.getAndSet(true)) {
thread(start = true) {
// give some time for transfers to start
TimeUnit.SECONDS.sleep(1)
for (i in 1..5){
netDown(i)
if (i == 5){
println("\n\n\nLAST TIME!\n\n\n")
TimeUnit.SECONDS.sleep(6)
} else {
TimeUnit.SECONDS.sleep(40)
}

            netUp()
            TimeUnit.SECONDS.sleep(2)
        }
        println("----------------------DONE ----------------------------------")
        println("----------------------DONE ----------------------------------")
        println("----------------------DONE ----------------------------------")
    }
    println("STARTED*******************************")
}

}
fun netUp() {
println("\n\n\nNETWORK UP!\n\n\n")
"/usr/bin/sudo -S /sbin/iptables -D OUTPUT -p tcp --dport 443 -j DROP".runCommand()
}
fun netDown(i:Int){
println("\n\n\nNETWORK DOWN! $i\n\n\n")
"/usr/bin/sudo -S /sbin/iptables -A OUTPUT -p tcp --dport 443 -j DROP".runCommand()
}

fun String.runCommand() {
ProcessBuilder("/bin/bash", "-c", "echo password| $this")
.redirectErrorStream(true)
.redirectOutput(ProcessBuilder.Redirect.INHERIT)
.redirectError(ProcessBuilder.Redirect.INHERIT)
.redirectInput(ProcessBuilder.Redirect.INHERIT)
.start().waitFor()
}
`

[zoewangg]

The fix is available in 1.11.750, closing the issue. Thanks again for reporting the issue and creating the PR!