@Component
class MultimediaHandler {
    fun upload(request: ServerRequest): Mono<ServerResponse> {
        return request.body(BodyExtractors.toMultipartData()).flatMap{ parts ->
            val map = parts.toSingleValueMap()
            ServerResponse.ok().build()
I am getting a StackOverflowError:
java.lang.StackOverflowError: null
	at io.undertow.conduits.FixedLengthStreamSourceConduit.read(FixedLengthStreamSourceConduit.java:249) ~[undertow-core-1.4.22.Final.jar:1.4.22.Final]
	at org.xnio.conduits.ConduitStreamSourceChannel.read(ConduitStreamSourceChannel.java:127) ~[xnio-api-3.3.8.Final.jar:3.3.8.Final]
	at io.undertow.channels.DetachableStreamSourceChannel.read(DetachableStreamSourceChannel.java:209) ~[undertow-core-1.4.22.Final.jar:1.4.22.Final]
	at io.undertow.server.HttpServerExchange$ReadDispatchChannel.read(HttpServerExchange.java:2332) ~[undertow-core-1.4.22.Final.jar:1.4.22.Final]
	at org.springframework.http.server.reactive.UndertowServerHttpRequest$RequestBodyPublisher.read(UndertowServerHttpRequest.java:171) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
	at org.springframework.http.server.reactive.UndertowServerHttpRequest$RequestBodyPublisher.read(UndertowServerHttpRequest.java:127) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
	at org.springframework.http.server.reactive.AbstractListenerReadPublisher.readAndPublish(AbstractListenerReadPublisher.java:145) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
	at org.springframework.http.server.reactive.AbstractListenerReadPublisher.access$1000(AbstractListenerReadPublisher.java:47) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
	at org.springframework.http.server.reactive.AbstractListenerReadPublisher$State$4.onDataAvailable(AbstractListenerReadPublisher.java:317) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
	at org.springframework.http.server.reactive.AbstractListenerReadPublisher.onDataAvailable(AbstractListenerReadPublisher.java:85) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
	at org.springframework.http.server.reactive.UndertowServerHttpRequest$RequestBodyPublisher.checkOnDataAvailable(UndertowServerHttpRequest.java:155) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
	at org.springframework.http.server.reactive.AbstractListenerReadPublisher.changeToDemandState(AbstractListenerReadPublisher.java:177) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
	at org.springframework.http.server.reactive.AbstractListenerReadPublisher.access$900(AbstractListenerReadPublisher.java:47) ~[spring-web-5.0.3.RELEASE.jar:5.0.3.RELEASE]
Affects: 5.0.3
Attachments:
foo-small.txt (2.44 kB)
upload-error-sample.zip (182.67 kB)
upload-error-sample-wo-kafka.zip (181.18 kB)
Issue Links:
Spring WebFlux + Undertow + HTTP/2 is not working with a POST request [SPR-16632] #21173 Spring WebFlux + Undertow + HTTP/2 is not working with a POST request ("is duplicated by")
spring webflux ServerRequest.bodyToMono().block will freeze for HTTP post request whose header size + body size > 1024 [SPR-16579] #21121 spring webflux ServerRequest.bodyToMono().block will freeze for HTTP post request whose header size + body size > 1024
FilePart transferTo fails with java.nio.file.NoSuchFileException [SPR-16546] #21089 FilePart transferTo fails with java.nio.file.NoSuchFileException
StackOverFlowError and memory leaking when sending large files slowly with Webflux + Undertow [SPR-16702] #21243 StackOverFlowError and memory leaking when sending large files slowly with Webflux + Undertow
[docs] Restructure chapter Functional Endpoints chpater [SPR-16547] #21090 [docs] Restructure chapter Functional Endpoints chpater
Referenced from: commits f9df8c7
          
Tamas Eppel commented
Additionally it seems that the example: https://github.com/sdeleuze/webflux-multipart/ Does not compile anymore. Errors in: https://github.com/sdeleuze/webflux-multipart/blob/master/src/main/java/com/example/MultipartController.java
It would be good to have this documented as well. The official docs only mention the Controller based approach and not the Router-Handler approach: https://docs.spring.io/spring/docs/5.0.4.RELEASE/spring-framework-reference/web-reactive.html#webflux-multipart-forms
          
Rossen Stoyanchev commented
I've created a PR to update the sample.
In the process of upgrading the sample, I did find and address one other issue, see #21089. After that, with the latest (Boot 2.0 GA, Spring Framework 5.0.5 snapshot) I am unable to reproduce the Undertow issue. Note that in the sample I removed the explicit (outdated 1.0.2) version of nio-multipart-parser and relied to the Boot auto configured version (1.1.0).
Would you mind retrying with the updates to confirm?
          
Rossen Stoyanchev commented
Hm, it works fine for me:
2018-03-02 11:53:44.137  INFO 21105 --- [           main] com.example.WebfluxMultipartApplication  : Starting WebfluxMultipartApplication on rossen-X1 with PID 21105 (/home/rossen/dev/github/sdeleuze/webflux-multipart/target/classes started by rossen in /home/rossen/dev/github/sdeleuze/webflux-multipart)
2018-03-02 11:53:44.141 DEBUG 21105 --- [           main] com.example.WebfluxMultipartApplication  : Running with Spring Boot v2.0.0.RELEASE, Spring v5.0.5.BUILD-SNAPSHOT
2018-03-02 11:53:44.143  INFO 21105 --- [           main] com.example.WebfluxMultipartApplication  : No active profile set, falling back to default profiles: default
2018-03-02 11:53:44.283  INFO 21105 --- [           main] onfigReactiveWebServerApplicationContext : Refreshing org.springframework.boot.web.reactive.context.AnnotationConfigReactiveWebServerApplicationContext@4c1d9d4b: startup date [Fri Mar 02 11:53:44 EST 2018]; root of context hierarchy
2018-03-02 11:53:45.838 DEBUG 21105 --- [           main] s.w.r.r.m.a.RequestMappingHandlerMapping : Looking for request mappings in application context: org.springframework.boot.web.reactive.context.AnnotationConfigReactiveWebServerApplicationContext@4c1d9d4b: startup date [Fri Mar 02 11:53:44 EST 2018]; root of context hierarchy
2018-03-02 11:53:45.925 DEBUG 21105 --- [           main] o.s.w.r.f.s.s.RouterFunctionMapping      : Looking for router functions in application context: org.springframework.boot.web.reactive.context.AnnotationConfigReactiveWebServerApplicationContext@4c1d9d4b: startup date [Fri Mar 02 11:53:44 EST 2018]; root of context hierarchy
2018-03-02 11:53:46.029  INFO 21105 --- [           main] o.s.w.r.f.s.s.RouterFunctionMapping      : Mapped (POST && /upload) -> com.example.MultipartRoute$$Lambda$242/1624817884@464649c
/** -> class path resource [static/]
2018-03-02 11:53:46.052  INFO 21105 --- [           main] o.s.w.r.handler.SimpleUrlHandlerMapping  : Mapped URL path [/webjars/**] onto handler of type [class org.springframework.web.reactive.resource.ResourceWebHandler]
2018-03-02 11:53:46.053  INFO 21105 --- [           main] o.s.w.r.handler.SimpleUrlHandlerMapping  : Mapped URL path [/**] onto handler of type [class org.springframework.web.reactive.resource.ResourceWebHandler]
2018-03-02 11:53:46.172  INFO 21105 --- [           main] o.s.w.r.r.m.a.ControllerMethodResolver   : Looking for @ControllerAdvice: org.springframework.boot.web.reactive.context.AnnotationConfigReactiveWebServerApplicationContext@4c1d9d4b: startup date [Fri Mar 02 11:53:44 EST 2018]; root of context hierarchy
2018-03-02 11:53:46.549  INFO 21105 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Registering beans for JMX exposure on startup
2018-03-02 11:53:46.577  INFO 21105 --- [           main] org.xnio                                 : XNIO version 3.3.8.Final
2018-03-02 11:53:46.592  INFO 21105 --- [           main] org.xnio.nio                             : XNIO NIO Implementation Version 3.3.8.Final
2018-03-02 11:53:46.683  INFO 21105 --- [           main] o.s.b.w.e.u.UndertowServletWebServer     : Undertow started on port(s) 8080 (http)
2018-03-02 11:53:46.686  INFO 21105 --- [           main] com.example.WebfluxMultipartApplication  : Started WebfluxMultipartApplication in 3.544 seconds (JVM running for 4.211)
2018-03-02 11:53:59.000 DEBUG 21105 --- [   XNIO-1 I/O-1] o.s.web.reactive.DispatcherHandler       : Processing GET request for [http://localhost:8080/index.html]
2018-03-02 11:54:29.455 DEBUG 21105 --- [   XNIO-1 I/O-1] o.s.web.reactive.DispatcherHandler       : Processing POST request for [http://localhost:8080/upload]
2018-03-02 11:54:29.456 DEBUG 21105 --- [   XNIO-1 I/O-1] o.s.w.r.function.server.RouterFunctions  : Predicate "(POST && /upload)" matches against "POST /upload"
2018-03-02 11:54:29.464 DEBUG 21105 --- [   XNIO-1 I/O-1] o.s.w.c.reactive.DefaultCorsProcessor    : Skip CORS: request is from same origin
          
Tamas Eppel commented




    

I am sorry for the late reply, was on holiday.
I have prepared a sample project. (zip, and input files attached)
Steps to reproduce:
unpack the zip file
start kafka with: docker-compose up
compile the spring project: ./gradlew build
start the app: java -jar ./build/libs/upload-error-sample-0.0.1-SNAPSHOT.jar
upload the files: curl -v -F import=\@/tmp/dc/foo-small.txt  http://localhost:8090/api/imports # Note: change actual path
To switch to netty uncomment the netty line in build.gradle and comment out undertow.
          
Rossen Stoyanchev commented
I spend some time trying to get the docker-compose command to run but but failed (and I'm not familiar enough). Is it possible to provide something that doesn't require Docker and Kafka to demonstrate the issue? For example can you make it fail in the webflux-multipart sample?
Looking at your MultimediaHandler, this code looks problematic:
val filePart = parts.getFirst("import") as FilePart
filePart.transferTo(tmpFile.toFile())
        .then(Mono.just(Files.newBufferedReader(tmpFile).lines().use { it.forEach { importProcessor.importLine().send(GenericMessage<String>(it)) } }))
        .then(ServerResponse.ok().body(BodyInserters.fromObject(importId.toString())))
You want Files.newBufferedReader(tmpFile) to be deferred so it's executed after transferTo completes. Something like:
val filePart = parts.getFirst("import") as FilePart
filePart.transferTo(tmpFile.toFile())
        .then(Mono.defer(() -> {
            // read file .. 
            return ServerResponse.ok().body(BodyInserters.fromObject(importId.toString()));
Even then the file reading and calls to importLine are synchronously executed and potentially blocking but they shouldn't block the event loop thread? I'm not sure if the send is blocking or not given it's Kafka. I presume it might in which case you might have to use publishOn.
All of those things aren't the original issue, but again I need a sample I can execute to be more specific or a full stack trace at least.
          
Tamas Eppel commented
I have uploaded one without kafka/docker. Thanks for the suggestions.
My original intention was to send each line on the file as it arrives, but then I realized I would need to parse the characters for new line (because a line can be in more parts) so I just do with the tempfile approach.
          
Tamas Eppel commented
Thanks for the suggestion I have rewritten it like this:
package de.techmatrix.dc.matcher.handler
import de.techmatrix.dc.matcher.component.UploadService
import org.springframework.core.ResolvableType
import org.springframework.core.codec.StringDecoder
import org.springframework.core.io.buffer.DataBuffer
import org.springframework.stereotype.Component
import org.springframework.util.MimeTypeUtils
import org.springframework.web.reactive.function.BodyExtractors
import org.springframework.web.reactive.function.BodyInserters
import org.springframework.web.reactive.function.server.ServerRequest
import org.springframework.web.reactive.function.server.ServerResponse
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.toFlux
import reactor.core.scheduler.Schedulers
import java.util.*
fun dataBuffers(request: ServerRequest): Flux<DataBuffer> =
        request.body(BodyExtractors.toMultipartData())
                .toFlux()
                .publishOn(Schedulers.parallel())
                .flatMap { parts -> parts.flatMap { it.value }.toFlux() }.flatMap { it.content() }
@Component
class MultimediaHandler(val uploadService: UploadService) {
    fun upload(request: ServerRequest): Mono<ServerResponse> {
        val importId = UUID.randomUUID()
        return StringDecoder.allMimeTypes().decode(
                dataBuffers(request), ResolvableType.forClass(String::class.java), MimeTypeUtils.TEXT_PLAIN, emptyMap())
                .index()
                .publishOn(Schedulers.parallel())
                .doOnNext { uploadService.importLine(importId, it.t2, it.t1.toInt()) }
                .map { 1 }
                .reduce { a, b -> a + b }
                .doOnSuccess { uploadService.importFinished(importId, it) }
                .then(ServerResponse.ok().body(BodyInserters.fromObject(importId.toString())))
It would be quite good to describe this, I did not find anything in the WebFlux documentation, or by searching.
          
Rossen Stoyanchev commented
Okay, thanks for that. I've been able to confirm the recursion issue:
at org.springframework.http.server.reactive.UndertowServerHttpRequest$RequestBodyPublisher.checkOnDataAvailable(UndertowServerHttpRequest.java:156) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.AbstractListenerReadPublisher.changeToDemandState(AbstractListenerReadPublisher.java:177) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.AbstractListenerReadPublisher.access$900(AbstractListenerReadPublisher.java:47) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.AbstractListenerReadPublisher$State$4.onDataAvailable(AbstractListenerReadPublisher.java:319) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.AbstractListenerReadPublisher.onDataAvailable(AbstractListenerReadPublisher.java:85) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.UndertowServerHttpRequest$RequestBodyPublisher.checkOnDataAvailable(UndertowServerHttpRequest.java:156) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.AbstractListenerReadPublisher.changeToDemandState(AbstractListenerReadPublisher.java:177) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.AbstractListenerReadPublisher.access$900(AbstractListenerReadPublisher.java:47) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.AbstractListenerReadPublisher$State$4.onDataAvailable(AbstractListenerReadPublisher.java:319) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.AbstractListenerReadPublisher.onDataAvailable(AbstractListenerReadPublisher.java:85) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
at org.springframework.http.server.reactive.UndertowServerHttpRequest$RequestBodyPublisher.checkOnDataAvailable(UndertowServerHttpRequest.java:156) ~[spring-web-5.0.4.RELEASE.jar:5.0.4.RELEASE]
This is triggered by the Expect: 100-continue header that curl sends in this case. However it would happen any time there is back pressure on the input side (i.e. demand present but waiting for data). It works with the "Expect" header forced to be empty:
$ curl -v -F --header "Expect:" import=@/home/rossen/Downloads/foo-small.txt  http://localhost:8090/api/imports
I did not find anything in the WebFlux documentation, or by searching.
The codecs are documented here. We have planned #21081. Any clues what you searched on, or what you expected to find?
      spring webflux ServerRequest.bodyToMono().block will freeze for HTTP post request whose header size + body size > 1024 [SPR-16579]
      #21121
      StackOverFlowError and memory leaking when sending large files slowly with Webflux + Undertow [SPR-16702]
      #21243