Repository: kagkarlsson/db-scheduler (java)

Issue:
Add a `maxRetriesExceededHandler` to the `MaxRetriesFailureHandler`
### Discussed in https://github.com/kagkarlsson/db-scheduler/discussions/394

<div type='discussions-op-text'>

<sup>Originally posted by **piotrooo** May 24, 2023</sup>
Hello!

What do you think about add a `maxRetriesExceededHandler` feature to the `MaxRetriesFailureHandler`. The reason behind this feature is to act some action when retries handler is exceeded. That's could be a simple consumer (`BiConsumer` - we have to pass a `ExecutionComplete` and `ExecutionOperations`). 
Default value should be a noop consumer.

PoC:

```java
class MaxRetriesFailureHandler<T> implements FailureHandler<T> {
    private static final Logger LOG = LoggerFactory.getLogger(MaxRetriesFailureHandler.class);
    private final int maxRetries;
    private final FailureHandler<T> failureHandler;
    private final BiConsumer<ExecutionComplete, ExecutionOperations<T>> maxRetriesExceededHandler;

    public MaxRetriesFailureHandler(int maxRetries, FailureHandler<T> failureHandler) {
        this(maxRetries, failureHandler, (executionComplete, executionOperations) -> {
        });
    }

    public MaxRetriesFailureHandler(int maxRetries, FailureHandler<T> failureHandler, BiConsumer<ExecutionComplete, ExecutionOperations<T>> maxRetriesExceededHandler) {
        this.maxRetries = maxRetries;
        this.failureHandler = failureHandler;
        this.maxRetriesExceededHandler = maxRetriesExceededHandler;
    }

    @Override
    public void onFailure(
            final ExecutionComplete executionComplete,
            final ExecutionOperations<T> executionOperations) {
        int consecutiveFailures = executionComplete.getExecution().consecutiveFailures;
        int totalNumberOfFailures = consecutiveFailures + 1;
        if (totalNumberOfFailures > maxRetries) {
            LOG.error(
                    "Execution has failed {} times for task instance {}. Cancelling execution.",
                    totalNumberOfFailures,
                    executionComplete.getExecution().taskInstance);
            executionOperations.stop();
            maxRetriesExceededHandler.accept(executionComplete, executionOperations);
        } else {
            this.failureHandler.onFailure(executionComplete, executionOperations);
        }
    }
}
```

Thanks for a great library!</div>


Discussion:

Patch:
diff --git a/db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/FailureHandler.java b/db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/FailureHandler.java
--- a/db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/FailureHandler.java
+++ b/db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/task/FailureHandler.java
@@ -20,6 +20,7 @@
 import com.github.kagkarlsson.scheduler.task.schedule.Schedule;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.function.BiConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,10 +67,19 @@ class MaxRetriesFailureHandler<T> implements FailureHandler<T> {
     private static final Logger LOG = LoggerFactory.getLogger(MaxRetriesFailureHandler.class);
     private final int maxRetries;
     private final FailureHandler<T> failureHandler;
+    private final BiConsumer<ExecutionComplete, ExecutionOperations<T>> maxRetriesExceededHandler;
 
     public MaxRetriesFailureHandler(int maxRetries, FailureHandler<T> failureHandler) {
+      this(maxRetries, failureHandler, (executionComplete, executionOperations) -> {});
+    }
+
+    public MaxRetriesFailureHandler(
+        int maxRetries,
+        FailureHandler<T> failureHandler,
+        BiConsumer<ExecutionComplete, ExecutionOperations<T>> maxRetriesExceededHandler) {
       this.maxRetries = maxRetries;
       this.failureHandler = failureHandler;
+      this.maxRetriesExceededHandler = maxRetriesExceededHandler;
     }
 
     @Override
@@ -84,6 +94,7 @@ public void onFailure(
             totalNumberOfFailures,
             executionComplete.getExecution().taskInstance);
         executionOperations.stop();
+        maxRetriesExceededHandler.accept(executionComplete, executionOperations);
       } else {
         this.failureHandler.onFailure(executionComplete, executionOperations);
       }


Test patch:
diff --git a/db-scheduler/src/test/java/com/github/kagkarlsson/scheduler/task/FailureHandlerTest.java b/db-scheduler/src/test/java/com/github/kagkarlsson/scheduler/task/FailureHandlerTest.java
new file mode 100644
--- /dev/null
+++ b/db-scheduler/src/test/java/com/github/kagkarlsson/scheduler/task/FailureHandlerTest.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) Gustav Karlsson
+ *
+ * <p>Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
+ * except in compliance with the License. You may obtain a copy of the License at
+ *
+ * <p>http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * <p>Unless required by applicable law or agreed to in writing, software distributed under the
+ * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.github.kagkarlsson.scheduler.task;
+
+import static java.time.Instant.now;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.mockito.Mockito.*;
+
+import com.github.kagkarlsson.scheduler.TestTasks;
+import com.github.kagkarlsson.scheduler.task.FailureHandler.MaxRetriesFailureHandler;
+import com.github.kagkarlsson.scheduler.task.helper.OneTimeTask;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+class FailureHandlerTest {
+  @Nested
+  class MaxRetriesFailureHandlerTest {
+    private final OneTimeTask<Integer> task =
+        TestTasks.oneTime("some-task", Integer.class, (instance, executionContext) -> {});
+
+    private final AtomicBoolean failureHandlerCalled = new AtomicBoolean(false);
+    private final FailureHandler<String> failureHandler =
+        (executionComplete, executionOperations) -> failureHandlerCalled.set(true);
+    private final AtomicBoolean maxRetriesExceededHandlerCalled = new AtomicBoolean(false);
+    private final BiConsumer<ExecutionComplete, ExecutionOperations<String>>
+        maxRetriesExceededHandler =
+            (executionComplete, stringExecutionOperations) ->
+                maxRetriesExceededHandlerCalled.set(true);
+
+    private final ExecutionComplete executionComplete = mock(ExecutionComplete.class);
+    private final ExecutionOperations<String> executionOperations = mock(ExecutionOperations.class);
+
+    @Test
+    void should_run_handler_when_max_retries_exceeded() {
+      MaxRetriesFailureHandler<String> maxRetriesFailureHandler =
+          new MaxRetriesFailureHandler<>(3, failureHandler, maxRetriesExceededHandler);
+
+      Execution execution = getExecutionWithFails(3);
+      when(executionComplete.getExecution()).thenReturn(execution);
+
+      maxRetriesFailureHandler.onFailure(executionComplete, executionOperations);
+
+      assertThat(failureHandlerCalled.get(), is(false));
+      assertThat(maxRetriesExceededHandlerCalled.get(), is(true));
+      verify(executionOperations).stop();
+    }
+
+    @Test
+    void should_do_not_run_max_retries_handler_when_retries_are_not_exceeded() {
+      MaxRetriesFailureHandler<String> maxRetriesFailureHandler =
+          new MaxRetriesFailureHandler<>(3, failureHandler, maxRetriesExceededHandler);
+
+      Execution execution = getExecutionWithFails(1);
+      when(executionComplete.getExecution()).thenReturn(execution);
+
+      maxRetriesFailureHandler.onFailure(executionComplete, executionOperations);
+
+      assertThat(failureHandlerCalled.get(), is(true));
+      assertThat(maxRetriesExceededHandlerCalled.get(), is(false));
+      verify(executionOperations, never()).stop();
+    }
+
+    private Execution getExecutionWithFails(int consecutiveFailures) {
+      return new Execution(
+          now(), task.instance("1"), false, null, null, null, consecutiveFailures, null, 1L);
+    }
+  }
+}