Ensure that the task is not in the tasks queue when it is rejected fr…

…om the executor

Closes eclipse-vertx#4900

src/main/java/io/vertx/core/impl/TaskQueue.java

@@ -125,19 +125,30 @@ public Consumer<Runnable> unschedule() {
    */
   public void execute(Runnable task, Executor executor) {
     synchronized (tasks) {
-      tasks.add(new ExecuteTask(task, executor));
+      ExecuteTask executeTask = new ExecuteTask(task, executor);
+      tasks.add(executeTask);
       if (currentExecutor == null) {
         currentExecutor = executor;
         try {
           executor.execute(runner);
         } catch (RejectedExecutionException e) {
           currentExecutor = null;
+          tasks.remove(executeTask);
           throw e;
         }
       }
     }
   }
 
+  /**
+   * Test if the task queue is empty and no current executor is running anymore.
+   */
+  public boolean isEmpty() {
+    synchronized (tasks) {
+      return tasks.isEmpty() && currentExecutor == null;
+    }
+  }
+
   /**
    * A task of this queue.
    */

src/test/java/io/vertx/core/impl/TaskQueueTest.java

@@ -0,0 +1,34 @@
+package io.vertx.core.impl;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * @author Alexander Schwartz
+ */
+public class TaskQueueTest {
+
+  Executor executorThatAlwaysThrowsRejectedExceptions = new Executor() {
+    @Override
+    public void execute(Runnable command) {
+      throw new RejectedExecutionException();
+    }
+  };
+
+  TaskQueue taskQueue = new TaskQueue();
+
+  @Test
+  public void shouldNotHaveTaskInQueueWhenTaskHasBeenRejected() {
+    assertThatThrownBy(
+      () -> taskQueue.execute(new Thread(), executorThatAlwaysThrowsRejectedExceptions)
+    ).isInstanceOf(RejectedExecutionException.class);
+
+    Assertions.assertThat(taskQueue.isEmpty()).isTrue();
+  }
+
+}