Use CompletableFuture<Void> in Exchange interfaces

CompletableFuture<Void> should be generally preferred. With a capturing
CompletableFuture<?> in the interface it is easy to mistakenly call
CompletableFuture#apply instead of CompletableFuture#handle and return
a CompletableFuture<CompletableFuture<?>> that will still match the
capture.

core/trino-main/src/main/java/io/trino/execution/buffer/SpoolingExchangeOutputBuffer.java

@@ -139,7 +139,7 @@ public void destroy(OutputBuffers.OutputBufferId bufferId)
     @Override
     public ListenableFuture<Void> isFull()
     {
-        return asVoid(toListenableFuture(exchangeSink.isBlocked()));
+        return toListenableFuture(exchangeSink.isBlocked());
     }
 
     @Override

core/trino-main/src/main/java/io/trino/operator/DeduplicatingDirectExchangeBuffer.java

@@ -807,9 +807,9 @@ public ListenableFuture<Void> isBlocked()
                 return nonCancellationPropagating(asVoid(exchangeSourceFuture));
             }
             if (exchangeSource != null) {
-                CompletableFuture<?> blocked = exchangeSource.isBlocked();
+                CompletableFuture<Void> blocked = exchangeSource.isBlocked();
                 if (!blocked.isDone()) {
-                    return nonCancellationPropagating(asVoid(toListenableFuture(blocked)));
+                    return nonCancellationPropagating(toListenableFuture(blocked));
                 }
             }
             return immediateVoidFuture();

core/trino-main/src/main/java/io/trino/operator/ExchangeOperator.java

@@ -47,7 +47,6 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
-import static io.airlift.concurrent.MoreFutures.asVoid;
 import static io.airlift.concurrent.MoreFutures.toListenableFuture;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
@@ -491,7 +490,7 @@ public boolean isFinished()
         @Override
         public ListenableFuture<Void> isBlocked()
         {
-            return asVoid(toListenableFuture(exchangeSource.isBlocked()));
+            return toListenableFuture(exchangeSource.isBlocked());
         }
 
         @Override

core/trino-main/src/main/java/io/trino/server/testing/exchange/LocalFileSystemExchange.java

