Repository: eclipse-vertx/vert.x

Problem statement:
TasksQueue doesn't remove entry from tasks on RejectedExecutionException
### Version

`master` branch.

### Context

I had a look at the code when analyzing a problem in the 4.4.x after I configured a thread pool with a limited backlog which would reject executions eventually.

I found that the code in `master` throws the exception that the task is rejected, but then the task is still in the `taskQueue`, so it will eventually be executed when another task is then successfully submitted.
So for the caller it is both rejected and accepted, which is confusing and IMHO wrong.

I'll create a different issue for that as it doesn't handle the RejectedExecutionException at all.

### Do you have a reproducer?

Here a unit test as a reproducer - not that (at least for the reproducer) I made the `tasks` attribute public to I can use it the test.  See https://github.com/ahus1/vert.x/tree/poc-show-how-taskqueue-fails-in-main

```
    TaskQueue taskQueue = new TaskQueue();
    Executor executorThatAlwaysThrowsRejectedExceptions = new Executor() {
        @Override
        public void execute(Runnable command) {
            throw new RejectedExecutionException();
        }
    };

    assertThatThrownBy(() -> {
        taskQueue.execute(new Thread(), executorThatAlwaysThrowsRejectedExceptions);
      }).isInstanceOf(RejectedExecutionException.class);

    // this fails in main
    Assertions.assertThat(taskQueue.tasks).isEmpty();
```

### Steps to reproduce

See test.

### Extra

I can start writing a test, but I'm not sure on how to test this.



Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/impl/TaskQueue.java b/src/main/java/io/vertx/core/impl/TaskQueue.java
--- a/src/main/java/io/vertx/core/impl/TaskQueue.java
+++ b/src/main/java/io/vertx/core/impl/TaskQueue.java
@@ -16,6 +16,7 @@
 
 import java.util.LinkedList;
 import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.function.Consumer;
 
 /**
@@ -124,11 +125,27 @@ public Consumer<Runnable> unschedule() {
    */
   public void execute(Runnable task, Executor executor) {
     synchronized (tasks) {
-      tasks.add(new ExecuteTask(task, executor));
-      if (this.currentExecutor == null) {
-        this.currentExecutor = executor;
-        executor.execute(runner);
+      if (currentExecutor == null) {
+        currentExecutor = executor;
+        try {
+          executor.execute(runner);
+        } catch (RejectedExecutionException e) {
+          currentExecutor = null;
+          throw e;
+        }
       }
+      // Add the task after the runner has been accepted to the executor
+      // to cover the case of a rejected execution exception.
+      tasks.add(new ExecuteTask(task, executor));
+    }
+  }
+
+  /**
+   * Test if the task queue is empty and no current executor is running anymore.
+   */
+  public boolean isEmpty() {
+    synchronized (tasks) {
+      return tasks.isEmpty() && currentExecutor == null;
     }
   }
 


Test patch:
diff --git a/src/test/java/io/vertx/core/impl/TaskQueueTest.java b/src/test/java/io/vertx/core/impl/TaskQueueTest.java
new file mode 100644
--- /dev/null
+++ b/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();
+  }
+
+}