Update azure-core-http-vertx to work with Vert.x 5.0 (#45567)

Update azure-core-http-vertx to work with Vert.x 5.0

Author: alzimmermsft

common/perf-test-core/pom.xml

@@ -57,7 +57,7 @@
                 <include>com.beust:jcommander:[1.82]</include> <!-- {x-include-update;com.beust:jcommander;external_dependency} -->
 
                 <include>io.projectreactor:reactor-core:[3.4.41]</include> <!-- {x-include-update;io.projectreactor:reactor-core;external_dependency} -->
-                <include>io.vertx:vertx-codegen:[4.5.13]</include> <!-- {x-include-update;io.vertx:vertx-codegen;external_dependency} -->
+                <include>io.vertx:vertx-codegen:[4.5.15]</include> <!-- {x-include-update;io.vertx:vertx-codegen;external_dependency} -->
               </includes>
             </bannedDependencies>
           </rules>
@@ -95,7 +95,7 @@
     <dependency>
       <groupId>io.vertx</groupId>
       <artifactId>vertx-codegen</artifactId>
-      <version>4.5.13</version> <!-- {x-version-update;io.vertx:vertx-codegen;external_dependency} -->
+      <version>4.5.15</version> <!-- {x-version-update;io.vertx:vertx-codegen;external_dependency} -->
       <scope>provided</scope>
     </dependency>
     <dependency>

eng/versioning/external_dependencies.txt

@@ -63,8 +63,8 @@ io.netty:netty-transport-native-unix-common;4.1.118.Final
 io.netty:netty-transport-native-kqueue;4.1.118.Final
 io.projectreactor.netty:reactor-netty-http;1.0.48
 io.projectreactor:reactor-core;3.4.41
-io.vertx:vertx-codegen;4.5.13
-io.vertx:vertx-core;4.5.13
+io.vertx:vertx-codegen;4.5.15
+io.vertx:vertx-core;4.5.15
 javax.websocket:javax.websocket-api;1.1
 org.apache.commons:commons-compress;1.26.0
 org.apache.ant:ant;1.10.14

sdk/core/azure-core-http-vertx/CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Other Changes
 
+- Few small changes to implementation to improve support for Vert.x 5.x.
+
 ## 1.0.3 (2025-03-10)
 
 ### Other Changes

sdk/core/azure-core-http-vertx/pom.xml

@@ -85,14 +85,14 @@
     <dependency>
       <groupId>io.vertx</groupId>
       <artifactId>vertx-codegen</artifactId>
-      <version>4.5.13</version> <!-- {x-version-update;io.vertx:vertx-codegen;external_dependency} -->
+      <version>4.5.15</version> <!-- {x-version-update;io.vertx:vertx-codegen;external_dependency} -->
       <scope>provided</scope>
     </dependency>
 
     <dependency>
       <groupId>io.vertx</groupId>
       <artifactId>vertx-core</artifactId>
-      <version>4.5.13</version> <!-- {x-version-update;io.vertx:vertx-core;external_dependency} -->
+      <version>4.5.15</version> <!-- {x-version-update;io.vertx:vertx-core;external_dependency} -->
     </dependency>
 
     <!-- test dependencies on azure-core, because we want to run tests inherited from this module using Vert.x Web Client -->
@@ -175,8 +175,8 @@
           <rules>
             <bannedDependencies>
               <includes>
-                <include>io.vertx:vertx-codegen:[4.5.13]</include> <!-- {x-include-update;io.vertx:vertx-codegen;external_dependency} -->
-                <include>io.vertx:vertx-core:[4.5.13]</include> <!-- {x-include-update;io.vertx:vertx-core;external_dependency} -->
+                <include>io.vertx:vertx-codegen:[4.5.15]</include> <!-- {x-include-update;io.vertx:vertx-codegen;external_dependency} -->
+                <include>io.vertx:vertx-core:[4.5.15]</include> <!-- {x-include-update;io.vertx:vertx-core;external_dependency} -->
               </includes>
             </bannedDependencies>
           </rules>

sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxHttpClient.java

@@ -24,10 +24,10 @@
 import com.azure.core.util.Contexts;
 import com.azure.core.util.ProgressReporter;
 import com.azure.core.util.logging.ClientLogger;
-import io.netty.buffer.Unpooled;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClientOptions;
 import io.vertx.core.http.HttpClientRequest;
 import io.vertx.core.http.HttpClientResponse;
 import io.vertx.core.http.HttpMethod;
@@ -50,15 +50,17 @@ class VertxHttpClient implements HttpClient {
     private static final ClientLogger LOGGER = new ClientLogger(VertxHttpClient.class);
 
     final io.vertx.core.http.HttpClient client;
+    final HttpClientOptions buildOptions;
     private final Duration responseTimeout;
 
     /**
      * Constructs a {@link VertxHttpClient}.
      *
      * @param client The Vert.x {@link io.vertx.core.http.HttpClient}
      */
-    VertxHttpClient(io.vertx.core.http.HttpClient client, Duration responseTimeout) {
+    VertxHttpClient(io.vertx.core.http.HttpClient client, HttpClientOptions buildOptions, Duration responseTimeout) {
         this.client = Objects.requireNonNull(client, "client cannot be null");
+        this.buildOptions = buildOptions;
         this.responseTimeout = responseTimeout;
     }
 
@@ -117,7 +119,7 @@ private CompletableFuture<HttpResponse> sendInternal(HttpRequest request, Contex
 
         // Create the Promise that will be returned. Promise.promise() is an uncompleted Promise.
         Promise<HttpResponse> promise = Promise.promise();
-        client.request(options, requestResult -> {
+        client.request(options).andThen(requestResult -> {
             if (requestResult.failed()) {
                 promise.fail(requestResult.cause());
                 return;
@@ -175,7 +177,6 @@ private CompletableFuture<HttpResponse> sendInternal(HttpRequest request, Contex
         return promise.future().toCompletionStage().toCompletableFuture();
     }
 
-    @SuppressWarnings("deprecation")
     private void sendBody(ContextView contextView, HttpRequest azureRequest, ProgressReporter progressReporter,
         HttpClientRequest vertxRequest, Promise<HttpResponse> promise) {
         BinaryData body = azureRequest.getBodyAsBinaryData();
@@ -188,14 +189,18 @@ private void sendBody(ContextView contextView, HttpRequest azureRequest, Progres
                 || bodyContent instanceof ByteBufferContent
                 || bodyContent instanceof StringContent
                 || bodyContent instanceof SerializableContent) {
+                // This cannot produce a NullPointerException for the BinaryDataContent types that trigger this as they
+                // are in-memory and have a length.
                 long contentLength = bodyContent.getLength();
-                vertxRequest.send(Buffer.buffer(Unpooled.wrappedBuffer(bodyContent.toByteBuffer())))
+                vertxRequest.send(Buffer.buffer(bodyContent.toBytes()))
                     .onSuccess(ignored -> reportProgress(contentLength, progressReporter))
                     .onFailure(promise::fail);
             } else {
                 // Right now both Flux<ByteBuffer> and InputStream bodies are being handled reactively.
                 azureRequest.getBody()
-                    .subscribe(new VertxRequestWriteSubscriber(vertxRequest, promise, progressReporter, contextView));
+                    .subscribe(new VertxRequestWriteSubscriber(vertxRequest::exceptionHandler,
+                        vertxRequest::drainHandler, vertxRequest::write, vertxRequest::writeQueueFull,
+                        vertxRequest::reset, vertxRequest::end, promise, progressReporter, contextView));
             }
         }
     }

sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxHttpClientBuilder.java

@@ -254,7 +254,7 @@ public HttpClient build() {
         }
 
         io.vertx.core.http.HttpClient client = configuredVertx.createHttpClient(buildOptions);
-        return new VertxHttpClient(client, getTimeout(this.responseTimeout, getDefaultResponseTimeout()));
+        return new VertxHttpClient(client, buildOptions, getTimeout(this.responseTimeout, getDefaultResponseTimeout()));
     }
 
     static Vertx getVertx(Iterator<VertxProvider> iterator) {
@@ -320,7 +320,7 @@ private static Runnable getVertxCloseRunnable(Vertx vertxToClose) {
         return () -> {
             CountDownLatch latch = new CountDownLatch(1);
             if (vertxToClose != null) {
-                vertxToClose.close(event -> {
+                vertxToClose.close().andThen(event -> {
                     if (event.failed() && event.cause() != null) {
                         LOGGER.logThrowableAsError(event.cause());
                     }

sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/implementation/VertxHttpResponse.java

@@ -53,7 +53,7 @@ private Flux<ByteBuffer> streamResponseBody() {
     public void close() {
         HttpClientResponse vertxHttpResponse = getVertxHttpResponse();
         if (vertxHttpResponse != null && !closed) {
-            vertxHttpResponse.netSocket().close(ignored -> closed = true);
+            vertxHttpResponse.netSocket().close().andThen(ignored -> closed = true);
         }
     }
 }

sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/implementation/VertxRequestWriteSubscriber.java

@@ -4,9 +4,11 @@
 package com.azure.core.http.vertx.implementation;
 
 import com.azure.core.http.HttpResponse;
+import com.azure.core.util.FluxUtil;
 import com.azure.core.util.ProgressReporter;
 import com.azure.core.util.logging.ClientLogger;
-import io.netty.buffer.Unpooled;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpClientRequest;
@@ -18,6 +20,10 @@
 import reactor.util.context.ContextView;
 
 import java.nio.ByteBuffer;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
 
 /**
  * Subscriber that writes a stream of {@link ByteBuffer ByteBuffers} to a {@link HttpClientRequest Vert.x request}.
@@ -26,7 +32,10 @@
 public final class VertxRequestWriteSubscriber implements Subscriber<ByteBuffer> {
     private static final ClientLogger LOGGER = new ClientLogger(VertxRequestWriteSubscriber.class);
 
-    private final HttpClientRequest request;
+    private final Function<Buffer, Future<Void>> writeHandler;
+    private final Supplier<Boolean> isWriteQueueFull;
+    private final BiConsumer<Long, Throwable> reset;
+    private final Supplier<Future<Void>> end;
     private final Promise<HttpResponse> promise;
     private final ProgressReporter progressReporter;
     private final ContextView contextView;
@@ -41,14 +50,28 @@ public final class VertxRequestWriteSubscriber implements Subscriber<ByteBuffer>
      * Creates a new {@link VertxRequestWriteSubscriber} that writes a stream of {@link ByteBuffer ByteBuffers} to a
      * {@link HttpClientRequest Vert.x request}.
      *
-     * @param request The {@link HttpClientRequest Vert.x request} to write to.
+     * @param exceptionHandlerUpdater A {@link Handler} that updates the
+     * {@link HttpClientRequest#exceptionHandler(Handler)} to be {@link #onError(Throwable)}.
+     * @param drainHandlerUpdater A {@link Handler} that updates the {@link HttpClientRequest#drainHandler(Handler)}
+     * to be {@link #requestNext()}.
+     * @param writeHandler A {@link Function} that will call {@link HttpClientRequest}.
+     * @param isWriteQueueFull A {@link Supplier} that will call {@link HttpClientRequest#writeQueueFull()}.
+     * @param reset A {@link BiConsumer} that will call {@link HttpClientRequest#reset(long, Throwable)}.
+     * @param end A {@link Supplier} that will call {@link HttpClientRequest#end()}.
      * @param promise The {@link MonoSink} to emit the {@link HttpResponse response} to.
      * @param progressReporter The {@link ProgressReporter} to report progress to.
      * @param contextView The {@link ContextView} to use when dropping errors.
      */
-    public VertxRequestWriteSubscriber(HttpClientRequest request, Promise<HttpResponse> promise,
-        ProgressReporter progressReporter, ContextView contextView) {
-        this.request = request.exceptionHandler(this::onError).drainHandler(ignored -> requestNext());
+    public VertxRequestWriteSubscriber(Consumer<Handler<Throwable>> exceptionHandlerUpdater,
+        Consumer<Handler<Void>> drainHandlerUpdater, Function<Buffer, Future<Void>> writeHandler,
+        Supplier<Boolean> isWriteQueueFull, BiConsumer<Long, Throwable> reset, Supplier<Future<Void>> end,
+        Promise<HttpResponse> promise, ProgressReporter progressReporter, ContextView contextView) {
+        exceptionHandlerUpdater.accept(this::onError);
+        drainHandlerUpdater.accept(ignored -> requestNext());
+        this.writeHandler = writeHandler;
+        this.isWriteQueueFull = isWriteQueueFull;
+        this.reset = reset;
+        this.end = end;
         this.promise = promise;
         this.progressReporter = progressReporter;
         this.contextView = contextView;
@@ -81,25 +104,24 @@ public void onNext(ByteBuffer bytes) {
         }
     }
 
-    @SuppressWarnings("deprecation")
     private void write(ByteBuffer bytes) {
-        int remaining = bytes.remaining();
-        request.write(Buffer.buffer(Unpooled.wrappedBuffer(bytes)), result -> {
+        final int remaining = bytes.remaining();
+        writeHandler.apply(Buffer.buffer(FluxUtil.byteBufferToArray(bytes))).onComplete(result -> {
             State state = this.state;
             if (state == State.WRITING) {
                 this.state = State.UNINITIALIZED;
             }
 
             if (result.succeeded()) {
-                if (progressReporter != null) {
-                    progressReporter.reportProgress(remaining);
-                }
                 if (state == State.WRITING) {
-                    if (!request.writeQueueFull()) {
+                    if (remaining > 0 && progressReporter != null) {
+                        progressReporter.reportProgress(remaining);
+                    }
+                    if (!isWriteQueueFull.get()) {
                         requestNext();
                     }
                 } else if (state == State.COMPLETE) {
-                    endRequest();
+                    endRequest(remaining);
                 } else if (state == State.ERROR) {
                     resetRequest(error);
                 }
@@ -177,7 +199,7 @@ private void resetRequest(Throwable throwable) {
                         + "completed successfully.", throwable);
             }
         }
-        request.reset(0, throwable);
+        reset.accept(0L, throwable);
     }
 
     @Override
@@ -192,12 +214,15 @@ public void onComplete() {
 
         this.state = State.COMPLETE;
         if (state != State.WRITING) {
-            endRequest();
+            endRequest(0);
         }
     }
 
-    private void endRequest() {
-        request.end().onFailure(promise::fail);
+    private void endRequest(int finishingWriteSize) {
+        if (finishingWriteSize > 0 && progressReporter != null) {
+            progressReporter.reportProgress(finishingWriteSize);
+        }
+        end.get().onFailure(promise::fail);
     }
 
     private enum State {