Non-deterministic "Body token not expected" in org.springframework.http.codec.multipart.PartGenerator

[giraone]

We are facing an error Body token not expected in in org.springframework.http.codec.multipart.PartGenerator
with Spring Boot 3.5.13/ Spring Framework 6.2.17 AND Spring Boot 4.0.5 / Spring Framework 7.0.6, when parsing a multipart/form-data response with larger
file parts (FilePart), that are stored in the TMP directory.

Error and StackTrace (from a Spring Boot CLI application running against a production service):

java.lang.IllegalStateException: Body token not expected
        at org.springframework.http.codec.multipart.PartGenerator$CreateFileState.body(PartGenerator.java:457)
        Suppressed: The stacktrace has been enhanced by Reactor, refer to additional information below: 
Error has been observed at the following site(s):
        *__checkpoint ⇢ Body from POST https://int-ediio-unstable-eric-submission-v43460.pcfsec.dev.datev.de/api/submit/UStVA_2025 [DefaultClientResponse]
Original Stack Trace:
                at java.lang.IllegalStateException: Body token not expected
        at org.springframework.http.codec.multipart.PartGenerator$CreateFileState.body(PartGenerator.java:457)
        Suppressed: The stacktrace has been enhanced by Reactor, refer to additional information below: 
Error has been observed at the following site(s):
        *__checkpoint ⇢ Body from POST https://int-ediio-unstable-eric-submission-v43460.pcfsec.dev.datev.de/api/submit/UStVA_2025 [DefaultClientResponse]
