Stream resetting with web client sendMultipartForm after update to 4.3.4 release

### Questions

Observing a stream reset using the WebClient using the sendMultiPartForm and also queried in the forum https://groups.google.com/g/vertx/c/94Ag-UhJuvI

This appears was working fine in 4.3.3 and then with the 4.3.4 release (vertx-core dependency), we are seeing the stream reset itself after sending the first few chunks of DATA (8k bytes each). Total buffer sending size is about 86k bytes.

I was reviewing the changes in that release and the only commit that seemed relevant was this one (https://github.com/eclipse-vertx/vert.x/issues/4482).

The production code and unit test uses the larger file mentioned above along with openapi yaml and router builder, but thinking with a simple reproducer like below we can see the same or very similar issue.

### Version

Starting from 4.3.4 and later releases

### Context

I encountered an exception which looks suspicious while ...

### Do you have a reproducer?

import io.vertx.core.Future;
import io.vertx.core.MultiMap;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpClient;
import io.vertx.core.http.HttpClientRequest;
import io.vertx.core.http.HttpHeaders;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServer;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.HttpServerResponse;
import io.vertx.core.http.HttpVersion;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
import io.vertx.ext.web.client.HttpRequest;
import io.vertx.ext.web.client.HttpResponse;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.client.WebClientOptions;
import io.vertx.ext.web.multipart.MultipartForm;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import static org.testng.Assert.assertEquals;
import org.testng.SkipException;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

public class StreamTest {

public static final String DEFAULT_HTTP_HOST = "localhost";
public static final int DEFAULT_HTTP_PORT = 50061;
public static final String DEFAULT_TEST_URI = "/some-uri";

private HttpServer server;
private WebClient client;
private HttpClient hclient;
private Vertx vertx = Vertx.vertx();

@BeforeClass
public void setUp() throws Exception {
HttpServerOptions baseServerOptions = new HttpServerOptions().setPort(DEFAULT_HTTP_PORT).setHost(DEFAULT_HTTP_HOST);

Router router = Router.router(vertx);
router.route(DEFAULT_TEST_URI)
.handler(ctx -> {
ctx.request().setExpectMultipart(true);
ctx.request().endHandler(v -> {
sendResponse(ctx, 200, 200);
});
});

server = vertx.createHttpServer(baseServerOptions).requestHandler(router);

WebClientOptions clientOptions = new WebClientOptions();
clientOptions.setProtocolVersion(HttpVersion.HTTP_2);
clientOptions.setHttp2MaxPoolSize(8);
clientOptions.setHttp2MultiplexingLimit(500);
clientOptions.setHttp2ClearTextUpgrade(false);

client = WebClient.create(vertx, clientOptions);

hclient = vertx.createHttpClient(clientOptions);

server.requestHandler(router).listen(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST);
}

@Test
public void testHttpClientMultipartForm() throws Exception {

CompletableFuture> result = new CompletableFuture<>();

String json = content();

String contentType = "multipart/form-data; boundary=a4e41223-a527-49b6-ac1c-315d76be757e";
String body = "--a4e41223-a527-49b6-ac1c-315d76be757e\r\n"
+ "Content-Disposition: form-data; name=\"file\"; filename=\"tmp-0.txt\"\r\n"
+ "Content-Type: application/json; charset=UTF-8\r\n"
+ "Content-Length: " + json.length() + "\r\n"
+ "\r\n"
+ json + "\r\n"
+ "--a4e41223-a527-49b6-ac1c-315d76be757e--\r\n";

Future fut = hclient.request(HttpMethod.PUT, DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI);
fut.onSuccess(req -> {
req.putHeader(HttpHeaders.CONTENT_TYPE, contentType);
req.send(body).onSuccess(h -> {
result.complete(null);
}).onFailure(t -> {
result.completeExceptionally(t);
});
});

result.get(5, TimeUnit.SECONDS);
}

@Test
public void testWebClientMultiMap() throws Exception {

CompletableFuture> result = new CompletableFuture<>();

MultiMap form = MultiMap.caseInsensitiveMultiMap();
form.add("param1", "param1_value");

HttpRequest request = client.post(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI);

//ADDING this hdr results in stream reset - probably needs to have the boundary part - omit this and it works fine
//request.putHeader("content-type", "multipart/form-data");
//
request.sendForm(form, ar -> {
if (ar.succeeded()) {
result.complete(ar.result());
} else {
result.completeExceptionally(ar.cause());
}
}
);

HttpResponse response = result.get(5, TimeUnit.SECONDS);
assertEquals(response.statusCode(), 200);
}

@Test
public void testWebClientMultipartForm() throws Exception {

if (true) {
//remove this to test
throw new SkipException("Vertx Bug or protocol error ?? - stream is resetting");
}

CompletableFuture> result = new CompletableFuture<>();

HttpRequest request = client.put(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST, DEFAULT_TEST_URI);

Buffer buffer = Buffer.buffer();
buffer.appendString(content());

MultipartForm form = MultipartForm.create().textFileUpload("upload", "import.txt", buffer, "application/json");
// When the stream len is unknown sendStream sends the file to the
// server using chunked transfer encoding
request.sendMultipartForm(form, ar -> {
if (ar.succeeded()) {
result.complete(ar.result());
} else {
result.completeExceptionally(ar.cause());
}
});

HttpResponse response = result.get(5, TimeUnit.SECONDS);
assertEquals(response.statusCode(), 200);

}

private String content() {
JsonArray entriesArray = new JsonArray();
for (int i = 0; i < 550; i++) {
JsonObject entry = new JsonObject();

JsonArray keys = new JsonArray();

JsonObject k1 = new JsonObject();
k1.put("key-value", "some-domain-" + i);
keys.add(k1);

JsonObject k2 = new JsonObject();
k2.put("key-value", "" + i);
keys.add(k2);

JsonObject k3 = new JsonObject();
k3.put("key-value", "some-name-" + i);
keys.add(k3);

entry.put("keys", keys);

JsonObject value = new JsonObject();
value.put("data-model-instance", "test");
value.put("data-store-type", "test");

entry.put("value", value);

entriesArray.add(entry);
}

String body = entriesArray.encode();
return body;
}

private static void sendResponse(RoutingContext ctx, int statusCode, int respCode) {

JsonObject json = new JsonObject();

if (statusCode == 200) {
json.put("status", "success");
} else {
json.put("status", "error");
}

String body = json.encode();

HttpServerResponse response = ctx.response();
response.setStatusCode(respCode);
response.putHeader(HttpHeaders.CONTENT_TYPE, "application/json");
response.putHeader(HttpHeaders.CONTENT_LENGTH, "" + body.length());
response.write(body);
response.end();

}

}

### Steps to reproduce

### Extra