The Future transformation operation keep a reference on the future to…

… transform as an argument of the transformation function that only needs an async result by its design. This can create a leak when a chain of transformation for serialization purpose is used, like in the SSLHelper update operation.

Instead the transformation operation should recreate the appropriate async result from the value/failure it obtains to avoid this leak.

Change the transformation operation to remove the ref on the future and instead create a succeeded/failed future to pass to the transformation function.

src/main/java/io/vertx/core/impl/future/FutureBase.java

@@ -94,7 +94,7 @@ public <U> Future<U> compose(Function<T, Future<U>> successMapper, Function<Thro
   @Override
   public <U> Future<U> transform(Function<AsyncResult<T>, Future<U>> mapper) {
     Objects.requireNonNull(mapper, "No null mapper accepted");
-    Transformation<T, U> operation = new Transformation<>(context, this, mapper);
+    Transformation<T, U> operation = new Transformation<>(context, mapper);
     addListener(operation);
     return operation;
   }

src/main/java/io/vertx/core/impl/future/Transformation.java

@@ -23,20 +23,18 @@
  */
 class Transformation<T, U> extends Operation<U> implements Listener<T> {
 
-  private final Future<T> future;
   private final Function<AsyncResult<T>, Future<U>> mapper;
 
-  Transformation(ContextInternal context, Future<T> future, Function<AsyncResult<T>, Future<U>> mapper) {
+  Transformation(ContextInternal context, Function<AsyncResult<T>, Future<U>> mapper) {
     super(context);
-    this.future = future;
     this.mapper = mapper;
   }
 
   @Override
   public void onSuccess(T value) {
     FutureInternal<U> future;
     try {
-      future = (FutureInternal<U>) mapper.apply(this.future);
+      future = (FutureInternal<U>) mapper.apply(Future.succeededFuture(value));
     } catch (Throwable e) {
       tryFail(e);
       return;
@@ -48,7 +46,7 @@ public void onSuccess(T value) {
   public void onFailure(Throwable failure) {
     FutureInternal<U> future;
     try {
-      future = (FutureInternal<U>) mapper.apply(this.future);
+      future = (FutureInternal<U>) mapper.apply(Future.failedFuture(failure));
     } catch (Throwable e) {
       tryFail(e);
       return;

src/test/java/io/vertx/core/FutureTest.java

@@ -311,7 +311,10 @@ private void testTransformToSuccess(Consumer<Promise<String>> consumer) {
     Promise<String> p3 = Promise.promise();
     Future<String> f3 = p3.future();
     Future<Integer> f4 = f3.transform(ar -> {
-      assertSame(f3, ar);
+      assertSame(f3.succeeded(), ar.succeeded());
+      assertSame(f3.failed(), ar.failed());
+      assertSame(f3.result(), ar.result());
+      assertSame(f3.cause(), ar.cause());
       cnt.incrementAndGet();
       return c;
     });
@@ -341,7 +344,10 @@ private void testTransformToFailure(Consumer<Promise<String>> consumer) {
     Promise<String> p3 = Promise.promise();
     Future<String> f3 = p3.future();
     Future<Integer> f4 = f3.transform(ar -> {
-      assertSame(f3, ar);
+      assertSame(f3.succeeded(), ar.succeeded());
+      assertSame(f3.failed(), ar.failed());
+      assertSame(f3.result(), ar.result());
+      assertSame(f3.cause(), ar.cause());
       cnt.incrementAndGet();
       return c;
     });