Original Stack Trace:
                at org.springframework.http.codec.multipart.PartGenerator$CreateFileState.body(PartGenerator.java:457)
                at org.springframework.http.codec.multipart.PartGenerator.hookOnNext(PartGenerator.java:126)
                at org.springframework.http.codec.multipart.PartGenerator.hookOnNext(PartGenerator.java:61)
                at reactor.core.publisher.BaseSubscriber.onNext(BaseSubscriber.java:163)
                at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:109)
                at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:679)
                at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:757)
                at reactor.core.publisher.FluxWindowPredicate$WindowFlux.request(FluxWindowPredicate.java:844)
                at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.request(FluxContextWrite.java:138)
                at reactor.core.publisher.BaseSubscriber.request(BaseSubscriber.java:217)
                at org.springframework.http.codec.multipart.PartGenerator.requestToken(PartGenerator.java:201)
                at org.springframework.http.codec.multipart.PartGenerator.lambda$createPart$1(PartGenerator.java:102)
                at reactor.core.publisher.MonoCreate$DefaultMonoSink.request(MonoCreate.java:285)
                at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.request(Operators.java:2325)
                at reactor.core.publisher.FluxConcatMapNoPrefetch$FluxConcatMapNoPrefetchSubscriber.request(FluxConcatMapNoPrefetch.java:338)
                at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:795)
                at reactor.core.publisher.FluxFlatMap$FlatMapMain.innerComplete(FluxFlatMap.java:899)
                at reactor.core.publisher.FluxFlatMap$FlatMapInner.onComplete(FluxFlatMap.java:1004)
                at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.onComplete(FluxDoFinally.java:128)
                at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onComplete(FluxMapFuseable.java:153)
                at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onComplete(MonoPeekTerminal.java:303)
                at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.complete(MonoIgnoreThen.java:298)
                at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onNext(MonoIgnoreThen.java:191)
                at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:240)
                at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onComplete(MonoIgnoreThen.java:207)
                at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onComplete(MonoPeekTerminal.java:303)
                at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onComplete(MonoSubscribeOn.java:163)
                at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:145)
                at org.springframework.http.codec.multipart.DefaultParts$FileContent.lambda$blockingOperation$0(DefaultParts.java:317)
                at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:61)
                at reactor.core.publisher.Mono.subscribe(Mono.java:4569)
                at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:127)
                at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:90)
                at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
                at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
                at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
                at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
                at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
                at java.base/java.lang.Thread.run(Thread.java:1583)
        Suppressed: java.lang.Exception: #block terminated with an error
                at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:104)
                at reactor.core.publisher.Mono.block(Mono.java:1773)
                at de.datev.ediio.eric.client.service.EricSubmissionClientCli.callApi(EricSubmissionClientCli.java:81)
                at de.datev.ediio.eric.client.ClientApplication.lambda$run$0(ClientApplication.java:26)
                at org.springframework.boot.SpringApplication.lambda$callRunner$1(SpringApplication.java:792)
                at org.springframework.util.function.ThrowingConsumer$1.acceptWithException(ThrowingConsumer.java:82)
                at org.springframework.util.function.ThrowingConsumer.accept(ThrowingConsumer.java:60)
                at org.springframework.util.function.ThrowingConsumer$1.accept(ThrowingConsumer.java:86)
                at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:800)
                at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:791)
                at org.springframework.boot.SpringApplication.lambda$callRunners$0(SpringApplication.java:776)
                at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
                at java.base/java.util.stream.SortedOps$SizedRefSortingSink.end(SortedOps.java:357)
                at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:510)
                at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
                at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
                at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
                at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
                at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596)
                at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:776)
                at org.springframework.boot.SpringApplication.run(SpringApplication.java:328)
                at org.springframework.boot.builder.SpringApplicationBuilder.run(SpringApplicationBuilder.java:154)
                at de.datev.ediio.eric.client.ClientApplication.main(ClientApplication.java:20)
                at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
                at java.base/java.lang.reflect.Method.invoke(Method.java:580)
                at org.springframework.boot.loader.launch.Launcher.launch(Launcher.java:106)
                at org.springframework.boot.loader.launch.Launcher.launch(Launcher.java:64)
                at org.springframework.boot.loader.launch.JarLauncher.main(JarLauncher.java:40)org.springframework.http.codec.multipart.PartGenerator$CreateFileState.body(PartGenerator.java:457)
                at org.springframework.http.codec.multipart.PartGenerator.hookOnNext(PartGenerator.java:126)
                at org.springframework.http.codec.multipart.PartGenerator.hookOnNext(PartGenerator.java:61)
                at reactor.core.publisher.BaseSubscriber.onNext(BaseSubscriber.java:163)
                at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:109)
                at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:679)
                at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:757)
                at reactor.core.publisher.FluxWindowPredicate$WindowFlux.request(FluxWindowPredicate.java:844)
                at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.request(FluxContextWrite.java:138)
                at reactor.core.publisher.BaseSubscriber.request(BaseSubscriber.java:217)
                at org.springframework.http.codec.multipart.PartGenerator.requestToken(PartGenerator.java:201)
                at org.springframework.http.codec.multipart.PartGenerator.lambda$createPart$1(PartGenerator.java:102)
                at reactor.core.publisher.MonoCreate$DefaultMonoSink.request(MonoCreate.java:285)
                at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.request(Operators.java:2325)
                at reactor.core.publisher.FluxConcatMapNoPrefetch$FluxConcatMapNoPrefetchSubscriber.request(FluxConcatMapNoPrefetch.java:338)
                at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:795)
                at reactor.core.publisher.FluxFlatMap$FlatMapMain.innerComplete(FluxFlatMap.java:899)
                at reactor.core.publisher.FluxFlatMap$FlatMapInner.onComplete(FluxFlatMap.java:1004)
                at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.onComplete(FluxDoFinally.java:128)
                at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onComplete(FluxMapFuseable.java:153)
                at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onComplete(MonoPeekTerminal.java:303)
                at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.complete(MonoIgnoreThen.java:298)
                at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onNext(MonoIgnoreThen.java:191)
                at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:240)
                at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onComplete(MonoIgnoreThen.java:207)
                at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onComplete(MonoPeekTerminal.java:303)
                at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onComplete(MonoSubscribeOn.java:163)
                at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:145)
                at org.springframework.http.codec.multipart.DefaultParts$FileContent.lambda$blockingOperation$0(DefaultParts.java:317)
                at reactor.core.publisher.MonoCreate.subscribe(MonoCreate.java:61)
                at reactor.core.publisher.Mono.subscribe(Mono.java:4569)
                at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:127)
                at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:90)
                at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
                at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
                at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
                at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
                at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
                at java.base/java.lang.Thread.run(Thread.java:1583)
        Suppressed: java.lang.Exception: #block terminated with an error
                at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:104)
                at reactor.core.publisher.Mono.block(Mono.java:1773)
                at de.datev.ediio.eric.client.service.EricSubmissionClientCli.callApi(EricSubmissionClientCli.java:81)
                at de.datev.ediio.eric.client.ClientApplication.lambda$run$0(ClientApplication.java:26)
                at org.springframework.boot.SpringApplication.lambda$callRunner$1(SpringApplication.java:792)
                at org.springframework.util.function.ThrowingConsumer$1.acceptWithException(ThrowingConsumer.java:82)
                at org.springframework.util.function.ThrowingConsumer.accept(ThrowingConsumer.java:60)
                at org.springframework.util.function.ThrowingConsumer$1.accept(ThrowingConsumer.java:86)
                at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:800)
                at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:791)
                at org.springframework.boot.SpringApplication.lambda$callRunners$0(SpringApplication.java:776)
                at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
                at java.base/java.util.stream.SortedOps$SizedRefSortingSink.end(SortedOps.java:357)
                at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:510)
                at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
                at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
                at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
                at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
                at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596)
                at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:776)
                at org.springframework.boot.SpringApplication.run(SpringApplication.java:328)
                at org.springframework.boot.builder.SpringApplicationBuilder.run(SpringApplicationBuilder.java:154)
                at de.datev.ediio.eric.client.ClientApplication.main(ClientApplication.java:20)
                at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
                at java.base/java.lang.reflect.Method.invoke(Method.java:580)
                at org.springframework.boot.loader.launch.Launcher.launch(Launcher.java:106)
                at org.springframework.boot.loader.launch.Launcher.launch(Launcher.java:64)
                at org.springframework.boot.loader.launch.JarLauncher.main(JarLauncher.java:40)

