mdedetrich
commented
2 years ago So I have figured out what the cause of the leak is (and I have updated the original post to reflect this). The leak actually wasn't due to S3 pagination but actually a peculiarity in how the S3.download function is designed, basically if you use the S3.download function to only check if an object exists which is is implied both by the API (i.e. it returns an Option in the Source, instead of throwing exception if an object doesn't exist) and by the fact that no S3.checkIfObjectExists function exists (where as there are other equivalent functions for other entities such as buckets) then a leak occurs due to the fact that you don't consume actual contents of the Source which I guess is due to the design of alpakka/akka-streams/akka-http (it creates a reference to a request which never gets consumed since you are not downloading the object?)
One peculiarity that arises is the design of the S3.download function itself. Its the only function in the Alpakka S3 API that returns an Option inside of a Source if the retrieved entity doesn't exist (see https://github.com/akka/alpakka/blob/master/s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala#L195-L197), no other S3 method does this and as mentioned before there is no equivalent S3.checkObjectExists function. An immediate step to this problem would be the creation of such a S3.checkObjectExists method (which as minor bonus could also be slightly faster since its a HEAD request rather than a GET) but the matter remains that S3.download still has such a leak for this "misuse".
In my opinion and also taking advantage of the fact that Alpakka 4.0 is a breaking release and around the corner, S3.download should either be changed and/or deprecated (via a new S3.getObject method?) so it returns a Source[ByteString, Future[ObjectMetadata]] like how FileIO.fromPath is designed rather than the currently and arguably convoluted Source[Option[(Source[ByteString, NotUsed], ObjectMetadata)], NotUsed] and just like the other methods if the S3 object doesn't happen to exist then we just throw an exception.
The other alternate option is to of course figure out and diagnose the leak while keeping the same function and not deprecating it but I suspect this would require quite a bit of time at least on my part because I have to dig into the internals.
ennru
bpiper
Versions used
Akka version: 2.6.19 Alpakka version: 3.0.4 Akka-Http version: 10.2.9
Expected Behavior
When using
S3.downloadto check the existence of an S3 object (this is reflecting the fact that the API returns an optional source and the fact that there is noS3.checkObjectExists) and you don't consume theSource(since you are only checking that it exists, you are not interested in downloading the object) there shouldn't be any leaks. An example of the code is hereActual Behavior
If you call
S3.downloadon an existing object with the ony concern for checking its existence (so you don't consume the source, rather you only check the optional value inside theSourceexists) and the object does happen to exist then a HTTP request ends up getting leaked because you never consume the source.Note that even if you increase the polling timeout using
akka.http.host-connection-pool.response-entity-subscription-timeoutfrom https://github.com/akka/akka-http/issues/1836#issuecomment-370807885 to something like 10 seconds (just to see if there is some buffering of open requests going on) it just delays the leak, not eliminate it.This can be confirmed by running an application that continuously calls
S3.listMultipartUploadwhen recovering from errors and looking at the memory usage which continuously grows over time.Relevant logs
This is just a sample of the logs showing the
WaitingForResponseEntitySubscriptionwarnings from the project behind the linked issue.Reproducible Test Case
https://github.com/mdedetrich/alpakka-s3download-memory-leak