all: prepare for ErrorProne's FutureReturnValueIgnored

Futures almost universally should be handled in some way when being
returned, either to receive the value or to cancel scheduled tasks to
prevent leaks.

Netty is a bit of a special case though, since it constantly returns
futures that you ignore (even adding a listener returns the "this"
future). So we want to suppress the warning for code using Netty instead
of trying to fix it. When we enable ErrorProne in the build, we should
start passing -Xep:FutureReturnValueIgnored:OFF in the compilerArgs.

benchmarks/src/jmh/java/io/grpc/benchmarks/TransportBenchmark.java

@@ -56,6 +56,7 @@ import io.netty.channel.local.LocalAddress;
 import io.netty.channel.local.LocalChannel;
 import io.netty.channel.local.LocalServerChannel;
 import java.net.InetSocketAddress;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -182,7 +183,7 @@ public class TransportBenchmark {
       throw new Exception("failed to shut down server");
     }
     if (groupToShutdown != null) {
-      groupToShutdown.shutdownGracefully(0, 1, TimeUnit.SECONDS);
+      Future<?> unused = groupToShutdown.shutdownGracefully(0, 1, TimeUnit.SECONDS);
       groupToShutdown.awaitTermination(1, TimeUnit.SECONDS);
       if (!groupToShutdown.isTerminated()) {
         throw new Exception("failed to shut down event loop group.");

context/src/test/java/io/grpc/DeadlineTest.java

@@ -43,6 +43,7 @@ import static org.mockito.Mockito.verify;
 
 import com.google.common.truth.Truth;
 import java.util.Arrays;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -189,7 +190,7 @@ public class DeadlineTest {
     Deadline base = Deadline.after(50, TimeUnit.MICROSECONDS, ticker);
     ScheduledExecutorService mockScheduler = mock(ScheduledExecutorService.class);
     final AtomicBoolean executed = new AtomicBoolean();
-    base.runOnExpiration(
+    Future<?> unused = base.runOnExpiration(
         new Runnable() {
           @Override
           public void run() {
@@ -208,7 +209,7 @@ public class DeadlineTest {
     Deadline base = Deadline.after(0, TimeUnit.MICROSECONDS, ticker);
     ScheduledExecutorService mockScheduler = mock(ScheduledExecutorService.class);
     final AtomicBoolean executed = new AtomicBoolean();
-    base.runOnExpiration(
+    Future<?> unused = base.runOnExpiration(
         new Runnable() {
           @Override
           public void run() {

core/src/test/java/io/grpc/internal/FakeClock.java

@@ -189,7 +189,8 @@ public final class FakeClock {
     }
 
     @Override public void execute(Runnable command) {
-      schedule(command, 0, TimeUnit.NANOSECONDS);
+      // Since it is being enqueued immediately, no point in tracing the future for cancellation.
+      Future<?> unused = schedule(command, 0, TimeUnit.NANOSECONDS);
     }
   }
 

core/src/test/java/io/grpc/internal/FakeClockTest.java

@@ -95,7 +95,7 @@ public class FakeClockTest {
   public void testScheduledExecutorService_result() {
     FakeClock fakeClock = new FakeClock();
     final boolean[] result = new boolean[]{false};
-    fakeClock.getScheduledExecutorService().schedule(
+    ScheduledFuture<?> unused = fakeClock.getScheduledExecutorService().schedule(
         new Runnable() {
           @Override
           public void run() {
@@ -137,6 +137,7 @@ public class FakeClockTest {
   }
 
   @Test
+  @SuppressWarnings("FutureReturnValueIgnored")
   public void testPendingAndDueTasks() {
     FakeClock fakeClock = new FakeClock();
     ScheduledExecutorService scheduledExecutorService = fakeClock.getScheduledExecutorService();

interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java

@@ -61,6 +61,7 @@ import java.util.List;
 import java.util.Queue;
 import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.concurrent.GuardedBy;
@@ -378,7 +379,8 @@ public class TestServiceImpl extends TestServiceGrpc.TestServiceImplBase {
         Chunk nextChunk = chunks.peek();
         if (nextChunk != null) {
           scheduled = true;
-          executor.schedule(new LogExceptionRunnable(dispatchTask),
+          // TODO(ejona): cancel future if RPC is cancelled
+          Future<?> unused = executor.schedule(new LogExceptionRunnable(dispatchTask),
               nextChunk.delayMicroseconds, TimeUnit.MICROSECONDS);
           return;
         }

interop-testing/src/test/java/io/grpc/testing/integration/ProxyTest.java

@@ -79,7 +79,7 @@ public class ProxyTest {
       throws UnknownHostException, IOException, InterruptedException, ExecutionException {
     server = new Server();
     int serverPort = server.init();
-    executor.submit(server);
+    executor.execute(server);
 
     int latency = (int) TimeUnit.MILLISECONDS.toNanos(50);
     proxy = new TrafficControlProxy(serverPort, 1024 * 1024, latency, TimeUnit.NANOSECONDS);
@@ -111,7 +111,7 @@ public class ProxyTest {
       throws UnknownHostException, IOException, InterruptedException, ExecutionException {
     server = new Server();
     int serverPort = server.init();
-    executor.submit(server);
+    executor.execute(server);
 
     int latency = (int) TimeUnit.MILLISECONDS.toNanos(250);
     proxy = new TrafficControlProxy(serverPort, 1024 * 1024, latency, TimeUnit.NANOSECONDS);
@@ -143,7 +143,7 @@ public class ProxyTest {
     server = new Server();
     int serverPort = server.init();
     server.setMode("stream");
-    executor.submit(server);
+    executor.execute(server);
     assertEquals(server.mode(), "stream");
 
     int bandwidth = 64 * 1024;
@@ -169,7 +169,7 @@ public class ProxyTest {
     server = new Server();
     int serverPort = server.init();
     server.setMode("stream");
-    executor.submit(server);
+    executor.execute(server);
     assertEquals(server.mode(), "stream");
     int bandwidth = 10 * 1024 * 1024;
     proxy = new TrafficControlProxy(serverPort, bandwidth, 200, TimeUnit.MILLISECONDS);

interop-testing/src/test/java/io/grpc/testing/integration/TrafficControlProxy.java

@@ -101,7 +101,7 @@ public final class TrafficControlProxy {
     // client normally would.
     clientAcceptor = new ServerSocket();
     clientAcceptor.bind(new InetSocketAddress(localhost, 0));
-    executor.submit(new Runnable() {
+    executor.execute(new Runnable() {
       @Override
       public void run() {
         try {
@@ -143,10 +143,10 @@ public final class TrafficControlProxy {
     MessageQueue clientPipe = new MessageQueue(clientIn, clientOut);
     MessageQueue serverPipe = new MessageQueue(serverIn, serverOut);
 
-    executor.submit(new Thread(new Reader(clientPipe)));
+    executor.execute(new Thread(new Reader(clientPipe)));
-    executor.submit(new Thread(new Writer(clientPipe)));
+    executor.execute(new Thread(new Writer(clientPipe)));
-    executor.submit(new Thread(new Reader(serverPipe)));
+    executor.execute(new Thread(new Reader(serverPipe)));
-    executor.submit(new Thread(new Writer(serverPipe)));
+    executor.execute(new Thread(new Writer(serverPipe)));
   }
 
   private final class Reader implements Runnable {