The error

• occurs non-deterministic - probably depending on network speed and how the data is read in ByteBuffers
• is not content related - originally we thought, it is a problem in the multipart content itself
  (different content-types and content-length, CR/LF handling, ...) but in the meanwhile we can exclude that
• doesn't seem to be concurrency related - we were able to produce it also in an extracted Spring Boot Main programm, where only one task (Thread) was using the multipart parsing
• occurs only when configurer.defaultCodecs().maxInMemorySize(VALUE) is set to a smaller value,
  than the Content-Length of parts in the multipart response
• the larger parts or all are using filename= in their Content-Disposition header

It could be that the error is related to
Multipart upload leak on client abort (ByteBuf.release() not called) #36262.

TMP directory content:

• When this type of multipart-parsing is used, there is always a sub-directory /tmp/spring-multipart-xxxxxx created - we assume per process (?)
• When the error occurs, an incomplete part /tmp/spring-multipart-xxxxxx/yyyyyy.multipart remains in the TMP directory - see below
• even, when the returned multipart is always the same, the file sizes at which the token error occurs are different

$ ls -l /tmp/spring-multipart*/*part
-rw------- 1 t08225a domänen-benutzer 21941 Apr 23 10:17 /tmp/spring-multipart-15193310154278380687/11903007326920653919.multipart
-rw------- 1 t08225a domänen-benutzer 21877 Apr 23 10:23 /tmp/spring-multipart-3633570938886180611/11750448232864871812.multipart
-rw------- 1 t08225a domänen-benutzer  8133 Apr 23 09:44 /tmp/spring-multipart-8339971065776059714/16435838971065595407.multipart

Environment:

• OS: Linux
• Java: OpenJDK 17 and OpenJDK 21

Reproducability:

• Really difficult - sometimes, we have to wait an hour to see the error!
• Providing a code example: possible, but we have to invest more time to remove the domain code

But, here are the main code parts:

public Mono<EricSubmissionResponse> handleSubmissionResponse(org.springframework.web.reactive.function.client.ClientResponse clientResponse) {

        final HttpStatusCode httpStatus = clientResponse.statusCode();
        // We need some variables independently of the individual parts
        final HttpHeaders headers = clientResponse.headers().asHttpHeaders();
        // Now let Spring do the multipart magic and extract the individual parts
        return clientResponse.bodyToFlux(Part.class)
            .flatMap(part -> savePartsToPayloadAccessWrapper(part))
            .collectMap(Map.Entry::getKey, Map.Entry::getValue)
            .map(payloadAccessWrappers -> new EricSubmissionResponse(httpStatus, headers, payloadAccessWrappers))
            .elapsed()
            .doOnSuccess(responseEntityTuple2 -> LOGGER.info("eric-submission CALL-SUCCESS after {} ms", responseEntityTuple2.getT1()))
            .map(Tuple2::getT2);
    }

    protected Mono<Map.Entry<String, PayloadAccessWrapper>> savePartsToPayloadAccessWrapper(Part part) {

        final String partName = part.name();
        final String logContext = logContextPrefix + partName;
        final String contentTypeString = part.headers().getContentType() != null ? part.headers().getContentType().toString() : "application/octet-stream";
        final long contentLength = part.headers().getContentLength();

        // FilePart already has content in a temp file on disk — transfer it directly without re-streaming the bytes through DataBufferUtils,
        // then delete the original FilePart temp file.
        if (part instanceof FilePart filePart) {
            LOGGER.info("Part \"{}\" with {} bytes and type \"{}\" is a FilePart — wrapping existing temp file directly", partName, contentLength, contentTypeString);
            return saveFilePartToPayloadAccessWrapper(partName, contentTypeString, filePart);
        } else {
            LOGGER.info("Part \"{}\" with {} bytes and type \"{}\" is held in memory", partName, contentLength, contentTypeString);
            return savePartToMemory(partName, contentTypeString, part.content());
        }
    }

[bclozel]

Thanks for raising this @giraone . Indeed, it's quite hard to reproduce consistently, but the stacktrace you have shared was really useful.

The main problem is that we are requesting additional tokens (PartGenerator.requestToken) while we are busy creating the temporary file (PartGenerator$CreateFileState.body). This shouldn't happen, as this state should only be about creating the file and then writing content to it. I think this particular problem only happens if the file creation takes enough time for a new token to be emitted.

I also found other lifecycle issues while reviewing this part of the codebase, including proper cleanup of resources in case of failures.