diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/Operator.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/Operator.java
index e962852cca..80069d9f68 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/Operator.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/Operator.java
@@ -18,6 +18,7 @@
 import io.javaoperatorsdk.operator.processing.EventDispatcher;
 import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
+import io.javaoperatorsdk.operator.processing.retry.Retry;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
@@ -36,19 +37,33 @@ public Operator(KubernetesClient k8sClient) {
     this.k8sClient = k8sClient;
   }
 
+  public <R extends CustomResource> void registerControllerForAllNamespaces(
+      ResourceController<R> controller, Retry retry) throws OperatorException {
+    registerController(controller, true, retry);
+  }
+
   public <R extends CustomResource> void registerControllerForAllNamespaces(
       ResourceController<R> controller) throws OperatorException {
-    registerController(controller, true);
+    registerController(controller, true, null);
+  }
+
+  public <R extends CustomResource> void registerController(
+      ResourceController<R> controller, Retry retry, String... targetNamespaces)
+      throws OperatorException {
+    registerController(controller, false, retry, targetNamespaces);
   }
 
   public <R extends CustomResource> void registerController(
       ResourceController<R> controller, String... targetNamespaces) throws OperatorException {
-    registerController(controller, false, targetNamespaces);
+    registerController(controller, false, null, targetNamespaces);
   }
 
   @SuppressWarnings("rawtypes")
   private <R extends CustomResource> void registerController(
-      ResourceController<R> controller, boolean watchAllNamespaces, String... targetNamespaces)
+      ResourceController<R> controller,
+      boolean watchAllNamespaces,
+      Retry retry,
+      String... targetNamespaces)
       throws OperatorException {
     Class<R> resClass = getCustomResourceClass(controller);
     CustomResourceDefinitionContext crd = getCustomResourceDefinitionForController(controller);
@@ -67,10 +82,10 @@ private <R extends CustomResource> void registerController(
     CustomResourceCache customResourceCache = new CustomResourceCache();
     DefaultEventHandler defaultEventHandler =
         new DefaultEventHandler(
-            customResourceCache, eventDispatcher, controller.getClass().getName());
+            customResourceCache, eventDispatcher, controller.getClass().getName(), retry);
     DefaultEventSourceManager eventSourceManager =
-        new DefaultEventSourceManager(defaultEventHandler);
-    defaultEventHandler.setDefaultEventSourceManager(eventSourceManager);
+        new DefaultEventSourceManager(defaultEventHandler, retry != null);
+    defaultEventHandler.setEventSourceManager(eventSourceManager);
     eventDispatcher.setEventSourceManager(eventSourceManager);
 
     customResourceClients.put(resClass, (CustomResourceOperationsImpl) client);
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/Context.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/Context.java
index 2b79a9d6ae..7d04a5db93 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/Context.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/Context.java
@@ -3,10 +3,13 @@
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.processing.event.EventList;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
+import java.util.Optional;
 
 public interface Context<T extends CustomResource> {
 
   EventSourceManager getEventSourceManager();
 
   EventList getEvents();
+
+  Optional<RetryInfo> getRetryInfo();
 }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/DefaultContext.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/DefaultContext.java
index 7e472324d6..c922d674da 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/DefaultContext.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/DefaultContext.java
@@ -3,13 +3,17 @@
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.processing.event.EventList;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
+import java.util.Optional;
 
 public class DefaultContext<T extends CustomResource> implements Context<T> {
 
+  private final RetryInfo retryInfo;
   private final EventList events;
   private final EventSourceManager eventSourceManager;
 
-  public DefaultContext(EventSourceManager eventSourceManager, EventList events) {
+  public DefaultContext(
+      EventSourceManager eventSourceManager, EventList events, RetryInfo retryInfo) {
+    this.retryInfo = retryInfo;
     this.events = events;
     this.eventSourceManager = eventSourceManager;
   }
@@ -23,4 +27,9 @@ public EventSourceManager getEventSourceManager() {
   public EventList getEvents() {
     return events;
   }
+
+  @Override
+  public Optional<RetryInfo> getRetryInfo() {
+    return Optional.ofNullable(retryInfo);
+  }
 }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/RetryInfo.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/RetryInfo.java
index 9c3b3c05cc..92149012a7 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/RetryInfo.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/api/RetryInfo.java
@@ -1,20 +1,8 @@
 package io.javaoperatorsdk.operator.api;
 
-public class RetryInfo {
+public interface RetryInfo {
 
-  private int retryNumber;
-  private boolean lastAttempt;
+  int getAttemptCount();
 
-  public RetryInfo(int retryNumber, boolean lastAttempt) {
-    this.retryNumber = retryNumber;
-    this.lastAttempt = lastAttempt;
-  }
-
-  public int getRetryNumber() {
-    return retryNumber;
-  }
-
-  public boolean isLastAttempt() {
-    return lastAttempt;
-  }
+  boolean isLastAttempt();
 }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java
index e852e546f1..1d9b21194f 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java
@@ -5,9 +5,13 @@
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.api.RetryInfo;
 import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
+import io.javaoperatorsdk.operator.processing.retry.Retry;
+import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
+import java.util.*;
 import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
@@ -30,16 +34,20 @@ public class DefaultEventHandler implements EventHandler {
   private final Set<String> underProcessing = new HashSet<>();
   private final ScheduledThreadPoolExecutor executor;
   private final EventDispatcher eventDispatcher;
-  private DefaultEventSourceManager defaultEventSourceManager;
+  private final Retry retry;
+  private final Map<String, RetryExecution> retryState = new HashMap<>();
+  private DefaultEventSourceManager eventSourceManager;
 
   private final ReentrantLock lock = new ReentrantLock();
 
   public DefaultEventHandler(
       CustomResourceCache customResourceCache,
       EventDispatcher eventDispatcher,
-      String relatedControllerName) {
+      String relatedControllerName,
+      Retry retry) {
     this.customResourceCache = customResourceCache;
     this.eventDispatcher = eventDispatcher;
+    this.retry = retry;
     eventBuffer = new EventBuffer();
     executor =
         new ScheduledThreadPoolExecutor(
@@ -52,8 +60,8 @@ public Thread newThread(Runnable runnable) {
             });
   }
 
-  public void setDefaultEventSourceManager(DefaultEventSourceManager defaultEventSourceManager) {
-    this.defaultEventSourceManager = defaultEventSourceManager;
+  public void setEventSourceManager(DefaultEventSourceManager eventSourceManager) {
+    this.eventSourceManager = eventSourceManager;
   }
 
   @Override
@@ -79,7 +87,8 @@ private void executeBufferedEvents(String customResourceUid) {
       ExecutionScope executionScope =
           new ExecutionScope(
               eventBuffer.getAndRemoveEventsForExecution(customResourceUid),
-              latestCustomResource.get());
+              latestCustomResource.get(),
+              retryInfo(customResourceUid));
       log.debug("Executing events for custom resource. Scope: {}", executionScope);
       executor.execute(new ExecutionConsumer(executionScope, eventDispatcher, this));
     } else {
@@ -93,12 +102,28 @@ private void executeBufferedEvents(String customResourceUid) {
     }
   }
 
+  private RetryInfo retryInfo(String customResourceUid) {
+    return retryState.get(customResourceUid);
+  }
+
   void eventProcessingFinished(
       ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
     try {
       lock.lock();
-      log.debug("Event processing finished. Scope: {}", executionScope);
+      log.debug(
+          "Event processing finished. Scope: {}, PostExecutionControl: {}",
+          executionScope,
+          postExecutionControl);
       unsetUnderExecution(executionScope.getCustomResourceUid());
+
+      if (retry != null && postExecutionControl.exceptionDuringExecution()) {
+        handleRetryOnException(executionScope);
+        return;
+      }
+
+      if (retry != null) {
+        markSuccessfulExecutionRegardingRetry(executionScope);
+      }
       if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
         cleanupAfterDeletedEvent(executionScope.getCustomResourceUid());
       } else {
@@ -110,6 +135,53 @@ void eventProcessingFinished(
     }
   }
 
+  /**
+   * Regarding the events there are 2 approaches we can take. Either retry always when there are new
+   * events (received meanwhile retry is in place or already in buffer) instantly or always wait
+   * according to the retry timing if there was an exception.
+   */
+  private void handleRetryOnException(ExecutionScope executionScope) {
+    RetryExecution execution = getOrInitRetryExecution(executionScope);
+    boolean newEventsExists = eventBuffer.newEventsExists(executionScope.getCustomResourceUid());
+    eventBuffer.putBackEvents(executionScope.getCustomResourceUid(), executionScope.getEvents());
+
+    if (newEventsExists) {
+      log.debug("New events exists for for resource id: {}", executionScope.getCustomResourceUid());
+      executeBufferedEvents(executionScope.getCustomResourceUid());
+      return;
+    }
+    Optional<Long> nextDelay = execution.nextDelay();
+
+    nextDelay.ifPresent(
+        delay -> {
+          log.debug(
+              "Scheduling timer event for retry with delay:{} for resource: {}",
+              delay,
+              executionScope.getCustomResourceUid());
+          eventSourceManager
+              .getRetryTimerEventSource()
+              .scheduleOnce(executionScope.getCustomResource(), delay);
+        });
+  }
+
+  private void markSuccessfulExecutionRegardingRetry(ExecutionScope executionScope) {
+    log.debug(
+        "Marking successful execution for resource: {}", executionScope.getCustomResourceUid());
+    retryState.remove(executionScope.getCustomResourceUid());
+    eventSourceManager
+        .getRetryTimerEventSource()
+        .cancelOnceSchedule(executionScope.getCustomResourceUid());
+  }
+
+  private RetryExecution getOrInitRetryExecution(ExecutionScope executionScope) {
+    RetryExecution retryExecution = retryState.get(executionScope.getCustomResourceUid());
+    if (retryExecution == null) {
+      retryExecution = retry.initExecution();
+      retryState.put(executionScope.getCustomResourceUid(), retryExecution);
+    }
+    return retryExecution;
+  }
+
   /**
    * Here we try to cache the latest resource after an update. The goal is to solve a concurrency
    * issue we've seen: If an execution is finished, where we updated a custom resource, but there
@@ -146,7 +218,7 @@ private void cacheUpdatedResourceIfChanged(
   }
 
   private void cleanupAfterDeletedEvent(String customResourceUid) {
-    defaultEventSourceManager.cleanup(customResourceUid);
+    eventSourceManager.cleanup(customResourceUid);
     eventBuffer.cleanup(customResourceUid);
     customResourceCache.cleanup(customResourceUid);
   }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventBuffer.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventBuffer.java
index 872e9df9f0..db6a82fc1b 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventBuffer.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventBuffer.java
@@ -17,6 +17,16 @@ public void addEvent(Event event) {
     crEvents.add(event);
   }
 
+  public boolean newEventsExists(String resourceId) {
+    return events.get(resourceId) != null && !events.get(resourceId).isEmpty();
+  }
+
+  public void putBackEvents(String resourceUid, List<Event> oldEvents) {
+    List<Event> crEvents =
+        events.computeIfAbsent(resourceUid, (id) -> new ArrayList<>(oldEvents.size()));
+    crEvents.addAll(0, oldEvents);
+  }
+
   public boolean containsEvents(String customResourceId) {
     return events.get(customResourceId) != null;
   }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventDispatcher.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventDispatcher.java
index baff222eb5..b3fc86228b 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventDispatcher.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventDispatcher.java
@@ -43,16 +43,16 @@ public void setEventSourceManager(EventSourceManager eventSourceManager) {
     this.eventSourceManager = eventSourceManager;
   }
 
-  public PostExecutionControl handleEvent(ExecutionScope event) {
+  public PostExecutionControl handleExecution(ExecutionScope executionScope) {
     try {
-      return handDispatch(event);
+      return handleDispatch(executionScope);
     } catch (RuntimeException e) {
-      log.error("Error during event processing {} failed.", event, e);
-      return PostExecutionControl.defaultDispatch();
+      log.error("Error during event processing {} failed.", executionScope, e);
+      return PostExecutionControl.exceptionDuringExecution(e);
     }
   }
 
-  private PostExecutionControl handDispatch(ExecutionScope executionScope) {
+  private PostExecutionControl handleDispatch(ExecutionScope executionScope) {
     CustomResource resource = executionScope.getCustomResource();
     log.debug(
         "Handling events: {} for resource {}", executionScope.getEvents(), resource.getMetadata());
@@ -72,7 +72,10 @@ private PostExecutionControl handDispatch(ExecutionScope executionScope) {
       return PostExecutionControl.defaultDispatch();
     }
     Context context =
-        new DefaultContext(eventSourceManager, new EventList(executionScope.getEvents()));
+        new DefaultContext(
+            eventSourceManager,
+            new EventList(executionScope.getEvents()),
+            executionScope.getRetryInfo());
     if (markedForDeletion(resource)) {
       return handleDelete(resource, context);
     } else {
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionConsumer.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionConsumer.java
index 45aa4b32d0..2b66ad68d4 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionConsumer.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionConsumer.java
@@ -22,7 +22,7 @@ class ExecutionConsumer implements Runnable {
 
   @Override
   public void run() {
-    PostExecutionControl postExecutionControl = eventDispatcher.handleEvent(executionScope);
+    PostExecutionControl postExecutionControl = eventDispatcher.handleExecution(executionScope);
     defaultEventHandler.eventProcessingFinished(executionScope, postExecutionControl);
   }
 }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionScope.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionScope.java
index a4b65f5f95..73603bc7b3 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionScope.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionScope.java
@@ -1,6 +1,7 @@
 package io.javaoperatorsdk.operator.processing;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.api.RetryInfo;
 import io.javaoperatorsdk.operator.processing.event.Event;
 import java.util.List;
 
@@ -10,9 +11,12 @@ public class ExecutionScope {
   // the latest custom resource from cache
   private CustomResource customResource;
 
-  public ExecutionScope(List<Event> list, CustomResource customResource) {
+  private RetryInfo retryInfo;
+
+  public ExecutionScope(List<Event> list, CustomResource customResource, RetryInfo retryInfo) {
     this.events = list;
     this.customResource = customResource;
+    this.retryInfo = retryInfo;
   }
 
   public List<Event> getEvents() {
@@ -38,4 +42,8 @@ public String toString() {
         + customResource.getMetadata().getResourceVersion()
         + '}';
   }
+
+  public RetryInfo getRetryInfo() {
+    return retryInfo;
+  }
 }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/PostExecutionControl.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/PostExecutionControl.java
index 0e60933a81..1e0c82f1b2 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/PostExecutionControl.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/PostExecutionControl.java
@@ -9,21 +9,31 @@ public final class PostExecutionControl {
 
   private final CustomResource updatedCustomResource;
 
-  private PostExecutionControl(boolean onlyFinalizerHandled, CustomResource updatedCustomResource) {
+  private final RuntimeException runtimeException;
+
+  private PostExecutionControl(
+      boolean onlyFinalizerHandled,
+      CustomResource updatedCustomResource,
+      RuntimeException runtimeException) {
     this.onlyFinalizerHandled = onlyFinalizerHandled;
     this.updatedCustomResource = updatedCustomResource;
+    this.runtimeException = runtimeException;
   }
 
   public static PostExecutionControl onlyFinalizerAdded() {
-    return new PostExecutionControl(true, null);
+    return new PostExecutionControl(true, null, null);
   }
 
   public static PostExecutionControl defaultDispatch() {
-    return new PostExecutionControl(false, null);
+    return new PostExecutionControl(false, null, null);
   }
 
   public static PostExecutionControl customResourceUpdated(CustomResource updatedCustomResource) {
-    return new PostExecutionControl(false, updatedCustomResource);
+    return new PostExecutionControl(false, updatedCustomResource, null);
+  }
+
+  public static PostExecutionControl exceptionDuringExecution(RuntimeException exception) {
+    return new PostExecutionControl(false, null, exception);
   }
 
   public boolean isOnlyFinalizerHandled() {
@@ -37,4 +47,24 @@ public Optional<CustomResource> getUpdatedCustomResource() {
   public boolean customResourceUpdatedDuringExecution() {
     return updatedCustomResource != null;
   }
+
+  public boolean exceptionDuringExecution() {
+    return runtimeException != null;
+  }
+
+  public Optional<RuntimeException> getRuntimeException() {
+    return Optional.ofNullable(runtimeException);
+  }
+
+  @Override
+  public String toString() {
+    return "PostExecutionControl{"
+        + "onlyFinalizerHandled="
+        + onlyFinalizerHandled
+        + ", updatedCustomResource="
+        + updatedCustomResource
+        + ", runtimeException="
+        + runtimeException
+        + '}';
+  }
 }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManager.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManager.java
index be8e76f643..24e0fe2d4f 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManager.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManager.java
@@ -2,6 +2,7 @@
 
 import io.javaoperatorsdk.operator.processing.DefaultEventHandler;
 import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.internal.TimerEventSource;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
@@ -12,15 +13,21 @@
 
 public class DefaultEventSourceManager implements EventSourceManager {
 
+  public static final String RETRY_TIMER_EVENT_SOURCE_NAME = "retry-timer-event-source";
   private static final Logger log = LoggerFactory.getLogger(DefaultEventSourceManager.class);
 
   private final ReentrantLock lock = new ReentrantLock();
   private Map<String, EventSource> eventSources = new ConcurrentHashMap<>();
   private CustomResourceEventSource customResourceEventSource;
   private DefaultEventHandler defaultEventHandler;
+  private TimerEventSource retryTimerEventSource;
 
-  public DefaultEventSourceManager(DefaultEventHandler defaultEventHandler) {
+  public DefaultEventSourceManager(DefaultEventHandler defaultEventHandler, boolean supportRetry) {
     this.defaultEventHandler = defaultEventHandler;
+    if (supportRetry) {
+      this.retryTimerEventSource = new TimerEventSource();
+      registerEventSource(RETRY_TIMER_EVENT_SOURCE_NAME, retryTimerEventSource);
+    }
   }
 
   public void registerCustomResourceEventSource(
@@ -66,6 +73,10 @@ public Optional<EventSource> deRegisterCustomResourceFromEventSource(
     }
   }
 
+  public TimerEventSource getRetryTimerEventSource() {
+    return retryTimerEventSource;
+  }
+
   @Override
   public Map<String, EventSource> getRegisteredEventSources() {
     return Collections.unmodifiableMap(eventSources);
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetry.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetry.java
index ca8d47aa84..ba8396ddeb 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetry.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetry.java
@@ -16,7 +16,7 @@ public static GenericRetry defaultLimitedExponentialRetry() {
   }
 
   public static GenericRetry noRetry() {
-    return new GenericRetry().setMaxAttempts(1);
+    return new GenericRetry().setMaxAttempts(0);
   }
 
   public static GenericRetry every10second10TimesRetry() {
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecution.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecution.java
index b15144f536..a2c7a9a609 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecution.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecution.java
@@ -14,15 +14,7 @@ public GenericRetryExecution(GenericRetry genericRetry) {
     this.currentInterval = genericRetry.getInitialInterval();
   }
 
-  /**
-   * Note that first attempt is always 0. Since this implementation is tailored for event
-   * scheduling.
-   */
   public Optional<Long> nextDelay() {
-    if (lastAttemptIndex == 0) {
-      lastAttemptIndex++;
-      return Optional.of(0L);
-    }
     if (genericRetry.getMaxAttempts() > -1 && lastAttemptIndex >= genericRetry.getMaxAttempts()) {
       return Optional.empty();
     }
@@ -37,7 +29,12 @@ public Optional<Long> nextDelay() {
   }
 
   @Override
-  public boolean isLastExecution() {
+  public boolean isLastAttempt() {
     return genericRetry.getMaxAttempts() > -1 && lastAttemptIndex >= genericRetry.getMaxAttempts();
   }
+
+  @Override
+  public int getAttemptCount() {
+    return lastAttemptIndex;
+  }
 }
diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/RetryExecution.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/RetryExecution.java
index b14a5966fb..e5fb22ae62 100644
--- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/RetryExecution.java
+++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/RetryExecution.java
@@ -1,8 +1,9 @@
 package io.javaoperatorsdk.operator.processing.retry;
 
+import io.javaoperatorsdk.operator.api.RetryInfo;
 import java.util.Optional;
 
-public interface RetryExecution {
+public interface RetryExecution extends RetryInfo {
 
   /**
    * Calculates the delay for the next execution. This method should return 0, when called first
@@ -11,10 +12,4 @@ public interface RetryExecution {
    * @return
    */
   Optional<Long> nextDelay();
-
-  /**
-   * @return true, if the last returned delay is, the last returned values, thus there will be no
-   *     further retry
-   */
-  boolean isLastExecution();
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/EventDispatcherTest.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/EventDispatcherTest.java
index 74e1725d37..4e37bc573d 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/EventDispatcherTest.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/EventDispatcherTest.java
@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.argThat;
@@ -12,9 +13,7 @@
 
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.client.Watcher;
-import io.javaoperatorsdk.operator.api.DeleteControl;
-import io.javaoperatorsdk.operator.api.ResourceController;
-import io.javaoperatorsdk.operator.api.UpdateControl;
+import io.javaoperatorsdk.operator.api.*;
 import io.javaoperatorsdk.operator.processing.EventDispatcher;
 import io.javaoperatorsdk.operator.processing.ExecutionScope;
 import io.javaoperatorsdk.operator.processing.event.Event;
@@ -25,6 +24,7 @@
 import java.util.List;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.ArgumentMatchers;
 
 class EventDispatcherTest {
@@ -54,7 +54,7 @@ void setup() {
 
   @Test
   void callCreateOrUpdateOnNewResource() {
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.ADDED, testCustomResource));
     verify(controller, times(1))
         .createOrUpdateResource(ArgumentMatchers.eq(testCustomResource), any());
@@ -65,7 +65,7 @@ void updatesOnlyStatusSubResource() {
     when(controller.createOrUpdateResource(eq(testCustomResource), any()))
         .thenReturn(UpdateControl.updateStatusSubResource(testCustomResource));
 
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.ADDED, testCustomResource));
 
     verify(customResourceFacade, times(1)).updateStatus(testCustomResource);
@@ -74,7 +74,7 @@ void updatesOnlyStatusSubResource() {
 
   @Test
   void callCreateOrUpdateOnModifiedResource() {
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
     verify(controller, times(1))
         .createOrUpdateResource(ArgumentMatchers.eq(testCustomResource), any());
@@ -82,7 +82,7 @@ void callCreateOrUpdateOnModifiedResource() {
 
   @Test
   void adsDefaultFinalizerOnCreateIfNotThere() {
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
     verify(controller, times(1))
         .createOrUpdateResource(
@@ -97,7 +97,7 @@ void callsDeleteIfObjectHasFinalizerAndMarkedForDelete() {
     testCustomResource.getMetadata().setDeletionTimestamp("2019-8-10");
     testCustomResource.getMetadata().getFinalizers().add(DEFAULT_FINALIZER);
 
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
 
     verify(controller, times(1)).deleteResource(eq(testCustomResource), any());
@@ -108,7 +108,7 @@ void callsDeleteIfObjectHasFinalizerAndMarkedForDelete() {
   void callDeleteOnControllerIfMarkedForDeletionButThereIsNoDefaultFinalizer() {
     markForDeletion(testCustomResource);
 
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
 
     verify(controller).deleteResource(eq(testCustomResource), any());
@@ -118,7 +118,7 @@ void callDeleteOnControllerIfMarkedForDeletionButThereIsNoDefaultFinalizer() {
   void removesDefaultFinalizerOnDelete() {
     markForDeletion(testCustomResource);
 
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
 
     assertEquals(0, testCustomResource.getMetadata().getFinalizers().size());
@@ -131,7 +131,7 @@ void doesNotRemovesTheFinalizerIfTheDeleteNotMethodInstructsIt() {
         .thenReturn(DeleteControl.NO_FINALIZER_REMOVAL);
     markForDeletion(testCustomResource);
 
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
 
     assertEquals(1, testCustomResource.getMetadata().getFinalizers().size());
@@ -143,7 +143,7 @@ void doesNotUpdateTheResourceIfNoUpdateUpdateControl() {
     when(controller.createOrUpdateResource(eq(testCustomResource), any()))
         .thenReturn(UpdateControl.noUpdate());
 
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
     verify(customResourceFacade, never()).replaceWithLock(any());
     verify(customResourceFacade, never()).updateStatus(testCustomResource);
@@ -155,7 +155,7 @@ void addsFinalizerIfNotMarkedForDeletionAndEmptyCustomResourceReturned() {
     when(controller.createOrUpdateResource(eq(testCustomResource), any()))
         .thenReturn(UpdateControl.noUpdate());
 
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
 
     assertEquals(1, testCustomResource.getMetadata().getFinalizers().size());
@@ -167,7 +167,7 @@ void doesNotCallDeleteIfMarkedForDeletionButNotOurFinalizer() {
     removeFinalizers(testCustomResource);
     markForDeletion(testCustomResource);
 
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
 
     verify(customResourceFacade, never()).replaceWithLock(any());
@@ -176,14 +176,41 @@ void doesNotCallDeleteIfMarkedForDeletionButNotOurFinalizer() {
 
   @Test
   void executeControllerRegardlessGenerationInNonGenerationAwareMode() {
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
-    eventDispatcher.handleEvent(
+    eventDispatcher.handleExecution(
         executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
 
     verify(controller, times(2)).createOrUpdateResource(eq(testCustomResource), any());
   }
 
+  @Test
+  void propagatesRetryInfoToContext() {
+    eventDispatcher.handleExecution(
+        new ExecutionScope(
+            Arrays.asList(),
+            testCustomResource,
+            new RetryInfo() {
+              @Override
+              public int getAttemptCount() {
+                return 2;
+              }
+
+              @Override
+              public boolean isLastAttempt() {
+                return true;
+              }
+            }));
+
+    ArgumentCaptor<Context<CustomResource>> contextArgumentCaptor =
+        ArgumentCaptor.forClass(Context.class);
+    verify(controller, times(1))
+        .createOrUpdateResource(eq(testCustomResource), contextArgumentCaptor.capture());
+    Context<CustomResource> context = contextArgumentCaptor.getValue();
+    assertThat(context.getRetryInfo().get().getAttemptCount()).isEqualTo(2);
+    assertThat(context.getRetryInfo().get().isLastAttempt()).isEqualTo(true);
+  }
+
   private void markForDeletion(CustomResource customResource) {
     customResource.getMetadata().setDeletionTimestamp("2019-8-10");
   }
@@ -198,6 +225,6 @@ public ExecutionScope executionScopeWithCREvent(
     List<Event> eventList = new ArrayList<>(1 + otherEvents.length);
     eventList.add(event);
     eventList.addAll(Arrays.asList(otherEvents));
-    return new ExecutionScope(eventList, resource);
+    return new ExecutionScope(eventList, resource, null);
   }
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/IntegrationTestSupport.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/IntegrationTestSupport.java
index bc6ef2db01..78512f6fb0 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/IntegrationTestSupport.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/IntegrationTestSupport.java
@@ -16,6 +16,7 @@
 import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
 import io.fabric8.kubernetes.client.utils.Serialization;
 import io.javaoperatorsdk.operator.api.ResourceController;
+import io.javaoperatorsdk.operator.processing.retry.Retry;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResourceSpec;
 import java.io.IOException;
@@ -41,6 +42,11 @@ public class IntegrationTestSupport {
 
   public void initialize(
       KubernetesClient k8sClient, ResourceController controller, String crdPath) {
+    initialize(k8sClient, controller, crdPath, null);
+  }
+
+  public void initialize(
+      KubernetesClient k8sClient, ResourceController controller, String crdPath, Retry retry) {
     log.info("Initializing integration test in namespace {}", TEST_NAMESPACE);
     this.k8sClient = k8sClient;
     CustomResourceDefinition crd = loadCRDAndApplyToCluster(crdPath);
@@ -62,7 +68,7 @@ public void initialize(
                   .build());
     }
     operator = new Operator(k8sClient);
-    operator.registerController(controller, TEST_NAMESPACE);
+    operator.registerController(controller, retry, TEST_NAMESPACE);
     log.info("Operator is running with {}", controller.getClass().getCanonicalName());
   }
 
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/RetryIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/RetryIT.java
new file mode 100644
index 0000000000..3b1a8bc82c
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/RetryIT.java
@@ -0,0 +1,72 @@
+package io.javaoperatorsdk.operator;
+
+import static io.javaoperatorsdk.operator.IntegrationTestSupport.TEST_NAMESPACE;
+import static io.javaoperatorsdk.operator.sample.event.EventSourceTestCustomResourceController.*;
+import static io.javaoperatorsdk.operator.sample.retry.RetryTestCustomResourceStatus.State.SUCCESS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.client.DefaultKubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
+import io.javaoperatorsdk.operator.processing.retry.Retry;
+import io.javaoperatorsdk.operator.sample.retry.RetryTestCustomResource;
+import io.javaoperatorsdk.operator.sample.retry.RetryTestCustomResourceController;
+import io.javaoperatorsdk.operator.sample.retry.RetryTestCustomResourceSpec;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class RetryIT {
+
+  public static final int RETRY_INTERVAL = 150;
+  private IntegrationTestSupport integrationTestSupport = new IntegrationTestSupport();
+
+  @BeforeEach
+  public void initAndCleanup() {
+    Retry retry =
+        new GenericRetry().setInitialInterval(RETRY_INTERVAL).withLinearRetry().setMaxAttempts(5);
+    KubernetesClient k8sClient = new DefaultKubernetesClient();
+    integrationTestSupport.initialize(
+        k8sClient, new RetryTestCustomResourceController(), "retry-test-crd.yaml", retry);
+    integrationTestSupport.cleanup();
+  }
+
+  @Test
+  public void retryFailedExecution() {
+    integrationTestSupport.teardownIfSuccess(
+        () -> {
+          RetryTestCustomResource resource = createTestCustomResource("1");
+          integrationTestSupport.getCrOperations().inNamespace(TEST_NAMESPACE).create(resource);
+
+          Thread.sleep(
+              RETRY_INTERVAL * (RetryTestCustomResourceController.NUMBER_FAILED_EXECUTIONS + 2));
+
+          assertThat(integrationTestSupport.numberOfControllerExecutions())
+              .isGreaterThanOrEqualTo(
+                  RetryTestCustomResourceController.NUMBER_FAILED_EXECUTIONS + 1);
+
+          RetryTestCustomResource finalResource =
+              (RetryTestCustomResource)
+                  integrationTestSupport
+                      .getCrOperations()
+                      .inNamespace(TEST_NAMESPACE)
+                      .withName(resource.getMetadata().getName())
+                      .get();
+          assertThat(finalResource.getStatus().getState()).isEqualTo(SUCCESS);
+        });
+  }
+
+  public RetryTestCustomResource createTestCustomResource(String id) {
+    RetryTestCustomResource resource = new RetryTestCustomResource();
+    resource.setMetadata(
+        new ObjectMetaBuilder()
+            .withName("retrysource-" + id)
+            .withNamespace(TEST_NAMESPACE)
+            .withFinalizers(RetryTestCustomResourceController.FINALIZER_NAME)
+            .build());
+    resource.setKind("retrysample");
+    resource.setSpec(new RetryTestCustomResourceSpec());
+    resource.getSpec().setValue(id);
+    return resource;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/DefaultEventHandlerTest.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/DefaultEventHandlerTest.java
index 5e194fad01..b998518d53 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/DefaultEventHandlerTest.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/DefaultEventHandlerTest.java
@@ -2,8 +2,10 @@
 
 import static io.javaoperatorsdk.operator.TestUtils.testCustomResource;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -14,35 +16,54 @@
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEvent;
 import io.javaoperatorsdk.operator.processing.event.internal.TimerEvent;
+import io.javaoperatorsdk.operator.processing.event.internal.TimerEventSource;
+import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
+import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.stubbing.Answer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 class DefaultEventHandlerTest {
 
+  private static final Logger log = LoggerFactory.getLogger(DefaultEventHandlerTest.class);
+
   public static final int FAKE_CONTROLLER_EXECUTION_DURATION = 250;
   public static final int SEPARATE_EXECUTION_TIMEOUT = 450;
   private EventDispatcher eventDispatcherMock = mock(EventDispatcher.class);
   private CustomResourceCache customResourceCache = new CustomResourceCache();
-  private DefaultEventHandler defaultEventHandler =
-      new DefaultEventHandler(customResourceCache, eventDispatcherMock, "Test");
   private DefaultEventSourceManager defaultEventSourceManagerMock =
       mock(DefaultEventSourceManager.class);
+  private TimerEventSource retryTimerEventSourceMock = mock(TimerEventSource.class);
+
+  private DefaultEventHandler defaultEventHandler =
+      new DefaultEventHandler(customResourceCache, eventDispatcherMock, "Test", null);
+
+  private DefaultEventHandler defaultEventHandlerWithRetry =
+      new DefaultEventHandler(
+          customResourceCache,
+          eventDispatcherMock,
+          "Test",
+          GenericRetry.defaultLimitedExponentialRetry());
 
   @BeforeEach
   public void setup() {
-    defaultEventHandler.setDefaultEventSourceManager(defaultEventSourceManagerMock);
+    when(defaultEventSourceManagerMock.getRetryTimerEventSource())
+        .thenReturn(retryTimerEventSourceMock);
+    defaultEventHandler.setEventSourceManager(defaultEventSourceManagerMock);
+    defaultEventHandlerWithRetry.setEventSourceManager(defaultEventSourceManagerMock);
   }
 
   @Test
   public void dispatchesEventsIfNoExecutionInProgress() {
     defaultEventHandler.handleEvent(prepareCREvent());
 
-    verify(eventDispatcherMock, timeout(50).times(1)).handleEvent(any());
+    verify(eventDispatcherMock, timeout(50).times(1)).handleExecution(any());
   }
 
   @Test
@@ -52,7 +73,7 @@ public void skipProcessingIfLatestCustomResourceNotInCache() {
 
     defaultEventHandler.handleEvent(event);
 
-    verify(eventDispatcherMock, timeout(50).times(0)).handleEvent(any());
+    verify(eventDispatcherMock, timeout(50).times(0)).handleExecution(any());
   }
 
   @Test
@@ -61,7 +82,8 @@ public void ifExecutionInProgressWaitsUntilItsFinished() throws InterruptedExcep
 
     defaultEventHandler.handleEvent(nonCREvent(resourceUid));
 
-    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1)).handleEvent(any());
+    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
+        .handleExecution(any());
   }
 
   @Test
@@ -73,7 +95,7 @@ public void buffersAllIncomingEventsWhileControllerInExecution() {
 
     ArgumentCaptor<ExecutionScope> captor = ArgumentCaptor.forClass(ExecutionScope.class);
     verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(2))
-        .handleEvent(captor.capture());
+        .handleExecution(captor.capture());
     List<Event> events = captor.getAllValues().get(1).getEvents();
     assertThat(events).hasSize(2);
     assertThat(events.get(0)).isInstanceOf(TimerEvent.class);
@@ -89,13 +111,99 @@ public void cleanUpAfterDeleteEvent() {
     String uid = customResource.getMetadata().getUid();
 
     defaultEventHandler.handleEvent(event);
-    // todo awaitility?
+
     waitMinimalTime();
 
     verify(defaultEventSourceManagerMock, times(1)).cleanup(uid);
     assertThat(customResourceCache.getLatestResource(uid)).isNotPresent();
   }
 
+  @Test
+  public void schedulesAnEventRetryOnException() {
+    Event event = prepareCREvent();
+    TestCustomResource customResource = testCustomResource();
+
+    ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource, null);
+    PostExecutionControl postExecutionControl =
+        PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
+
+    defaultEventHandlerWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
+
+    verify(retryTimerEventSourceMock, times(1))
+        .scheduleOnce(eq(customResource), eq(GenericRetry.DEFAULT_INITIAL_INTERVAL));
+  }
+
+  @Test
+  public void executesTheControllerInstantlyAfterErrorIfEventsBuffered() {
+    Event event = prepareCREvent();
+    TestCustomResource customResource = testCustomResource();
+    customResource.getMetadata().setUid(event.getRelatedCustomResourceUid());
+    ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource, null);
+    PostExecutionControl postExecutionControl =
+        PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
+
+    // start processing an event
+    defaultEventHandlerWithRetry.handleEvent(event);
+    // buffer an another event
+    defaultEventHandlerWithRetry.handleEvent(event);
+    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
+        .handleExecution(any());
+
+    defaultEventHandlerWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
+
+    ArgumentCaptor<ExecutionScope> executionScopeArgumentCaptor =
+        ArgumentCaptor.forClass(ExecutionScope.class);
+    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(2))
+        .handleExecution(executionScopeArgumentCaptor.capture());
+    List<ExecutionScope> allValues = executionScopeArgumentCaptor.getAllValues();
+    assertThat(allValues).hasSize(2);
+    assertThat(allValues.get(1).getEvents()).hasSize(2);
+    verify(retryTimerEventSourceMock, never())
+        .scheduleOnce(eq(customResource), eq(GenericRetry.DEFAULT_INITIAL_INTERVAL));
+  }
+
+  @Test
+  public void successfulExecutionResetsTheRetry() {
+    log.info("Starting successfulExecutionResetsTheRetry");
+
+    Event event = prepareCREvent();
+    TestCustomResource customResource = testCustomResource();
+    customResource.getMetadata().setUid(event.getRelatedCustomResourceUid());
+    ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource, null);
+    PostExecutionControl postExecutionControlWithException =
+        PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
+    PostExecutionControl defaultDispatchControl = PostExecutionControl.defaultDispatch();
+
+    when(eventDispatcherMock.handleExecution(any()))
+        .thenReturn(postExecutionControlWithException)
+        .thenReturn(defaultDispatchControl);
+
+    ArgumentCaptor<ExecutionScope> executionScopeArgumentCaptor =
+        ArgumentCaptor.forClass(ExecutionScope.class);
+
+    defaultEventHandlerWithRetry.handleEvent(event);
+
+    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
+        .handleExecution(any());
+    defaultEventHandlerWithRetry.handleEvent(event);
+
+    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(2))
+        .handleExecution(any());
+    defaultEventHandlerWithRetry.handleEvent(event);
+
+    verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(3))
+        .handleExecution(executionScopeArgumentCaptor.capture());
+    log.info("Finished successfulExecutionResetsTheRetry");
+
+    List<ExecutionScope> executionScopes = executionScopeArgumentCaptor.getAllValues();
+
+    assertThat(executionScopes).hasSize(3);
+    assertThat(executionScopes.get(0).getRetryInfo()).isNull();
+    assertThat(executionScopes.get(2).getRetryInfo()).isNull();
+    assertThat(executionScopes.get(1).getRetryInfo().getAttemptCount()).isEqualTo(1);
+    assertThat(executionScopes.get(1).getRetryInfo().isLastAttempt()).isEqualTo(false);
+  }
+
   private void waitMinimalTime() {
     try {
       Thread.sleep(50);
@@ -105,7 +213,7 @@ private void waitMinimalTime() {
   }
 
   private String eventAlreadyUnderProcessing() {
-    when(eventDispatcherMock.handleEvent(any()))
+    when(eventDispatcherMock.handleExecution(any()))
         .then(
             (Answer<PostExecutionControl>)
                 invocationOnMock -> {
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManagerTest.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManagerTest.java
index a42c824e8c..39ae1c5517 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManagerTest.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManagerTest.java
@@ -21,7 +21,7 @@ class DefaultEventSourceManagerTest {
 
   private DefaultEventHandler defaultEventHandlerMock = mock(DefaultEventHandler.class);
   private DefaultEventSourceManager defaultEventSourceManager =
-      new DefaultEventSourceManager(defaultEventHandlerMock);
+      new DefaultEventSourceManager(defaultEventHandlerMock, false);
 
   @Test
   public void registersEventSource() {
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecutionTest.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecutionTest.java
index ab7aa98fd3..df934fdec5 100644
--- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecutionTest.java
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecutionTest.java
@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.processing.retry;
 
+import static io.javaoperatorsdk.operator.processing.retry.GenericRetry.DEFAULT_INITIAL_INTERVAL;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Optional;
@@ -8,27 +9,26 @@
 public class GenericRetryExecutionTest {
 
   @Test
-  public void forFirstBackOffAlwaysReturnsZero() {
-    assertThat(getDefaultRetryExecution().nextDelay().get()).isEqualTo(0);
+  public void forFirstBackOffAlwaysReturnsInitialInterval() {
+    assertThat(getDefaultRetryExecution().nextDelay().get()).isEqualTo(DEFAULT_INITIAL_INTERVAL);
   }
 
   @Test
   public void delayIsMultipliedEveryNextDelayCall() {
     RetryExecution retryExecution = getDefaultRetryExecution();
 
-    Optional<Long> res = callNextDelayNTimes(retryExecution, 2);
-    assertThat(res.get()).isEqualTo(GenericRetry.DEFAULT_INITIAL_INTERVAL);
+    Optional<Long> res = callNextDelayNTimes(retryExecution, 1);
+    assertThat(res.get()).isEqualTo(DEFAULT_INITIAL_INTERVAL);
 
     res = retryExecution.nextDelay();
     assertThat(res.get())
-        .isEqualTo(
-            (long) (GenericRetry.DEFAULT_INITIAL_INTERVAL * GenericRetry.DEFAULT_MULTIPLIER));
+        .isEqualTo((long) (DEFAULT_INITIAL_INTERVAL * GenericRetry.DEFAULT_MULTIPLIER));
 
     res = retryExecution.nextDelay();
     assertThat(res.get())
         .isEqualTo(
             (long)
-                (GenericRetry.DEFAULT_INITIAL_INTERVAL
+                (DEFAULT_INITIAL_INTERVAL
                     * GenericRetry.DEFAULT_MULTIPLIER
                     * GenericRetry.DEFAULT_MULTIPLIER));
   }
@@ -37,7 +37,7 @@ public void delayIsMultipliedEveryNextDelayCall() {
   public void noNextDelayIfMaxAttemptLimitReached() {
     RetryExecution retryExecution =
         GenericRetry.defaultLimitedExponentialRetry().setMaxAttempts(3).initExecution();
-    Optional<Long> res = callNextDelayNTimes(retryExecution, 3);
+    Optional<Long> res = callNextDelayNTimes(retryExecution, 2);
     assertThat(res).isNotEmpty();
 
     res = retryExecution.nextDelay();
@@ -61,18 +61,26 @@ public void canLimitMaxIntervalLength() {
   @Test
   public void supportsNoRetry() {
     RetryExecution retryExecution = GenericRetry.noRetry().initExecution();
-    assertThat(retryExecution.nextDelay().get()).isZero();
     assertThat(retryExecution.nextDelay()).isEmpty();
   }
 
   @Test
   public void supportsIsLastExecution() {
     GenericRetryExecution execution = new GenericRetry().setMaxAttempts(2).initExecution();
-    assertThat(execution.isLastExecution()).isFalse();
+    assertThat(execution.isLastAttempt()).isFalse();
 
     execution.nextDelay();
     execution.nextDelay();
-    assertThat(execution.isLastExecution()).isTrue();
+    assertThat(execution.isLastAttempt()).isTrue();
+  }
+
+  @Test
+  public void returnAttemptIndex() {
+    RetryExecution retryExecution = GenericRetry.defaultLimitedExponentialRetry().initExecution();
+
+    assertThat(retryExecution.getAttemptCount()).isEqualTo(0);
+    retryExecution.nextDelay();
+    assertThat(retryExecution.getAttemptCount()).isEqualTo(1);
   }
 
   private RetryExecution getDefaultRetryExecution() {
@@ -80,7 +88,7 @@ private RetryExecution getDefaultRetryExecution() {
   }
 
   public Optional<Long> callNextDelayNTimes(RetryExecution retryExecution, int n) {
-    for (int i = 0; i < n - 1; i++) {
+    for (int i = 0; i < n; i++) {
       retryExecution.nextDelay();
     }
     return retryExecution.nextDelay();
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResource.java
new file mode 100644
index 0000000000..a351a78a33
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResource.java
@@ -0,0 +1,38 @@
+package io.javaoperatorsdk.operator.sample.retry;
+
+import io.fabric8.kubernetes.client.CustomResource;
+
+public class RetryTestCustomResource extends CustomResource {
+
+  private RetryTestCustomResourceSpec spec;
+
+  private RetryTestCustomResourceStatus status;
+
+  public RetryTestCustomResourceSpec getSpec() {
+    return spec;
+  }
+
+  public void setSpec(RetryTestCustomResourceSpec spec) {
+    this.spec = spec;
+  }
+
+  public RetryTestCustomResourceStatus getStatus() {
+    return status;
+  }
+
+  public void setStatus(RetryTestCustomResourceStatus status) {
+    this.status = status;
+  }
+
+  @Override
+  public String toString() {
+    return "TestCustomResource{"
+        + "spec="
+        + spec
+        + ", status="
+        + status
+        + ", extendedFrom="
+        + super.toString()
+        + '}';
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceController.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceController.java
new file mode 100644
index 0000000000..6554446922
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceController.java
@@ -0,0 +1,61 @@
+package io.javaoperatorsdk.operator.sample.retry;
+
+import io.javaoperatorsdk.operator.TestExecutionInfoProvider;
+import io.javaoperatorsdk.operator.api.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Controller(crdName = RetryTestCustomResourceController.CRD_NAME)
+public class RetryTestCustomResourceController
+    implements ResourceController<RetryTestCustomResource>, TestExecutionInfoProvider {
+
+  public static final int NUMBER_FAILED_EXECUTIONS = 2;
+
+  public static final String CRD_NAME = "retrysamples.sample.javaoperatorsdk";
+  public static final String FINALIZER_NAME = CRD_NAME + "/finalizer";
+  private static final Logger log =
+      LoggerFactory.getLogger(RetryTestCustomResourceController.class);
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  @Override
+  public DeleteControl deleteResource(
+      RetryTestCustomResource resource, Context<RetryTestCustomResource> context) {
+    return DeleteControl.DEFAULT_DELETE;
+  }
+
+  @Override
+  public UpdateControl<RetryTestCustomResource> createOrUpdateResource(
+      RetryTestCustomResource resource, Context<RetryTestCustomResource> context) {
+    numberOfExecutions.addAndGet(1);
+
+    if (!resource.getMetadata().getFinalizers().contains(FINALIZER_NAME)) {
+      throw new IllegalStateException("Finalizer is not present.");
+    }
+    log.info("Value: " + resource.getSpec().getValue());
+
+    if (numberOfExecutions.get() < NUMBER_FAILED_EXECUTIONS + 1) {
+      throw new RuntimeException("Testing Retry");
+    }
+    if (context.getRetryInfo().isEmpty() || context.getRetryInfo().get().isLastAttempt() == true) {
+      throw new IllegalStateException("Not expected retry info: " + context.getRetryInfo());
+    }
+
+    ensureStatusExists(resource);
+    resource.getStatus().setState(RetryTestCustomResourceStatus.State.SUCCESS);
+
+    return UpdateControl.updateStatusSubResource(resource);
+  }
+
+  private void ensureStatusExists(RetryTestCustomResource resource) {
+    RetryTestCustomResourceStatus status = resource.getStatus();
+    if (status == null) {
+      status = new RetryTestCustomResourceStatus();
+      resource.setStatus(status);
+    }
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceSpec.java
new file mode 100644
index 0000000000..ec34d11df8
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceSpec.java
@@ -0,0 +1,15 @@
+package io.javaoperatorsdk.operator.sample.retry;
+
+public class RetryTestCustomResourceSpec {
+
+  private String value;
+
+  public String getValue() {
+    return value;
+  }
+
+  public RetryTestCustomResourceSpec setValue(String value) {
+    this.value = value;
+    return this;
+  }
+}
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceStatus.java
new file mode 100644
index 0000000000..5514c04465
--- /dev/null
+++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceStatus.java
@@ -0,0 +1,20 @@
+package io.javaoperatorsdk.operator.sample.retry;
+
+public class RetryTestCustomResourceStatus {
+
+  private State state;
+
+  public State getState() {
+    return state;
+  }
+
+  public RetryTestCustomResourceStatus setState(State state) {
+    this.state = state;
+    return this;
+  }
+
+  public enum State {
+    SUCCESS,
+    ERROR
+  }
+}
diff --git a/operator-framework/src/test/resources/io/javaoperatorsdk/operator/retry-test-crd.yaml b/operator-framework/src/test/resources/io/javaoperatorsdk/operator/retry-test-crd.yaml
new file mode 100644
index 0000000000..e61f3202ca
--- /dev/null
+++ b/operator-framework/src/test/resources/io/javaoperatorsdk/operator/retry-test-crd.yaml
@@ -0,0 +1,16 @@
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: retrysamples.sample.javaoperatorsdk
+spec:
+  group: sample.javaoperatorsdk
+  version: v1
+  subresources:
+    status: {}
+  scope: Namespaced
+  names:
+    plural: retrysamples
+    singular: retrysample
+    kind: retrysample
+    shortNames:
+      - rs