DataBufferUtils does not release DataBuffer on error cases [SPR-16782]

[spring-projects-issues]

Brian Clozel opened SPR-16782 and commented

DataBufferUtils can write DataBuffer streams to file channels; the current API does leave full control to the developer and the javadoc is pretty clear on the fact that developers are in charge of releasing DataBuffer instances.

Now let's take an example; we'd like to fetch a large file using the WebClient and pipe it into a file on disk. An initial approach could be:

Flux<DataBuffer> data = WebClient.create().get()
    .uri("http://example.org/largefile.bin")
    .retrieve()
    .bodyToFlux(DataBuffer.class);
  
// We'd like to write it to a file on disk
Path file = Files.createTempFile("spring", null);
WritableByteChannel channel = Files.newByteChannel(file, StandardOpenOption.WRITE);

DataBufferUtils.write(data, channel)
      // the release consumer releases buffer written on disk, one by one
      .subscribe(DataBufferUtils.releaseConsumer(), throwable -> {
        // when an error occurs, we don't have access to the current buffer
        // nor the following ones: we can't release anything properly.
      });  
}

As explained in the comments, if an IOException happens the current DataBuffer is not released, neither are the ones that might be instantiated already and about to be published to the pipeline.

I think we should document this or assist developers here, as there is no obvious workaround here. DataBufferUtils is an utils class, so we need to make sure to strike the right balance between doing the right thing automatically and providing flexibility.

---

Issue Links:

• DataBufferUtils#join could leak buffers in case of error from the source [SPR-17025] #21563 DataBufferUtils#join could leak buffers in case of error from the source
• Review DataBuffer handling code for proper release in case of error or cancellation [SPR-17408] #21941 Review DataBuffer handling code for proper release in case of error or cancellation

Referenced from: commits 1a0522b, 952315c

Backported to: 5.0.10

4 votes, 4 watchers

[spring-projects-issues]

Deryl Spielman commented

Per your suggestion I am using this and indeed noticing the channel is not closed. If you do a try-with-resources like below you get a ClosedChannelException

try (WritableByteChannel channel = Files.newByteChannel(file, StandardOpenOption.WRITE);) { // ...

 

[spring-projects-issues]

Brian Clozel commented

Deryl Spielman this code snippet is indeed incomplete, as the channel needs to be properly closed in a callback, such as Flux.doOnTerminate.

[spring-projects-issues]

Arjen Poutsma commented

Fixed by making sure that any received data buffers are returned as part of the returned flux, even when an error occurs or is received. As a result, the DataBufferUtils.releaseConsumer() can be used to release these "error buffers" as well.

See 1a0522b

[spring-projects-issues]

Arjen Poutsma commented

Reopening for back port to 5.0.x

[spring-projects-issues]

Arjen Poutsma commented

Backported in 952315c