http-netty: Fix flaky FullDuplexAndSequentialModeTest (#3070)

Motivation:

This test suite has been flaky for a while. As detailed in issue #1894,
this is because under some conditions the test running thread ends up
waiting on a latch that it is actually responsible to release.

Modifications:

Make sure the publishing of the InputStreamPublisher happens in on a
different thread so we don't end up in deadlock.

Fixes #1894

servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/FullDuplexAndSequentialModeTest.java

@@ -15,6 +15,7 @@
  */
 package io.servicetalk.http.netty;
 
+import io.servicetalk.http.api.HttpExecutionContext;
 import io.servicetalk.http.api.StreamingHttpConnection;
 import io.servicetalk.http.api.StreamingHttpResponse;
 import io.servicetalk.http.utils.EnforceSequentialModeRequesterFilter;
@@ -90,18 +91,23 @@ private static InputStream payload() {
         return new BufferedInputStream(new ByteArrayInputStream(array));
     }
 
-    private static Future<StreamingHttpResponse> stallingSendRequest(StreamingHttpConnection connection,
+    private Future<StreamingHttpResponse> stallingSendRequest(StreamingHttpConnection connection,
                                                                      CountDownLatch continueRequest,
                                                                      InputStream payload) {
-        return connection.request(connection.post(SVC_ECHO).payloadBody(fromInputStream(payload, CHUNK_SIZE)
+        final HttpExecutionContext ctx = connection.executionContext();
+        final String callerThreadName = Thread.currentThread().getName();
+        return connection.request(connection.post(SVC_ECHO).payloadBody(fromInputStream(payload, 1)
+                        // We use `publishOn` to make sure we only publish from offload threads. Otherwise, in rare
+                        // circumstances, we end up publishing from the test runner and end up deadlocked.
+                        .publishOn(ctx.executor(), () -> callerThreadName.equals(Thread.currentThread().getName()))
                 .map(chunk -> {
                     try {
                         continueRequest.await();    // wait until the InputStream is closed
                     } catch (InterruptedException ie) {
                         Thread.currentThread().interrupt();
                         throwException(ie);
                     }
-                    return connection.executionContext().bufferAllocator().wrap(chunk);
+                    return ctx.bufferAllocator().wrap(chunk);
                 }))).toFuture();
     }
 }