diff --git a/docs/documentation/features.md b/docs/documentation/features.md
index 95924e24dd..35deab4664 100644
--- a/docs/documentation/features.md
+++ b/docs/documentation/features.md
@@ -275,9 +275,9 @@ In order to facilitate error reporting Reconciler can implement the following
 [interface](https://github.com/java-operator-sdk/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusHandler.java):
 
 ```java
-public interface ErrorStatusHandler<T extends HasMetadata> {
+public interface ErrorStatusHandler<P extends HasMetadata> {
 
-    Optional<T> updateErrorStatus(T resource, RetryInfo retryInfo, RuntimeException e);
+    ErrorStatusUpdateControl<P> updateErrorStatus(P resource, Context<P> context, Exception e);
 
 }
 ```
@@ -294,6 +294,12 @@ will also produce an event, and will result in a reconciliation if the controlle
 The scope of this feature is only the `reconcile` method of the reconciler, since there should not be updates on custom
 resource after it is marked for deletion.
 
+Retry can be skipped for the cases of unrecoverable errors:
+
+```java
+ ErrorStatusUpdateControl.updateStatus(customResource).withNoRetry();
+```
+
 ### Correctness and Automatic Retries
 
 There is a possibility to turn off the automatic retries. This is not desirable, unless there is a very specific reason.
diff --git a/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetrics.java b/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetrics.java
index d08f1c7669..25058e198e 100644
--- a/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetrics.java
+++ b/micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetrics.java
@@ -6,6 +6,7 @@
 import java.util.Map;
 import java.util.Optional;
 
+import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
 import io.javaoperatorsdk.operator.processing.event.Event;
@@ -33,7 +34,13 @@ public <T> T timeControllerExecution(ControllerExecution<T> execution) {
             .publishPercentileHistogram()
             .register(registry);
     try {
-      final var result = timer.record(execution::execute);
+      final var result = timer.record(() -> {
+        try {
+          return execution.execute();
+        } catch (Exception e) {
+          throw new OperatorException(e);
+        }
+      });
       final var successType = execution.successTypeName(result);
       registry
           .counter(execName + ".success", "controller", name, "type", successType)
@@ -58,6 +65,7 @@ public void cleanupDoneFor(ResourceID customResourceUid) {
     incrementCounter(customResourceUid, "events.delete");
   }
 
+  @Override
   public void reconcileCustomResource(ResourceID resourceID, RetryInfo retryInfoNullable) {
     Optional<RetryInfo> retryInfo = Optional.ofNullable(retryInfoNullable);
     incrementCounter(resourceID, RECONCILIATIONS + "started",
@@ -72,7 +80,7 @@ public void finishedReconciliation(ResourceID resourceID) {
     incrementCounter(resourceID, RECONCILIATIONS + "success");
   }
 
-  public void failedReconciliation(ResourceID resourceID, RuntimeException exception) {
+  public void failedReconciliation(ResourceID resourceID, Exception exception) {
     var cause = exception.getCause();
     if (cause == null) {
       cause = exception;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/OperatorException.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/OperatorException.java
index b11ff0a61a..895d643fcb 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/OperatorException.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/OperatorException.java
@@ -8,6 +8,10 @@ public OperatorException(String message) {
     super(message);
   }
 
+  public OperatorException(Throwable cause) {
+    super(cause);
+  }
+
   public OperatorException(String message, Throwable cause) {
     super(message, cause);
   }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java
index d7daba49ae..b1be115bc4 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java
@@ -13,7 +13,7 @@ default void receivedEvent(Event event) {}
 
   default void reconcileCustomResource(ResourceID resourceID, RetryInfo retryInfo) {}
 
-  default void failedReconciliation(ResourceID resourceID, RuntimeException exception) {}
+  default void failedReconciliation(ResourceID resourceID, Exception exception) {}
 
   default void cleanupDoneFor(ResourceID customResourceUid) {}
 
@@ -27,10 +27,10 @@ interface ControllerExecution<T> {
 
     String successTypeName(T result);
 
-    T execute();
+    T execute() throws Exception;
   }
 
-  default <T> T timeControllerExecution(ControllerExecution<T> execution) {
+  default <T> T timeControllerExecution(ControllerExecution<T> execution) throws Exception {
     return execution.execute();
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java
index d3a5b7bc18..d80be8b3ae 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java
@@ -9,7 +9,7 @@
 
 public class DefaultContext<P extends HasMetadata> implements Context<P> {
 
-  private final RetryInfo retryInfo;
+  private RetryInfo retryInfo;
   private final Controller<P> controller;
   private final P primaryResource;
   private final ControllerConfiguration<P> controllerConfiguration;
@@ -44,4 +44,9 @@ public ControllerConfiguration<P> getControllerConfiguration() {
   public ManagedDependentResourceContext managedDependentResourceContext() {
     return managedDependentResourceContext;
   }
+
+  public DefaultContext<P> setRetryInfo(RetryInfo retryInfo) {
+    this.retryInfo = retryInfo;
+    return this;
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusHandler.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusHandler.java
index 22a16e4ccd..c7bd09a930 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusHandler.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusHandler.java
@@ -1,16 +1,14 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
-import java.util.Optional;
-
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
-public interface ErrorStatusHandler<T extends HasMetadata> {
+public interface ErrorStatusHandler<P extends HasMetadata> {
 
   /**
    * <p>
    * Reconciler can implement this interface in order to update the status sub-resource in the case
    * an exception in thrown. In that case
-   * {@link #updateErrorStatus(HasMetadata, RetryInfo, RuntimeException)} is called automatically.
+   * {@link #updateErrorStatus(HasMetadata, Context, Exception)} is called automatically.
    * <p>
    * The result of the method call is used to make a status update on the custom resource. This is
    * always a sub-resource update request, so no update on custom resource itself (like spec of
@@ -21,10 +19,10 @@ public interface ErrorStatusHandler<T extends HasMetadata> {
    * should not be updates on custom resource after it is marked for deletion.
    *
    * @param resource to update the status on
-   * @param retryInfo the current retry status
+   * @param context the current context
    * @param e exception thrown from the reconciler
    * @return the updated resource
    */
-  Optional<T> updateErrorStatus(T resource, RetryInfo retryInfo, RuntimeException e);
+  ErrorStatusUpdateControl<P> updateErrorStatus(P resource, Context<P> context, Exception e);
 
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusUpdateControl.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusUpdateControl.java
new file mode 100644
index 0000000000..beb8ddc028
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ErrorStatusUpdateControl.java
@@ -0,0 +1,41 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public class ErrorStatusUpdateControl<P extends HasMetadata> {
+
+  private final P resource;
+  private boolean noRetry = false;
+
+  public static <T extends HasMetadata> ErrorStatusUpdateControl<T> updateStatus(T resource) {
+    return new ErrorStatusUpdateControl<>(resource);
+  }
+
+  public static <T extends HasMetadata> ErrorStatusUpdateControl<T> noStatusUpdate() {
+    return new ErrorStatusUpdateControl<>(null);
+  }
+
+  private ErrorStatusUpdateControl(P resource) {
+    this.resource = resource;
+  }
+
+  /**
+   * Instructs the controller to not retry the error. This is useful for non-recoverable errors.
+   *
+   * @return ErrorStatusUpdateControl
+   */
+  public ErrorStatusUpdateControl<P> withNoRetry() {
+    this.noRetry = true;
+    return this;
+  }
+
+  public Optional<P> getResource() {
+    return Optional.ofNullable(resource);
+  }
+
+  public boolean isNoRetry() {
+    return noRetry;
+  }
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Reconciler.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Reconciler.java
index 0be4d58ae5..446df36ca1 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Reconciler.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Reconciler.java
@@ -13,7 +13,7 @@ public interface Reconciler<R extends HasMetadata> {
    * @return UpdateControl to manage updates on the custom resource (usually the status) after
    *         reconciliation.
    */
-  UpdateControl<R> reconcile(R resource, Context<R> context);
+  UpdateControl<R> reconcile(R resource, Context<R> context) throws Exception;
 
   /**
    * Note that this method is used in combination with finalizers. If automatic finalizer handling
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java
index d769324be9..5fc1b9ede7 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java
@@ -60,32 +60,36 @@ public Controller(Reconciler<P> reconciler,
   public DeleteControl cleanup(P resource, Context<P> context) {
     dependents.cleanup(resource, context);
 
-    return metrics().timeControllerExecution(
-        new ControllerExecution<>() {
-          @Override
-          public String name() {
-            return "cleanup";
-          }
+    try {
+      return metrics().timeControllerExecution(
+          new ControllerExecution<>() {
+            @Override
+            public String name() {
+              return "cleanup";
+            }
 
-          @Override
-          public String controllerName() {
-            return configuration.getName();
-          }
+            @Override
+            public String controllerName() {
+              return configuration.getName();
+            }
 
-          @Override
-          public String successTypeName(DeleteControl deleteControl) {
-            return deleteControl.isRemoveFinalizer() ? "delete" : "finalizerNotRemoved";
-          }
+            @Override
+            public String successTypeName(DeleteControl deleteControl) {
+              return deleteControl.isRemoveFinalizer() ? "delete" : "finalizerNotRemoved";
+            }
 
-          @Override
-          public DeleteControl execute() {
-            return reconciler.cleanup(resource, context);
-          }
-        });
+            @Override
+            public DeleteControl execute() {
+              return reconciler.cleanup(resource, context);
+            }
+          });
+    } catch (Exception e) {
+      throw new OperatorException(e);
+    }
   }
 
   @Override
-  public UpdateControl<P> reconcile(P resource, Context<P> context) {
+  public UpdateControl<P> reconcile(P resource, Context<P> context) throws Exception {
     dependents.reconcile(resource, context);
 
     return metrics().timeControllerExecution(
@@ -113,7 +117,7 @@ public String successTypeName(UpdateControl<P> result) {
           }
 
           @Override
-          public UpdateControl<P> execute() {
+          public UpdateControl<P> execute() throws Exception {
             return reconciler.reconcile(resource, context);
           }
         });
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
index e44df906d8..4e7ffa4660 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java
@@ -242,7 +242,7 @@ TimerEventSource<R> retryEventSource() {
    * according to the retry timing if there was an exception.
    */
   private void handleRetryOnException(
-      ExecutionScope<R> executionScope, RuntimeException exception) {
+      ExecutionScope<R> executionScope, Exception exception) {
     RetryExecution execution = getOrInitRetryExecution(executionScope);
     var customResourceID = executionScope.getCustomResourceID();
     boolean eventPresent = eventMarker.eventPresent(customResourceID);
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/PostExecutionControl.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/PostExecutionControl.java
index 5d9a623e20..1171a6d03c 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/PostExecutionControl.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/PostExecutionControl.java
@@ -8,14 +8,14 @@ final class PostExecutionControl<R extends HasMetadata> {
 
   private final boolean onlyFinalizerHandled;
   private final R updatedCustomResource;
-  private final RuntimeException runtimeException;
+  private final Exception runtimeException;
 
   private Long reScheduleDelay = null;
 
   private PostExecutionControl(
       boolean onlyFinalizerHandled,
       R updatedCustomResource,
-      RuntimeException runtimeException) {
+      Exception runtimeException) {
     this.onlyFinalizerHandled = onlyFinalizerHandled;
     this.updatedCustomResource = updatedCustomResource;
     this.runtimeException = runtimeException;
@@ -35,7 +35,7 @@ public static <R extends HasMetadata> PostExecutionControl<R> customResourceUpda
   }
 
   public static <R extends HasMetadata> PostExecutionControl<R> exceptionDuringExecution(
-      RuntimeException exception) {
+      Exception exception) {
     return new PostExecutionControl<>(false, null, exception);
   }
 
@@ -60,7 +60,7 @@ public PostExecutionControl<R> withReSchedule(long delay) {
     return this;
   }
 
-  public Optional<RuntimeException> getRuntimeException() {
+  public Optional<Exception> getRuntimeException() {
     return Optional.ofNullable(runtimeException);
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java
index 088592b3a4..180fbc2d28 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java
@@ -48,13 +48,14 @@ public ReconciliationDispatcher(Controller<R> controller) {
   public PostExecutionControl<R> handleExecution(ExecutionScope<R> executionScope) {
     try {
       return handleDispatch(executionScope);
-    } catch (RuntimeException e) {
+    } catch (Exception e) {
       log.error("Error during event processing {} failed.", executionScope, e);
       return PostExecutionControl.exceptionDuringExecution(e);
     }
   }
 
-  private PostExecutionControl<R> handleDispatch(ExecutionScope<R> executionScope) {
+  private PostExecutionControl<R> handleDispatch(ExecutionScope<R> executionScope)
+      throws Exception {
     R resource = executionScope.getResource();
     log.debug("Handling dispatch for resource {}", getName(resource));
 
@@ -67,7 +68,7 @@ private PostExecutionControl<R> handleDispatch(ExecutionScope<R> executionScope)
       return PostExecutionControl.defaultDispatch();
     }
 
-    Context context = new DefaultContext<>(executionScope.getRetryInfo(), controller, resource);
+    Context<R> context = new DefaultContext<>(executionScope.getRetryInfo(), controller, resource);
     if (markedForDeletion) {
       return handleCleanup(resource, context);
     } else {
@@ -91,7 +92,7 @@ private boolean shouldNotDispatchToDelete(R resource) {
   }
 
   private PostExecutionControl<R> handleReconcile(
-      ExecutionScope<R> executionScope, R originalResource, Context context) {
+      ExecutionScope<R> executionScope, R originalResource, Context<R> context) throws Exception {
     if (configuration().useFinalizer()
         && !originalResource.hasFinalizer(configuration().getFinalizer())) {
       /*
@@ -105,11 +106,10 @@ private PostExecutionControl<R> handleReconcile(
     } else {
       try {
         var resourceForExecution =
-            cloneResourceForErrorStatusHandlerIfNeeded(originalResource, context);
+            cloneResourceForErrorStatusHandlerIfNeeded(originalResource);
         return reconcileExecution(executionScope, resourceForExecution, originalResource, context);
-      } catch (RuntimeException e) {
-        handleErrorStatusHandler(originalResource, context, e);
-        throw e;
+      } catch (Exception e) {
+        return handleErrorStatusHandler(originalResource, context, e);
       }
     }
   }
@@ -121,7 +121,7 @@ private PostExecutionControl<R> handleReconcile(
    * resource is changed during an execution, and it's much cleaner to have to original resource in
    * place for status update.
    */
-  private R cloneResourceForErrorStatusHandlerIfNeeded(R resource, Context context) {
+  private R cloneResourceForErrorStatusHandlerIfNeeded(R resource) {
     if (isErrorStatusHandlerPresent() ||
         shouldUpdateObservedGenerationAutomatically(resource)) {
       final var cloner = ConfigurationServiceProvider.instance().getResourceCloner();
@@ -132,7 +132,7 @@ private R cloneResourceForErrorStatusHandlerIfNeeded(R resource, Context context
   }
 
   private PostExecutionControl<R> reconcileExecution(ExecutionScope<R> executionScope,
-      R resourceForExecution, R originalResource, Context context) {
+      R resourceForExecution, R originalResource, Context<R> context) throws Exception {
     log.debug(
         "Reconciling resource {} with version: {} with execution scope: {}",
         getName(resourceForExecution),
@@ -163,8 +163,8 @@ && shouldUpdateObservedGenerationAutomatically(resourceForExecution)) {
   }
 
   @SuppressWarnings("unchecked")
-  private void handleErrorStatusHandler(R resource, Context<R> context,
-      RuntimeException e) {
+  private PostExecutionControl<R> handleErrorStatusHandler(R resource, Context<R> context,
+      Exception e) throws Exception {
     if (isErrorStatusHandlerPresent()) {
       try {
         RetryInfo retryInfo = context.getRetryInfo().orElse(new RetryInfo() {
@@ -178,13 +178,28 @@ public boolean isLastAttempt() {
             return controller.getConfiguration().getRetryConfiguration() == null;
           }
         });
-        var updatedResource = ((ErrorStatusHandler<R>) controller.getReconciler())
-            .updateErrorStatus(resource, retryInfo, e);
-        updatedResource.ifPresent(customResourceFacade::updateStatus);
+        ((DefaultContext<R>) context).setRetryInfo(retryInfo);
+        var errorStatusUpdateControl = ((ErrorStatusHandler<R>) controller.getReconciler())
+            .updateErrorStatus(resource, context, e);
+
+        R updatedResource = null;
+        if (errorStatusUpdateControl.getResource().isPresent()) {
+          updatedResource =
+              customResourceFacade
+                  .updateStatus(errorStatusUpdateControl.getResource().orElseThrow());
+        }
+        if (errorStatusUpdateControl.isNoRetry()) {
+          if (updatedResource != null) {
+            return PostExecutionControl.customResourceUpdated(updatedResource);
+          } else {
+            return PostExecutionControl.defaultDispatch();
+          }
+        }
       } catch (RuntimeException ex) {
         log.error("Error during error status handling.", ex);
       }
     }
+    throw e;
   }
 
   private boolean isErrorStatusHandlerPresent() {
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java
index 5e230d0418..2e6ee4aa0d 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java
@@ -22,13 +22,7 @@
 import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.RetryConfiguration;
-import io.javaoperatorsdk.operator.api.reconciler.Constants;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
-import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
-import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
-import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.*;
 import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.CustomResourceFacade;
 import io.javaoperatorsdk.operator.sample.observedgeneration.ObservedGenCustomResource;
@@ -116,7 +110,7 @@ private <R extends HasMetadata> ReconciliationDispatcher<R> init(R customResourc
   }
 
   @Test
-  void addFinalizerOnNewResource() {
+  void addFinalizerOnNewResource() throws Exception {
     assertFalse(testCustomResource.hasFinalizer(DEFAULT_FINALIZER));
     reconciliationDispatcher.handleExecution(executionScopeWithCREvent(testCustomResource));
     verify(reconciler, never())
@@ -128,7 +122,7 @@ void addFinalizerOnNewResource() {
   }
 
   @Test
-  void callCreateOrUpdateOnNewResourceIfFinalizerSet() {
+  void callCreateOrUpdateOnNewResourceIfFinalizerSet() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
     reconciliationDispatcher.handleExecution(executionScopeWithCREvent(testCustomResource));
     verify(reconciler, times(1))
@@ -136,7 +130,7 @@ void callCreateOrUpdateOnNewResourceIfFinalizerSet() {
   }
 
   @Test
-  void updatesOnlyStatusSubResourceIfFinalizerSet() {
+  void updatesOnlyStatusSubResourceIfFinalizerSet() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciler.reconcile = (r, c) -> UpdateControl.updateStatus(testCustomResource);
@@ -148,7 +142,7 @@ void updatesOnlyStatusSubResourceIfFinalizerSet() {
   }
 
   @Test
-  void updatesBothResourceAndStatusIfFinalizerSet() {
+  void updatesBothResourceAndStatusIfFinalizerSet() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciler.reconcile = (r, c) -> UpdateControl.updateResourceAndStatus(testCustomResource);
@@ -161,7 +155,7 @@ void updatesBothResourceAndStatusIfFinalizerSet() {
   }
 
   @Test
-  void callCreateOrUpdateOnModifiedResourceIfFinalizerSet() {
+  void callCreateOrUpdateOnModifiedResourceIfFinalizerSet() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciliationDispatcher.handleExecution(executionScopeWithCREvent(testCustomResource));
@@ -235,7 +229,7 @@ void doesNotRemovesTheSetFinalizerIfTheDeleteNotMethodInstructsIt() {
   }
 
   @Test
-  void doesNotUpdateTheResourceIfNoUpdateUpdateControlIfFinalizerSet() {
+  void doesNotUpdateTheResourceIfNoUpdateUpdateControlIfFinalizerSet() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciler.reconcile = (r, c) -> UpdateControl.noUpdate();
@@ -246,7 +240,7 @@ void doesNotUpdateTheResourceIfNoUpdateUpdateControlIfFinalizerSet() {
   }
 
   @Test
-  void addsFinalizerIfNotMarkedForDeletionAndEmptyCustomResourceReturned() {
+  void addsFinalizerIfNotMarkedForDeletionAndEmptyCustomResourceReturned() throws Exception {
     removeFinalizers(testCustomResource);
 
     reconciler.reconcile = (r, c) -> UpdateControl.noUpdate();
@@ -269,7 +263,8 @@ void doesNotCallDeleteIfMarkedForDeletionButNotOurFinalizer() {
   }
 
   @Test
-  void executeControllerRegardlessGenerationInNonGenerationAwareModeIfFinalizerSet() {
+  void executeControllerRegardlessGenerationInNonGenerationAwareModeIfFinalizerSet()
+      throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
     reconciliationDispatcher.handleExecution(executionScopeWithCREvent(testCustomResource));
     reconciliationDispatcher.handleExecution(executionScopeWithCREvent(testCustomResource));
@@ -278,7 +273,7 @@ void executeControllerRegardlessGenerationInNonGenerationAwareModeIfFinalizerSet
   }
 
   @Test
-  void propagatesRetryInfoToContextIfFinalizerSet() {
+  void propagatesRetryInfoToContextIfFinalizerSet() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciliationDispatcher.handleExecution(
@@ -307,7 +302,7 @@ public boolean isLastAttempt() {
   }
 
   @Test
-  void setReScheduleToPostExecutionControlFromUpdateControl() {
+  void setReScheduleToPostExecutionControlFromUpdateControl() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciler.reconcile =
@@ -336,7 +331,7 @@ void reScheduleOnDeleteWithoutFinalizerRemoval() {
   }
 
   @Test
-  void setObservedGenerationForStatusIfNeeded() {
+  void setObservedGenerationForStatusIfNeeded() throws Exception {
     var observedGenResource = createObservedGenCustomResource();
 
     Reconciler<ObservedGenCustomResource> reconciler = mock(Reconciler.class);
@@ -358,7 +353,7 @@ void setObservedGenerationForStatusIfNeeded() {
   }
 
   @Test
-  void updatesObservedGenerationOnNoUpdateUpdateControl() {
+  void updatesObservedGenerationOnNoUpdateUpdateControl() throws Exception {
     var observedGenResource = createObservedGenCustomResource();
 
     Reconciler<ObservedGenCustomResource> reconciler = mock(Reconciler.class);
@@ -379,7 +374,7 @@ void updatesObservedGenerationOnNoUpdateUpdateControl() {
   }
 
   @Test
-  void updateObservedGenerationOnCustomResourceUpdate() {
+  void updateObservedGenerationOnCustomResourceUpdate() throws Exception {
     var observedGenResource = createObservedGenCustomResource();
 
     Reconciler<ObservedGenCustomResource> reconciler = mock(Reconciler.class);
@@ -401,7 +396,7 @@ void updateObservedGenerationOnCustomResourceUpdate() {
   }
 
   @Test
-  void callErrorStatusHandlerIfImplemented() {
+  void callErrorStatusHandlerIfImplemented() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciler.reconcile = (r, c) -> {
@@ -409,7 +404,7 @@ void callErrorStatusHandlerIfImplemented() {
     };
     reconciler.errorHandler = (r, ri, e) -> {
       testCustomResource.getStatus().setConfigMapStatus(ERROR_MESSAGE);
-      return Optional.of(testCustomResource);
+      return ErrorStatusUpdateControl.updateStatus(testCustomResource);
     };
 
     reconciliationDispatcher.handleExecution(
@@ -433,7 +428,7 @@ public boolean isLastAttempt() {
   }
 
   @Test
-  void callErrorStatusHandlerEvenOnFirstError() {
+  void callErrorStatusHandlerEvenOnFirstError() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciler.reconcile = (r, c) -> {
@@ -441,19 +436,62 @@ void callErrorStatusHandlerEvenOnFirstError() {
     };
     reconciler.errorHandler = (r, ri, e) -> {
       testCustomResource.getStatus().setConfigMapStatus(ERROR_MESSAGE);
-      return Optional.of(testCustomResource);
+      return ErrorStatusUpdateControl.updateStatus(testCustomResource);
     };
 
-    reconciliationDispatcher.handleExecution(
+    var postExecControl = reconciliationDispatcher.handleExecution(
+        new ExecutionScope(
+            testCustomResource, null));
+    verify(customResourceFacade, times(1)).updateStatus(testCustomResource);
+    verify(((ErrorStatusHandler) reconciler), times(1)).updateErrorStatus(eq(testCustomResource),
+        any(), any());
+    assertThat(postExecControl.exceptionDuringExecution()).isTrue();
+  }
+
+  @Test
+  void errorHandlerCanInstructNoRetryWithUpdate() {
+    testCustomResource.addFinalizer(DEFAULT_FINALIZER);
+    reconciler.reconcile = (r, c) -> {
+      throw new IllegalStateException("Error Status Test");
+    };
+    reconciler.errorHandler = (r, ri, e) -> {
+      testCustomResource.getStatus().setConfigMapStatus(ERROR_MESSAGE);
+      return ErrorStatusUpdateControl.updateStatus(testCustomResource).withNoRetry();
+    };
+
+    var postExecControl = reconciliationDispatcher.handleExecution(
         new ExecutionScope(
             testCustomResource, null));
+
+    verify(((ErrorStatusHandler) reconciler), times(1)).updateErrorStatus(eq(testCustomResource),
+        any(), any());
     verify(customResourceFacade, times(1)).updateStatus(testCustomResource);
+    assertThat(postExecControl.exceptionDuringExecution()).isFalse();
+  }
+
+  @Test
+  void errorHandlerCanInstructNoRetryNoUpdate() {
+    testCustomResource.addFinalizer(DEFAULT_FINALIZER);
+    reconciler.reconcile = (r, c) -> {
+      throw new IllegalStateException("Error Status Test");
+    };
+    reconciler.errorHandler = (r, ri, e) -> {
+      testCustomResource.getStatus().setConfigMapStatus(ERROR_MESSAGE);
+      return ErrorStatusUpdateControl.<TestCustomResource>noStatusUpdate().withNoRetry();
+    };
+
+    var postExecControl = reconciliationDispatcher.handleExecution(
+        new ExecutionScope(
+            testCustomResource, null));
+
     verify(((ErrorStatusHandler) reconciler), times(1)).updateErrorStatus(eq(testCustomResource),
         any(), any());
+    verify(customResourceFacade, times(0)).updateStatus(testCustomResource);
+    assertThat(postExecControl.exceptionDuringExecution()).isFalse();
   }
 
   @Test
-  void schedulesReconciliationIfMaxDelayIsSet() {
+  void schedulesReconciliationIfMaxDelayIsSet() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciler.reconcile = (r, c) -> UpdateControl.noUpdate();
@@ -466,7 +504,7 @@ void schedulesReconciliationIfMaxDelayIsSet() {
   }
 
   @Test
-  void canSkipSchedulingMaxDelayIf() {
+  void canSkipSchedulingMaxDelayIf() throws Exception {
     testCustomResource.addFinalizer(DEFAULT_FINALIZER);
 
     reconciler.reconcile = (r, c) -> UpdateControl.noUpdate();
@@ -524,10 +562,11 @@ public DeleteControl cleanup(TestCustomResource resource, Context context) {
     }
 
     @Override
-    public Optional<TestCustomResource> updateErrorStatus(TestCustomResource resource,
-        RetryInfo retryInfo, RuntimeException e) {
-      return errorHandler != null ? errorHandler.updateErrorStatus(resource, retryInfo, e)
-          : Optional.empty();
+    public ErrorStatusUpdateControl<TestCustomResource> updateErrorStatus(
+        TestCustomResource resource,
+        Context<TestCustomResource> context, Exception e) {
+      return errorHandler != null ? errorHandler.updateErrorStatus(resource, context, e)
+          : ErrorStatusUpdateControl.noStatusUpdate();
     }
   }
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/errorstatushandler/ErrorStatusHandlerTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/errorstatushandler/ErrorStatusHandlerTestReconciler.java
index c3b8632e02..17ea106104 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/errorstatushandler/ErrorStatusHandlerTestReconciler.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/errorstatushandler/ErrorStatusHandlerTestReconciler.java
@@ -1,6 +1,5 @@
 package io.javaoperatorsdk.operator.sample.errorstatushandler;
 
-import java.util.Optional;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.slf4j.Logger;
@@ -47,11 +46,13 @@ public int getNumberOfExecutions() {
   }
 
   @Override
-  public Optional<ErrorStatusHandlerTestCustomResource> updateErrorStatus(
-      ErrorStatusHandlerTestCustomResource resource, RetryInfo retryInfo, RuntimeException e) {
+  public ErrorStatusUpdateControl<ErrorStatusHandlerTestCustomResource> updateErrorStatus(
+      ErrorStatusHandlerTestCustomResource resource,
+      Context<ErrorStatusHandlerTestCustomResource> context, Exception e) {
     log.info("Setting status.");
     ensureStatusExists(resource);
-    resource.getStatus().getMessages().add(ERROR_STATUS_MESSAGE + retryInfo.getAttemptCount());
-    return Optional.of(resource);
+    resource.getStatus().getMessages()
+        .add(ERROR_STATUS_MESSAGE + context.getRetryInfo().orElseThrow().getAttemptCount());
+    return ErrorStatusUpdateControl.updateStatus(resource);
   }
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/standalonedependent/StandaloneDependentTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/standalonedependent/StandaloneDependentTestReconciler.java
index aa8689e471..5e3d41b32f 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/standalonedependent/StandaloneDependentTestReconciler.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/standalonedependent/StandaloneDependentTestReconciler.java
@@ -65,14 +65,15 @@ public KubernetesClient getKubernetesClient() {
   }
 
   @Override
-  public Optional<StandaloneDependentTestCustomResource> updateErrorStatus(
-      StandaloneDependentTestCustomResource resource, RetryInfo retryInfo, RuntimeException e) {
+  public ErrorStatusUpdateControl<StandaloneDependentTestCustomResource> updateErrorStatus(
+      StandaloneDependentTestCustomResource resource,
+      Context<StandaloneDependentTestCustomResource> context, Exception e) {
     // this can happen when a namespace is terminated in test
     if (e instanceof KubernetesClientException) {
-      return Optional.empty();
+      return ErrorStatusUpdateControl.noStatusUpdate();
     }
     errorOccurred = true;
-    return Optional.empty();
+    return ErrorStatusUpdateControl.noStatusUpdate();
   }
 
   public boolean isErrorOccurred() {
diff --git a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java
index eb66f918ab..b52d7931da 100644
--- a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java
+++ b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java
@@ -1,17 +1,10 @@
 package io.javaoperatorsdk.operator.sample;
 
-import java.util.Optional;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.Secret;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
-import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
-import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
-import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.*;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent;
 import io.javaoperatorsdk.operator.sample.dependent.SchemaDependentResource;
 import io.javaoperatorsdk.operator.sample.dependent.SecretDependentResource;
@@ -50,15 +43,16 @@ public UpdateControl<MySQLSchema> reconcile(MySQLSchema schema, Context<MySQLSch
   }
 
   @Override
-  public Optional<MySQLSchema> updateErrorStatus(MySQLSchema schema, RetryInfo retryInfo,
-      RuntimeException e) {
+  public ErrorStatusUpdateControl<MySQLSchema> updateErrorStatus(MySQLSchema schema,
+      Context<MySQLSchema> context,
+      Exception e) {
     SchemaStatus status = new SchemaStatus();
     status.setUrl(null);
     status.setUserName(null);
     status.setSecretName(null);
     status.setStatus("ERROR: " + e.getMessage());
     schema.setStatus(status);
-    return Optional.empty();
+    return ErrorStatusUpdateControl.updateStatus(schema);
   }
 
 
diff --git a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconciler.java b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconciler.java
index ef0c2f5d17..bc7bb83894 100644
--- a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconciler.java
+++ b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconciler.java
@@ -210,9 +210,9 @@ private static String serviceName(WebPage nginx) {
   }
 
   @Override
-  public Optional<WebPage> updateErrorStatus(
-      WebPage resource, RetryInfo retryInfo, RuntimeException e) {
+  public ErrorStatusUpdateControl<WebPage> updateErrorStatus(
+      WebPage resource, Context<WebPage> context, Exception e) {
     resource.getStatus().setErrorMessage("Error: " + e.getMessage());
-    return Optional.of(resource);
+    return ErrorStatusUpdateControl.updateStatus(resource);
   }
 }
diff --git a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconcilerDependentResources.java b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconcilerDependentResources.java
index 00a0d50d3b..ad3ef36840 100644
--- a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconcilerDependentResources.java
+++ b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageReconcilerDependentResources.java
@@ -75,10 +75,10 @@ private WebPageStatus createStatus(String configMapName) {
   }
 
   @Override
-  public Optional<WebPage> updateErrorStatus(
-      WebPage resource, RetryInfo retryInfo, RuntimeException e) {
+  public ErrorStatusUpdateControl<WebPage> updateErrorStatus(
+      WebPage resource, Context<WebPage> retryInfo, Exception e) {
     resource.getStatus().setErrorMessage("Error: " + e.getMessage());
-    return Optional.of(resource);
+    return ErrorStatusUpdateControl.updateStatus(resource);
   }
 
   private void createDependentResources(KubernetesClient client) {
@@ -91,7 +91,7 @@ private void createDependentResources(KubernetesClient client) {
         new CrudKubernetesDependentResource<>() {
 
           @Override
-          protected Deployment desired(WebPage webPage, Context context) {
+          protected Deployment desired(WebPage webPage, Context<WebPage> context) {
             var deploymentName = deploymentName(webPage);
             Deployment deployment = loadYaml(Deployment.class, getClass(), "deployment.yaml");
             deployment.getMetadata().setName(deploymentName);