address comments from @dapengzhang0 and @carl-mastrangelo

including interruption handling fix

Author: njhill

stub/src/main/java/io/grpc/stub/ClientCalls.java

@@ -642,12 +642,15 @@ private static final class ThreadlessExecutor extends ConcurrentLinkedQueue<Runn
      * Waits until there is a Runnable, then executes it and all queued Runnables after it.
      */
     public void waitAndDrain() throws InterruptedException {
+      final Thread currentThread = Thread.currentThread();
+      throwIfInterrupted(currentThread);
       Runnable runnable = poll();
       if (runnable == null) {
-        waiter = Thread.currentThread();
+        waiter = currentThread;
         try {
           while ((runnable = poll()) == null) {
             LockSupport.park(this);
+            throwIfInterrupted(currentThread);
           }
         } finally {
           waiter = null;
@@ -662,13 +665,16 @@ public void waitAndDrain() throws InterruptedException {
       } while ((runnable = poll()) != null);
     }
 
+    private static void throwIfInterrupted(Thread currentThread) throws InterruptedException {
+      if (currentThread.isInterrupted()) {
+        throw new InterruptedException();
+      }
+    }
+
     @Override
     public void execute(Runnable runnable) {
       add(runnable);
-      final Thread waitingThread = waiter;
-      if (waitingThread != null) {
-        LockSupport.unpark(waitingThread);
-      }
+      LockSupport.unpark(waiter); // no-op if null
     }
   }
 }