Webflux streaming DataBuffer with zip resulting in corrupt file

Currently I'm implementing a non blocking i/o application with Spring Boot 2.7.12 and Webflux to download files with a webclient, zip them and stream the zip file to the browser.
Downloading and zipping is working fine if I write the ZipOutputStream to a local file.
However, if I stream the zip back to the caller (Flux<DataBuffer>) it is corrupted.

I'm not sure, if I misunderstand the concept of the DataBuffer, or if it is a bug in the spring framework.

I have created a small sample. If you download the zip file, each entry is duplicated multiple times and the last entry is corrupted.

Thank you
Roberto

@GetMapping(value = "/zip", produces = "application/zip")
  public Flux<DefaultDataBuffer> zip() {
    var files = Arrays.asList("File1", "File2", "File3", "File4", "File5");
    var responseDataBuffer = new DefaultDataBufferFactory().allocateBuffer();
    ZipOutputStream zipOutputStream = new ZipOutputStream(responseDataBuffer.asOutputStream());
    return Flux.fromStream(files.stream())
        .map(file -> putZipEntry(file, zipOutputStream))
        .map(x -> responseDataBuffer)
        .doOnComplete(() -> closeZipOutputStream(zipOutputStream));
  }

  private void closeZipOutputStream(ZipOutputStream zipOutputStream) {
    try {
    zipOutputStream.close();
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  private ZipOutputStream putZipEntry(String file, ZipOutputStream zipOutputStream) {
    try {
      zipOutputStream.putNextEntry(new ZipEntry(file + ".txt"));
      zipOutputStream.write(file.getBytes());
      zipOutputStream.closeEntry();
      return zipOutputStream;
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

Why all the nonsense?

private static final String ZIP_FILE = "UEsDBAoAAAAAAN1s5lYZ+CeZBgAAAAYAAAAJAAAAdGVzdDEudHh0VGVzdCAxUEsDBAoAAAAAAPVs5lajqS4ABgAAAAYAAAAJAAAAdGVzdDIudHh0VGVzdCAyUEsDBAoAAAAAAPhs5lY1mSl3BgAAAAYAAAAJAAAAdGVzdDMudHh0VGVzdCAzUEsDBAoAAAAAAPts5laWDE3pBgAAAAYAAAAJAAAAdGVzdDQudHh0VGVzdCA0UEsDBAoAAAAAAABt5lYAPEqeBgAAAAYAAAAJAAAAdGVzdDUudHh0VGVzdCA1UEsBAj8ACgAAAAAA3WzmVhn4J5kGAAAABgAAAAkAJAAAAAAAAAAgAAAAAAAAAHRlc3QxLnR4dAoAIAAAAAAAAQAYAN4h/3L+r9kBCLkhXwKw2QEkqm9t/q/ZAVBLAQI/AAoAAAAAAPVs5lajqS4ABgAAAAYAAAAJACQAAAAAAAAAIAAAAC0AAAB0ZXN0Mi50eHQKACAAAAAAAAEAGAD2Z+2M/q/ZAVPgIV8CsNkBbwSLdP6v2QFQSwECPwAKAAAAAAD4bOZWNZkpdwYAAAAGAAAACQAkAAAAAAAAACAAAABaAAAAdGVzdDMudHh0CgAgAAAAAAABABgAGsOxkP6v2QHcByJfArDZAW2Je3b+r9kBUEsBAj8ACgAAAAAA+2zmVpYMTekGAAAABgAAAAkAJAAAAAAAAAAgAAAAhwAAAHRlc3Q0LnR4dAoAIAAAAAAAAQAYAKkIkZT+r9kBWlUiXwKw2QGkwld4/q/ZAVBLAQI/AAoAAAAAAABt5lYAPEqeBgAAAAYAAAAJACQAAAAAAAAAIAAAALQAAAB0ZXN0NS50eHQKACAAAAAAAAEAGADtvU2Y/q/ZAZ/xIl8CsNkBEj5/d/6v2QFQSwUGAAAAAAUABQDHAQAA4QAAAAAA";

private static final int BUFFER_SIZE = 10;

@GetMapping(value = "/zip", produces = "application/zip")
public Flux<DataBuffer> test() {
    return DataBufferUtils.readInputStream(() -> new ByteArrayInputStream(Base64.getDecoder().decode(ZIP_FILE)), new DefaultDataBufferFactory(), BUFFER_SIZE);
}

getFluxPublisherFunction is incorrect and logic does not make sense. What it is doing is
>> DataBufferUtils.write(source, outputStream)

source is Flux of DataBuffer from ZIP_FILE. OutputStream is another data buffer created with separate buffer (named defaultDataBuffer). This method returns same buffers as in source

>> .map(buffer -> defaultDataBuffer)

For each buffer ( chunk of ZIP_file), it returns complete destimation buffer which is getting populated using write operation. Destination buffer at any point might contain one of more parts of ZIP_FILE.

1. Why code works with 1000 buffer is because this is sufficient buffer to contain the entire contents of ZIP_FILE. getRead method in this case returns single DataBuffer when DataBufferUtils.write(source, outputStream) is subscribed. This does not work when buffer size is small because you are returning partial chunks accumulated in defaultDataBuffer.
2. To return zip to browser, simply return the Flux<DataBuffer> created by getRead method with different buffers. It will work of buffer size of 10 or 100 or 1000.

I'm not sure how Spring handles that, but to my understanding it has no knowledge of when you finish the writing and you return the buffer as soon as you have first chunk written to it. So I guess that is what the receiver sees too. In this case what you need is to return the Data Buffer only when it is fully written.

Try this:

return Flux.fromStream(files.stream())
    .map(file -> putZipEntry(file, zipOutputStream))
    .then(Mono.fromCallable { 
        closeZipOutputStream(zipOutputStream);
        responseDataBuffer
   })

PS in your case it doesn't make sense to use Reactor because you write it to a single buffer and wait until it finishes. It would be better if you produce response in chunks, but I don't see how you can do that with standard ZipOutputStream. However, since it already produces an OutputStream it would be easier just to produce it as is (with InputStreamResource and ResponseEntity)