WebClient: Broken order of Netty writes

[danielkec]

Environment Details

• Helidon Version: 2.4.1-SNAPSHOT
• Helidon SE
• JDK version:
  openjdk version "11.0.10" 2021-01-19
  OpenJDK Runtime Environment GraalVM CE 21.0.0.2 (build 11.0.10+8-jvmci-21.0-b06)
  OpenJDK 64-Bit Server VM GraalVM CE 21.0.0.2 (build 11.0.10+8-jvmci-21.0-b06, mixed mode, sharing)
• OS:
  Distributor ID: Ubuntu
  Description: Ubuntu 21.10
  Release: 21.10
  Codename: impish

---

Problem Description

WebClient break Netty order of writes when consuming stream powered by multiple threads.

Steps to reproduce

        final ExecutorService exec = Executors.newSingleThreadExecutor();
        WebServer server = WebServer.builder()
                    .routing(Routing.builder()
                            .post((req, res) -> req.content()
                                    .onComplete(res::send)
                                    .onError(res::send)
                                    .ignoreElements())
                            .build())
                    .host("localhost")
                    .port(0)
                    .build()
                    .start()
                    .await(TIME_OUT);

            WebClient.builder()
                    .baseUri("http://localhost:" + server.port())
                    .connectTimeout(TIME_OUT.toMillis(), TimeUnit.MILLISECONDS)
                    .build()
                    .post()
                    .submit(Multi.just("1", "2", "3")
                            .map(String::valueOf)
                            .map(String::getBytes)
                            // Condition: initiate write from different thread than event loop
                            .observeOn(exec)
                            .map(n -> DataChunk.create(ByteBuffer.wrap(n)))
                            // Condition: mix-up upstream threads, flatMap has a prefetch with cache
                            // eg. some onNexts are going to be served by requesting thread
                            .flatMap(Single::just)
                    )
                    .await(TIME_OUT);

Ends with:

java.util.concurrent.CompletionException: java.lang.IllegalStateException: (client reqID: 1) Failure when sending a content!

	at io.helidon.common.reactive@2.4.1-SNAPSHOT/io.helidon.common.reactive.Awaitable.await(Awaitable.java:90)
	at io.helidon.webserver@2.4.1-SNAPSHOT/io.helidon.webserver.OrderOfWritesTest.threadMixUp2(OrderOfWritesTest.java:137)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:688)
	at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
...

Caused by: java.lang.IllegalStateException: (client reqID: 1) Failure when sending a content!
	at io.helidon.webclient.RequestContentSubscriber.lambda$completeOnFailureListener$8(RequestContentSubscriber.java:185)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:578)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:571)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:550)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:491)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:616)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:609)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:117)
	at io.netty.common@4.1.63.Final/io.netty.util.internal.PromiseNotificationUtil.tryFailure(PromiseNotificationUtil.java:64)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext.notifyOutboundHandlerException(AbstractChannelHandlerContext.java:814)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:719)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:709)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:792)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:702)
	at io.netty.handler@4.1.63.Final/io.netty.handler.timeout.IdleStateHandler.write(IdleStateHandler.java:302)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:717)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext.invokeWriteAndFlush(AbstractChannelHandlerContext.java:764)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext$WriteTask.run(AbstractChannelHandlerContext.java:1071)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
	at io.netty.transport@4.1.63.Final/io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
	at io.netty.common@4.1.63.Final/io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
	at io.netty.common@4.1.63.Final/io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: io.netty.handler.codec.EncoderException: java.lang.IllegalStateException: unexpected message type: DefaultHttpContent, state: 0
	at io.netty.codec@4.1.63.Final/io.netty.handler.codec.MessageToMessageEncoder.write(MessageToMessageEncoder.java:104)
	at io.netty.transport@4.1.63.Final/io.netty.channel.CombinedChannelDuplexHandler.write(CombinedChannelDuplexHandler.java:346)
	at io.netty.transport@4.1.63.Final/io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:717)
	... 13 more
Caused by: java.lang.IllegalStateException: unexpected message type: DefaultHttpContent, state: 0
	at io.netty.codec.http@4.1.63.Final/io.netty.handler.codec.http.HttpObjectEncoder.encode(HttpObjectEncoder.java:124)
	at io.netty.codec.http@4.1.63.Final/io.netty.handler.codec.http.HttpClientCodec$Encoder.encode(HttpClientCodec.java:189)
	at io.netty.codec@4.1.63.Final/io.netty.handler.codec.MessageToMessageEncoder.write(MessageToMessageEncoder.java:89)
	... 15 more