@@ -232,7 +232,7 @@ public ExchangeSourceSplitter split(ExchangeSourceHandle handle, long targetSize
         return new ExchangeSourceSplitter()
         {
             @Override
-            public CompletableFuture<?> isBlocked()
+            public CompletableFuture<Void> isBlocked()
             {
                 return completedFuture(null);
             }

core/trino-main/src/main/java/io/trino/server/testing/exchange/LocalFileSystemExchangeSink.java

@@ -74,7 +74,7 @@ public LocalFileSystemExchangeSink(Path outputDirectory, int outputPartitionCoun
     }
 
     @Override
-    public CompletableFuture<?> isBlocked()
+    public CompletableFuture<Void> isBlocked()
     {
         return NOT_BLOCKED;
     }
@@ -111,7 +111,7 @@ public synchronized long getMemoryUsage()
     }
 
     @Override
-    public synchronized CompletableFuture<?> finish()
+    public synchronized CompletableFuture<Void> finish()
     {
         if (closed) {
             return completedFuture(null);
@@ -143,7 +143,7 @@ public synchronized CompletableFuture<?> finish()
     }
 
     @Override
-    public synchronized CompletableFuture<?> abort()
+    public synchronized CompletableFuture<Void> abort()
     {
         if (closed) {
             return completedFuture(null);

core/trino-main/src/main/java/io/trino/server/testing/exchange/LocalFileSystemExchangeSource.java

@@ -54,7 +54,7 @@ public LocalFileSystemExchangeSource(List<Path> files)
     }
 
     @Override
-    public CompletableFuture<?> isBlocked()
+    public CompletableFuture<Void> isBlocked()
     {
         return NOT_BLOCKED;
     }

core/trino-main/src/test/java/io/trino/execution/buffer/TestSpoolingExchangeOutputBuffer.java

@@ -54,7 +54,7 @@ public void testIsFull()
         assertEquals(outputBuffer.getState(), NO_MORE_BUFFERS);
         assertNotBlocked(outputBuffer.isFull());
 
-        CompletableFuture<?> blocked = new CompletableFuture<>();
+        CompletableFuture<Void> blocked = new CompletableFuture<>();
         exchangeSink.setBlocked(blocked);
 
         ListenableFuture<Void> full = outputBuffer.isFull();
@@ -68,7 +68,7 @@ public void testIsFull()
     public void testFinishSuccess()
     {
         TestingExchangeSink exchangeSink = new TestingExchangeSink();
-        CompletableFuture<?> finish = new CompletableFuture<>();
+        CompletableFuture<Void> finish = new CompletableFuture<>();
         exchangeSink.setFinish(finish);
 
         OutputBuffer outputBuffer = createSpoolingExchangeOutputBuffer(exchangeSink);
@@ -87,7 +87,7 @@ public void testFinishSuccess()
     public void testFinishFailure()
     {
         TestingExchangeSink exchangeSink = new TestingExchangeSink();
-        CompletableFuture<?> finish = new CompletableFuture<>();
+        CompletableFuture<Void> finish = new CompletableFuture<>();
         exchangeSink.setFinish(finish);
 
         OutputBuffer outputBuffer = createSpoolingExchangeOutputBuffer(exchangeSink);
@@ -108,7 +108,7 @@ public void testFinishFailure()
     public void testDestroyAfterFinishCompletion()
     {
         TestingExchangeSink exchangeSink = new TestingExchangeSink();
-        CompletableFuture<?> finish = new CompletableFuture<>();
+        CompletableFuture<Void> finish = new CompletableFuture<>();
         exchangeSink.setFinish(finish);
 
         OutputBuffer outputBuffer = createSpoolingExchangeOutputBuffer(exchangeSink);
@@ -130,7 +130,7 @@ public void testDestroyAfterFinishCompletion()
     public void testDestroyBeforeFinishCompletion()
     {
         TestingExchangeSink exchangeSink = new TestingExchangeSink();
-        CompletableFuture<?> finish = new CompletableFuture<>();
+        CompletableFuture<Void> finish = new CompletableFuture<>();
         exchangeSink.setFinish(finish);
 
         OutputBuffer outputBuffer = createSpoolingExchangeOutputBuffer(exchangeSink);
@@ -164,9 +164,9 @@ public void testAbortBeforeNoMorePages()
     public void testAbortBeforeFinishCompletion()
     {
         TestingExchangeSink exchangeSink = new TestingExchangeSink();
-        CompletableFuture<?> finish = new CompletableFuture<>();
+        CompletableFuture<Void> finish = new CompletableFuture<>();
         exchangeSink.setFinish(finish);
-        CompletableFuture<?> abort = new CompletableFuture<>();
+        CompletableFuture<Void> abort = new CompletableFuture<>();
         exchangeSink.setAbort(abort);
 
         OutputBuffer outputBuffer = createSpoolingExchangeOutputBuffer(exchangeSink);
@@ -190,9 +190,9 @@ public void testAbortBeforeFinishCompletion()
     public void testAbortAfterFinishCompletion()
     {
         TestingExchangeSink exchangeSink = new TestingExchangeSink();
-        CompletableFuture<?> finish = new CompletableFuture<>();
+        CompletableFuture<Void> finish = new CompletableFuture<>();
         exchangeSink.setFinish(finish);
-        CompletableFuture<?> abort = new CompletableFuture<>();
+        CompletableFuture<Void> abort = new CompletableFuture<>();
         exchangeSink.setAbort(abort);
 
         OutputBuffer outputBuffer = createSpoolingExchangeOutputBuffer(exchangeSink);
@@ -219,7 +219,7 @@ public void testAbortAfterFinishCompletion()
     public void testEnqueueAfterFinish()
     {
         TestingExchangeSink exchangeSink = new TestingExchangeSink();
-        CompletableFuture<?> finish = new CompletableFuture<>();
+        CompletableFuture<Void> finish = new CompletableFuture<>();
         exchangeSink.setFinish(finish);
 
         OutputBuffer outputBuffer = createSpoolingExchangeOutputBuffer(exchangeSink);
@@ -252,7 +252,7 @@ public void testEnqueueAfterFinish()
     public void testEnqueueAfterAbort()
     {
         TestingExchangeSink exchangeSink = new TestingExchangeSink();
-        CompletableFuture<?> abort = new CompletableFuture<>();
+        CompletableFuture<Void> abort = new CompletableFuture<>();
         exchangeSink.setAbort(abort);
 
         OutputBuffer outputBuffer = createSpoolingExchangeOutputBuffer(exchangeSink);
@@ -304,20 +304,20 @@ private static class TestingExchangeSink
             implements ExchangeSink
     {
         private final ListMultimap<Integer, Slice> dataBuffer = ArrayListMultimap.create();
-        private CompletableFuture<?> blocked = CompletableFuture.completedFuture(null);
-        private CompletableFuture<?> finish = CompletableFuture.completedFuture(null);
-        private CompletableFuture<?> abort = CompletableFuture.completedFuture(null);
+        private CompletableFuture<Void> blocked = CompletableFuture.completedFuture(null);
+        private CompletableFuture<Void> finish = CompletableFuture.completedFuture(null);
+        private CompletableFuture<Void> abort = CompletableFuture.completedFuture(null);
 
         private boolean finishCalled;
         private boolean abortCalled;
 
         @Override
-        public CompletableFuture<?> isBlocked()
+        public CompletableFuture<Void> isBlocked()
         {
             return blocked;
         }
 
-        public void setBlocked(CompletableFuture<?> blocked)
+        public void setBlocked(CompletableFuture<Void> blocked)
         {
             this.blocked = requireNonNull(blocked, "blocked is null");
         }
@@ -341,28 +341,28 @@ public long getMemoryUsage()
         }
 
         @Override
-        public CompletableFuture<?> finish()
+        public CompletableFuture<Void> finish()
         {
             assertFalse(abortCalled);
             assertFalse(finishCalled);
             finishCalled = true;
             return finish;
         }
 
-        public void setFinish(CompletableFuture<?> finish)
+        public void setFinish(CompletableFuture<Void> finish)
         {
             this.finish = requireNonNull(finish, "finish is null");
         }
 
         @Override
-        public CompletableFuture<?> abort()
+        public CompletableFuture<Void> abort()
         {
             assertFalse(abortCalled);
             abortCalled = true;
             return abort;
         }
 
-        public void setAbort(CompletableFuture<?> abort)
+        public void setAbort(CompletableFuture<Void> abort)
         {
             this.abort = requireNonNull(abort, "abort is null");
         }

core/trino-main/src/test/java/io/trino/execution/scheduler/TestingExchange.java

@@ -110,7 +110,7 @@ public ExchangeSourceSplitter split(ExchangeSourceHandle handle, long targetSize
         return new ExchangeSourceSplitter()
         {
             @Override
-            public CompletableFuture<?> isBlocked()
+            public CompletableFuture<Void> isBlocked()
             {
                 return completedFuture(null);
             }

core/trino-spi/src/main/java/io/trino/spi/exchange/ExchangeSink.java

@@ -22,14 +22,14 @@
 @ThreadSafe
 public interface ExchangeSink
 {
-    CompletableFuture<?> NOT_BLOCKED = CompletableFuture.completedFuture(null);
+    CompletableFuture<Void> NOT_BLOCKED = CompletableFuture.completedFuture(null);
 
     /**
      * Returns a future that will be completed when the exchange sink becomes
      * unblocked.  If the exchange sink is not blocked, this method should return
      * {@code NOT_BLOCKED}
      */
-    CompletableFuture<?> isBlocked();
+    CompletableFuture<Void> isBlocked();
 
     /**
      * Appends arbitrary {@code data} to a partition specified by {@code partitionId}.
@@ -53,7 +53,7 @@ public interface ExchangeSink
      *
      * @return future that will be resolved when the finish operation either succeeds or fails
      */
-    CompletableFuture<?> finish();
+    CompletableFuture<Void> finish();
 
     /**
      * Notifies the exchange that the write operation has been aborted.
@@ -63,5 +63,5 @@ public interface ExchangeSink
      *
      * @return future that will be resolved when the abort operation either succeeds or fails
      */
-    CompletableFuture<?> abort();
+    CompletableFuture<Void> abort();
 }

core/trino-spi/src/main/java/io/trino/spi/exchange/ExchangeSource.java

@@ -25,14 +25,14 @@
 public interface ExchangeSource
         extends Closeable
 {
-    CompletableFuture<?> NOT_BLOCKED = CompletableFuture.completedFuture(null);
+    CompletableFuture<Void> NOT_BLOCKED = CompletableFuture.completedFuture(null);
 
     /**
      * Returns a future that will be completed when the exchange source becomes
      * unblocked.  If the exchange source is not blocked, this method should return
      * {@code NOT_BLOCKED}
      */
-    CompletableFuture<?> isBlocked();
+    CompletableFuture<Void> isBlocked();
 
     /**
      * Once isFinished returns true, {@link #read()} will never return a non-null result

core/trino-spi/src/main/java/io/trino/spi/exchange/ExchangeSourceSplitter.java

@@ -23,7 +23,7 @@ public interface ExchangeSourceSplitter
     /**
      * Returns a future that will be completed when the splitter becomes unblocked.
      */
-    CompletableFuture<?> isBlocked();
+    CompletableFuture<Void> isBlocked();
 
     /**
      * Returns next sub partition, or {@link Optional#empty()} if the splitting process is finished.