Skip to content
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Commit 62ea9f6

Browse files
s-sorooshs-soroosh
authoredDec 14, 2020
Merge pull request #252 from java-operator-sdk/retry-experiment
Retry Support
2 parents f2660be + f30f35e commit 62ea9f6

25 files changed

+611
-99
lines changed
 

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/Operator.java

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import io.javaoperatorsdk.operator.processing.EventDispatcher;
1919
import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
2020
import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
21+
import io.javaoperatorsdk.operator.processing.retry.Retry;
2122
import java.util.Arrays;
2223
import java.util.HashMap;
2324
import java.util.Map;
@@ -36,19 +37,33 @@ public Operator(KubernetesClient k8sClient) {
3637
this.k8sClient = k8sClient;
3738
}
3839

40+
public <R extends CustomResource> void registerControllerForAllNamespaces(
41+
ResourceController<R> controller, Retry retry) throws OperatorException {
42+
registerController(controller, true, retry);
43+
}
44+
3945
public <R extends CustomResource> void registerControllerForAllNamespaces(
4046
ResourceController<R> controller) throws OperatorException {
41-
registerController(controller, true);
47+
registerController(controller, true, null);
48+
}
49+
50+
public <R extends CustomResource> void registerController(
51+
ResourceController<R> controller, Retry retry, String... targetNamespaces)
52+
throws OperatorException {
53+
registerController(controller, false, retry, targetNamespaces);
4254
}
4355

4456
public <R extends CustomResource> void registerController(
4557
ResourceController<R> controller, String... targetNamespaces) throws OperatorException {
46-
registerController(controller, false, targetNamespaces);
58+
registerController(controller, false, null, targetNamespaces);
4759
}
4860

4961
@SuppressWarnings("rawtypes")
5062
private <R extends CustomResource> void registerController(
51-
ResourceController<R> controller, boolean watchAllNamespaces, String... targetNamespaces)
63+
ResourceController<R> controller,
64+
boolean watchAllNamespaces,
65+
Retry retry,
66+
String... targetNamespaces)
5267
throws OperatorException {
5368
Class<R> resClass = getCustomResourceClass(controller);
5469
CustomResourceDefinitionContext crd = getCustomResourceDefinitionForController(controller);
@@ -67,10 +82,10 @@ private <R extends CustomResource> void registerController(
6782
CustomResourceCache customResourceCache = new CustomResourceCache();
6883
DefaultEventHandler defaultEventHandler =
6984
new DefaultEventHandler(
70-
customResourceCache, eventDispatcher, controller.getClass().getName());
85+
customResourceCache, eventDispatcher, controller.getClass().getName(), retry);
7186
DefaultEventSourceManager eventSourceManager =
72-
new DefaultEventSourceManager(defaultEventHandler);
73-
defaultEventHandler.setDefaultEventSourceManager(eventSourceManager);
87+
new DefaultEventSourceManager(defaultEventHandler, retry != null);
88+
defaultEventHandler.setEventSourceManager(eventSourceManager);
7489
eventDispatcher.setEventSourceManager(eventSourceManager);
7590

7691
customResourceClients.put(resClass, (CustomResourceOperationsImpl) client);

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/api/Context.java

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,13 @@
33
import io.fabric8.kubernetes.client.CustomResource;
44
import io.javaoperatorsdk.operator.processing.event.EventList;
55
import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
6+
import java.util.Optional;
67

78
public interface Context<T extends CustomResource> {
89

910
EventSourceManager getEventSourceManager();
1011

1112
EventList getEvents();
13+
14+
Optional<RetryInfo> getRetryInfo();
1215
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/api/DefaultContext.java

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,17 @@
33
import io.fabric8.kubernetes.client.CustomResource;
44
import io.javaoperatorsdk.operator.processing.event.EventList;
55
import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
6+
import java.util.Optional;
67

78
public class DefaultContext<T extends CustomResource> implements Context<T> {
89

10+
private final RetryInfo retryInfo;
911
private final EventList events;
1012
private final EventSourceManager eventSourceManager;
1113

12-
public DefaultContext(EventSourceManager eventSourceManager, EventList events) {
14+
public DefaultContext(
15+
EventSourceManager eventSourceManager, EventList events, RetryInfo retryInfo) {
16+
this.retryInfo = retryInfo;
1317
this.events = events;
1418
this.eventSourceManager = eventSourceManager;
1519
}
@@ -23,4 +27,9 @@ public EventSourceManager getEventSourceManager() {
2327
public EventList getEvents() {
2428
return events;
2529
}
30+
31+
@Override
32+
public Optional<RetryInfo> getRetryInfo() {
33+
return Optional.ofNullable(retryInfo);
34+
}
2635
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/api/RetryInfo.java

Lines changed: 3 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,8 @@
11
package io.javaoperatorsdk.operator.api;
22

3-
public class RetryInfo {
3+
public interface RetryInfo {
44

5-
private int retryNumber;
6-
private boolean lastAttempt;
5+
int getAttemptCount();
76

8-
public RetryInfo(int retryNumber, boolean lastAttempt) {
9-
this.retryNumber = retryNumber;
10-
this.lastAttempt = lastAttempt;
11-
}
12-
13-
public int getRetryNumber() {
14-
return retryNumber;
15-
}
16-
17-
public boolean isLastAttempt() {
18-
return lastAttempt;
19-
}
7+
boolean isLastAttempt();
208
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java

Lines changed: 79 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,13 @@
55
import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
66

77
import io.fabric8.kubernetes.client.CustomResource;
8+
import io.javaoperatorsdk.operator.api.RetryInfo;
89
import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
910
import io.javaoperatorsdk.operator.processing.event.Event;
1011
import io.javaoperatorsdk.operator.processing.event.EventHandler;
12+
import io.javaoperatorsdk.operator.processing.retry.Retry;
13+
import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
14+
import java.util.*;
1115
import java.util.HashSet;
1216
import java.util.Optional;
1317
import java.util.Set;
@@ -30,16 +34,20 @@ public class DefaultEventHandler implements EventHandler {
3034
private final Set<String> underProcessing = new HashSet<>();
3135
private final ScheduledThreadPoolExecutor executor;
3236
private final EventDispatcher eventDispatcher;
33-
private DefaultEventSourceManager defaultEventSourceManager;
37+
private final Retry retry;
38+
private final Map<String, RetryExecution> retryState = new HashMap<>();
39+
private DefaultEventSourceManager eventSourceManager;
3440

3541
private final ReentrantLock lock = new ReentrantLock();
3642

3743
public DefaultEventHandler(
3844
CustomResourceCache customResourceCache,
3945
EventDispatcher eventDispatcher,
40-
String relatedControllerName) {
46+
String relatedControllerName,
47+
Retry retry) {
4148
this.customResourceCache = customResourceCache;
4249
this.eventDispatcher = eventDispatcher;
50+
this.retry = retry;
4351
eventBuffer = new EventBuffer();
4452
executor =
4553
new ScheduledThreadPoolExecutor(
@@ -52,8 +60,8 @@ public Thread newThread(Runnable runnable) {
5260
});
5361
}
5462

55-
public void setDefaultEventSourceManager(DefaultEventSourceManager defaultEventSourceManager) {
56-
this.defaultEventSourceManager = defaultEventSourceManager;
63+
public void setEventSourceManager(DefaultEventSourceManager eventSourceManager) {
64+
this.eventSourceManager = eventSourceManager;
5765
}
5866

5967
@Override
@@ -79,7 +87,8 @@ private void executeBufferedEvents(String customResourceUid) {
7987
ExecutionScope executionScope =
8088
new ExecutionScope(
8189
eventBuffer.getAndRemoveEventsForExecution(customResourceUid),
82-
latestCustomResource.get());
90+
latestCustomResource.get(),
91+
retryInfo(customResourceUid));
8392
log.debug("Executing events for custom resource. Scope: {}", executionScope);
8493
executor.execute(new ExecutionConsumer(executionScope, eventDispatcher, this));
8594
} else {
@@ -93,12 +102,28 @@ private void executeBufferedEvents(String customResourceUid) {
93102
}
94103
}
95104

105+
private RetryInfo retryInfo(String customResourceUid) {
106+
return retryState.get(customResourceUid);
107+
}
108+
96109
void eventProcessingFinished(
97110
ExecutionScope executionScope, PostExecutionControl postExecutionControl) {
98111
try {
99112
lock.lock();
100-
log.debug("Event processing finished. Scope: {}", executionScope);
113+
log.debug(
114+
"Event processing finished. Scope: {}, PostExecutionControl: {}",
115+
executionScope,
116+
postExecutionControl);
101117
unsetUnderExecution(executionScope.getCustomResourceUid());
118+
119+
if (retry != null && postExecutionControl.exceptionDuringExecution()) {
120+
handleRetryOnException(executionScope);
121+
return;
122+
}
123+
124+
if (retry != null) {
125+
markSuccessfulExecutionRegardingRetry(executionScope);
126+
}
102127
if (containsCustomResourceDeletedEvent(executionScope.getEvents())) {
103128
cleanupAfterDeletedEvent(executionScope.getCustomResourceUid());
104129
} else {
@@ -110,6 +135,53 @@ void eventProcessingFinished(
110135
}
111136
}
112137

138+
/**
139+
* Regarding the events there are 2 approaches we can take. Either retry always when there are new
140+
* events (received meanwhile retry is in place or already in buffer) instantly or always wait
141+
* according to the retry timing if there was an exception.
142+
*/
143+
private void handleRetryOnException(ExecutionScope executionScope) {
144+
RetryExecution execution = getOrInitRetryExecution(executionScope);
145+
boolean newEventsExists = eventBuffer.newEventsExists(executionScope.getCustomResourceUid());
146+
eventBuffer.putBackEvents(executionScope.getCustomResourceUid(), executionScope.getEvents());
147+
148+
if (newEventsExists) {
149+
log.debug("New events exists for for resource id: {}", executionScope.getCustomResourceUid());
150+
executeBufferedEvents(executionScope.getCustomResourceUid());
151+
return;
152+
}
153+
Optional<Long> nextDelay = execution.nextDelay();
154+
155+
nextDelay.ifPresent(
156+
delay -> {
157+
log.debug(
158+
"Scheduling timer event for retry with delay:{} for resource: {}",
159+
delay,
160+
executionScope.getCustomResourceUid());
161+
eventSourceManager
162+
.getRetryTimerEventSource()
163+
.scheduleOnce(executionScope.getCustomResource(), delay);
164+
});
165+
}
166+
167+
private void markSuccessfulExecutionRegardingRetry(ExecutionScope executionScope) {
168+
log.debug(
169+
"Marking successful execution for resource: {}", executionScope.getCustomResourceUid());
170+
retryState.remove(executionScope.getCustomResourceUid());
171+
eventSourceManager
172+
.getRetryTimerEventSource()
173+
.cancelOnceSchedule(executionScope.getCustomResourceUid());
174+
}
175+
176+
private RetryExecution getOrInitRetryExecution(ExecutionScope executionScope) {
177+
RetryExecution retryExecution = retryState.get(executionScope.getCustomResourceUid());
178+
if (retryExecution == null) {
179+
retryExecution = retry.initExecution();
180+
retryState.put(executionScope.getCustomResourceUid(), retryExecution);
181+
}
182+
return retryExecution;
183+
}
184+
113185
/**
114186
* Here we try to cache the latest resource after an update. The goal is to solve a concurrency
115187
* issue we've seen: If an execution is finished, where we updated a custom resource, but there
@@ -146,7 +218,7 @@ private void cacheUpdatedResourceIfChanged(
146218
}
147219

148220
private void cleanupAfterDeletedEvent(String customResourceUid) {
149-
defaultEventSourceManager.cleanup(customResourceUid);
221+
eventSourceManager.cleanup(customResourceUid);
150222
eventBuffer.cleanup(customResourceUid);
151223
customResourceCache.cleanup(customResourceUid);
152224
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventBuffer.java

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,16 @@ public void addEvent(Event event) {
1717
crEvents.add(event);
1818
}
1919

20+
public boolean newEventsExists(String resourceId) {
21+
return events.get(resourceId) != null && !events.get(resourceId).isEmpty();
22+
}
23+
24+
public void putBackEvents(String resourceUid, List<Event> oldEvents) {
25+
List<Event> crEvents =
26+
events.computeIfAbsent(resourceUid, (id) -> new ArrayList<>(oldEvents.size()));
27+
crEvents.addAll(0, oldEvents);
28+
}
29+
2030
public boolean containsEvents(String customResourceId) {
2131
return events.get(customResourceId) != null;
2232
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/EventDispatcher.java

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -43,16 +43,16 @@ public void setEventSourceManager(EventSourceManager eventSourceManager) {
4343
this.eventSourceManager = eventSourceManager;
4444
}
4545

46-
public PostExecutionControl handleEvent(ExecutionScope event) {
46+
public PostExecutionControl handleExecution(ExecutionScope executionScope) {
4747
try {
48-
return handDispatch(event);
48+
return handleDispatch(executionScope);
4949
} catch (RuntimeException e) {
50-
log.error("Error during event processing {} failed.", event, e);
51-
return PostExecutionControl.defaultDispatch();
50+
log.error("Error during event processing {} failed.", executionScope, e);
51+
return PostExecutionControl.exceptionDuringExecution(e);
5252
}
5353
}
5454

55-
private PostExecutionControl handDispatch(ExecutionScope executionScope) {
55+
private PostExecutionControl handleDispatch(ExecutionScope executionScope) {
5656
CustomResource resource = executionScope.getCustomResource();
5757
log.debug(
5858
"Handling events: {} for resource {}", executionScope.getEvents(), resource.getMetadata());
@@ -72,7 +72,10 @@ private PostExecutionControl handDispatch(ExecutionScope executionScope) {
7272
return PostExecutionControl.defaultDispatch();
7373
}
7474
Context context =
75-
new DefaultContext(eventSourceManager, new EventList(executionScope.getEvents()));
75+
new DefaultContext(
76+
eventSourceManager,
77+
new EventList(executionScope.getEvents()),
78+
executionScope.getRetryInfo());
7679
if (markedForDeletion(resource)) {
7780
return handleDelete(resource, context);
7881
} else {

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionConsumer.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ class ExecutionConsumer implements Runnable {
2222

2323
@Override
2424
public void run() {
25-
PostExecutionControl postExecutionControl = eventDispatcher.handleEvent(executionScope);
25+
PostExecutionControl postExecutionControl = eventDispatcher.handleExecution(executionScope);
2626
defaultEventHandler.eventProcessingFinished(executionScope, postExecutionControl);
2727
}
2828
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/ExecutionScope.java

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package io.javaoperatorsdk.operator.processing;
22

33
import io.fabric8.kubernetes.client.CustomResource;
4+
import io.javaoperatorsdk.operator.api.RetryInfo;
45
import io.javaoperatorsdk.operator.processing.event.Event;
56
import java.util.List;
67

@@ -10,9 +11,12 @@ public class ExecutionScope {
1011
// the latest custom resource from cache
1112
private CustomResource customResource;
1213

13-
public ExecutionScope(List<Event> list, CustomResource customResource) {
14+
private RetryInfo retryInfo;
15+
16+
public ExecutionScope(List<Event> list, CustomResource customResource, RetryInfo retryInfo) {
1417
this.events = list;
1518
this.customResource = customResource;
19+
this.retryInfo = retryInfo;
1620
}
1721

1822
public List<Event> getEvents() {
@@ -38,4 +42,8 @@ public String toString() {
3842
+ customResource.getMetadata().getResourceVersion()
3943
+ '}';
4044
}
45+
46+
public RetryInfo getRetryInfo() {
47+
return retryInfo;
48+
}
4149
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/PostExecutionControl.java

Lines changed: 34 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,21 +9,31 @@ public final class PostExecutionControl {
99

1010
private final CustomResource updatedCustomResource;
1111

12-
private PostExecutionControl(boolean onlyFinalizerHandled, CustomResource updatedCustomResource) {
12+
private final RuntimeException runtimeException;
13+
14+
private PostExecutionControl(
15+
boolean onlyFinalizerHandled,
16+
CustomResource updatedCustomResource,
17+
RuntimeException runtimeException) {
1318
this.onlyFinalizerHandled = onlyFinalizerHandled;
1419
this.updatedCustomResource = updatedCustomResource;
20+
this.runtimeException = runtimeException;
1521
}
1622

1723
public static PostExecutionControl onlyFinalizerAdded() {
18-
return new PostExecutionControl(true, null);
24+
return new PostExecutionControl(true, null, null);
1925
}
2026

2127
public static PostExecutionControl defaultDispatch() {
22-
return new PostExecutionControl(false, null);
28+
return new PostExecutionControl(false, null, null);
2329
}
2430

2531
public static PostExecutionControl customResourceUpdated(CustomResource updatedCustomResource) {
26-
return new PostExecutionControl(false, updatedCustomResource);
32+
return new PostExecutionControl(false, updatedCustomResource, null);
33+
}
34+
35+
public static PostExecutionControl exceptionDuringExecution(RuntimeException exception) {
36+
return new PostExecutionControl(false, null, exception);
2737
}
2838

2939
public boolean isOnlyFinalizerHandled() {
@@ -37,4 +47,24 @@ public Optional<CustomResource> getUpdatedCustomResource() {
3747
public boolean customResourceUpdatedDuringExecution() {
3848
return updatedCustomResource != null;
3949
}
50+
51+
public boolean exceptionDuringExecution() {
52+
return runtimeException != null;
53+
}
54+
55+
public Optional<RuntimeException> getRuntimeException() {
56+
return Optional.ofNullable(runtimeException);
57+
}
58+
59+
@Override
60+
public String toString() {
61+
return "PostExecutionControl{"
62+
+ "onlyFinalizerHandled="
63+
+ onlyFinalizerHandled
64+
+ ", updatedCustomResource="
65+
+ updatedCustomResource
66+
+ ", runtimeException="
67+
+ runtimeException
68+
+ '}';
69+
}
4070
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManager.java

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import io.javaoperatorsdk.operator.processing.DefaultEventHandler;
44
import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
5+
import io.javaoperatorsdk.operator.processing.event.internal.TimerEventSource;
56
import java.util.Collections;
67
import java.util.Map;
78
import java.util.Optional;
@@ -12,15 +13,21 @@
1213

1314
public class DefaultEventSourceManager implements EventSourceManager {
1415

16+
public static final String RETRY_TIMER_EVENT_SOURCE_NAME = "retry-timer-event-source";
1517
private static final Logger log = LoggerFactory.getLogger(DefaultEventSourceManager.class);
1618

1719
private final ReentrantLock lock = new ReentrantLock();
1820
private Map<String, EventSource> eventSources = new ConcurrentHashMap<>();
1921
private CustomResourceEventSource customResourceEventSource;
2022
private DefaultEventHandler defaultEventHandler;
23+
private TimerEventSource retryTimerEventSource;
2124

22-
public DefaultEventSourceManager(DefaultEventHandler defaultEventHandler) {
25+
public DefaultEventSourceManager(DefaultEventHandler defaultEventHandler, boolean supportRetry) {
2326
this.defaultEventHandler = defaultEventHandler;
27+
if (supportRetry) {
28+
this.retryTimerEventSource = new TimerEventSource();
29+
registerEventSource(RETRY_TIMER_EVENT_SOURCE_NAME, retryTimerEventSource);
30+
}
2431
}
2532

2633
public void registerCustomResourceEventSource(
@@ -66,6 +73,10 @@ public Optional<EventSource> deRegisterCustomResourceFromEventSource(
6673
}
6774
}
6875

76+
public TimerEventSource getRetryTimerEventSource() {
77+
return retryTimerEventSource;
78+
}
79+
6980
@Override
7081
public Map<String, EventSource> getRegisteredEventSources() {
7182
return Collections.unmodifiableMap(eventSources);

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetry.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ public static GenericRetry defaultLimitedExponentialRetry() {
1616
}
1717

1818
public static GenericRetry noRetry() {
19-
return new GenericRetry().setMaxAttempts(1);
19+
return new GenericRetry().setMaxAttempts(0);
2020
}
2121

2222
public static GenericRetry every10second10TimesRetry() {

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecution.java

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -14,15 +14,7 @@ public GenericRetryExecution(GenericRetry genericRetry) {
1414
this.currentInterval = genericRetry.getInitialInterval();
1515
}
1616

17-
/**
18-
* Note that first attempt is always 0. Since this implementation is tailored for event
19-
* scheduling.
20-
*/
2117
public Optional<Long> nextDelay() {
22-
if (lastAttemptIndex == 0) {
23-
lastAttemptIndex++;
24-
return Optional.of(0L);
25-
}
2618
if (genericRetry.getMaxAttempts() > -1 && lastAttemptIndex >= genericRetry.getMaxAttempts()) {
2719
return Optional.empty();
2820
}
@@ -37,7 +29,12 @@ public Optional<Long> nextDelay() {
3729
}
3830

3931
@Override
40-
public boolean isLastExecution() {
32+
public boolean isLastAttempt() {
4133
return genericRetry.getMaxAttempts() > -1 && lastAttemptIndex >= genericRetry.getMaxAttempts();
4234
}
35+
36+
@Override
37+
public int getAttemptCount() {
38+
return lastAttemptIndex;
39+
}
4340
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/retry/RetryExecution.java

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
11
package io.javaoperatorsdk.operator.processing.retry;
22

3+
import io.javaoperatorsdk.operator.api.RetryInfo;
34
import java.util.Optional;
45

5-
public interface RetryExecution {
6+
public interface RetryExecution extends RetryInfo {
67

78
/**
89
* Calculates the delay for the next execution. This method should return 0, when called first
@@ -11,10 +12,4 @@ public interface RetryExecution {
1112
* @return
1213
*/
1314
Optional<Long> nextDelay();
14-
15-
/**
16-
* @return true, if the last returned delay is, the last returned values, thus there will be no
17-
* further retry
18-
*/
19-
boolean isLastExecution();
2015
}

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/EventDispatcherTest.java

Lines changed: 44 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.javaoperatorsdk.operator;
22

3+
import static org.assertj.core.api.Assertions.assertThat;
34
import static org.junit.jupiter.api.Assertions.assertEquals;
45
import static org.mockito.Mockito.any;
56
import static org.mockito.Mockito.argThat;
@@ -12,9 +13,7 @@
1213

1314
import io.fabric8.kubernetes.client.CustomResource;
1415
import io.fabric8.kubernetes.client.Watcher;
15-
import io.javaoperatorsdk.operator.api.DeleteControl;
16-
import io.javaoperatorsdk.operator.api.ResourceController;
17-
import io.javaoperatorsdk.operator.api.UpdateControl;
16+
import io.javaoperatorsdk.operator.api.*;
1817
import io.javaoperatorsdk.operator.processing.EventDispatcher;
1918
import io.javaoperatorsdk.operator.processing.ExecutionScope;
2019
import io.javaoperatorsdk.operator.processing.event.Event;
@@ -25,6 +24,7 @@
2524
import java.util.List;
2625
import org.junit.jupiter.api.BeforeEach;
2726
import org.junit.jupiter.api.Test;
27+
import org.mockito.ArgumentCaptor;
2828
import org.mockito.ArgumentMatchers;
2929

3030
class EventDispatcherTest {
@@ -54,7 +54,7 @@ void setup() {
5454

5555
@Test
5656
void callCreateOrUpdateOnNewResource() {
57-
eventDispatcher.handleEvent(
57+
eventDispatcher.handleExecution(
5858
executionScopeWithCREvent(Watcher.Action.ADDED, testCustomResource));
5959
verify(controller, times(1))
6060
.createOrUpdateResource(ArgumentMatchers.eq(testCustomResource), any());
@@ -65,7 +65,7 @@ void updatesOnlyStatusSubResource() {
6565
when(controller.createOrUpdateResource(eq(testCustomResource), any()))
6666
.thenReturn(UpdateControl.updateStatusSubResource(testCustomResource));
6767

68-
eventDispatcher.handleEvent(
68+
eventDispatcher.handleExecution(
6969
executionScopeWithCREvent(Watcher.Action.ADDED, testCustomResource));
7070

7171
verify(customResourceFacade, times(1)).updateStatus(testCustomResource);
@@ -74,15 +74,15 @@ void updatesOnlyStatusSubResource() {
7474

7575
@Test
7676
void callCreateOrUpdateOnModifiedResource() {
77-
eventDispatcher.handleEvent(
77+
eventDispatcher.handleExecution(
7878
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
7979
verify(controller, times(1))
8080
.createOrUpdateResource(ArgumentMatchers.eq(testCustomResource), any());
8181
}
8282

8383
@Test
8484
void adsDefaultFinalizerOnCreateIfNotThere() {
85-
eventDispatcher.handleEvent(
85+
eventDispatcher.handleExecution(
8686
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
8787
verify(controller, times(1))
8888
.createOrUpdateResource(
@@ -97,7 +97,7 @@ void callsDeleteIfObjectHasFinalizerAndMarkedForDelete() {
9797
testCustomResource.getMetadata().setDeletionTimestamp("2019-8-10");
9898
testCustomResource.getMetadata().getFinalizers().add(DEFAULT_FINALIZER);
9999

100-
eventDispatcher.handleEvent(
100+
eventDispatcher.handleExecution(
101101
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
102102

103103
verify(controller, times(1)).deleteResource(eq(testCustomResource), any());
@@ -108,7 +108,7 @@ void callsDeleteIfObjectHasFinalizerAndMarkedForDelete() {
108108
void callDeleteOnControllerIfMarkedForDeletionButThereIsNoDefaultFinalizer() {
109109
markForDeletion(testCustomResource);
110110

111-
eventDispatcher.handleEvent(
111+
eventDispatcher.handleExecution(
112112
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
113113

114114
verify(controller).deleteResource(eq(testCustomResource), any());
@@ -118,7 +118,7 @@ void callDeleteOnControllerIfMarkedForDeletionButThereIsNoDefaultFinalizer() {
118118
void removesDefaultFinalizerOnDelete() {
119119
markForDeletion(testCustomResource);
120120

121-
eventDispatcher.handleEvent(
121+
eventDispatcher.handleExecution(
122122
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
123123

124124
assertEquals(0, testCustomResource.getMetadata().getFinalizers().size());
@@ -131,7 +131,7 @@ void doesNotRemovesTheFinalizerIfTheDeleteNotMethodInstructsIt() {
131131
.thenReturn(DeleteControl.NO_FINALIZER_REMOVAL);
132132
markForDeletion(testCustomResource);
133133

134-
eventDispatcher.handleEvent(
134+
eventDispatcher.handleExecution(
135135
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
136136

137137
assertEquals(1, testCustomResource.getMetadata().getFinalizers().size());
@@ -143,7 +143,7 @@ void doesNotUpdateTheResourceIfNoUpdateUpdateControl() {
143143
when(controller.createOrUpdateResource(eq(testCustomResource), any()))
144144
.thenReturn(UpdateControl.noUpdate());
145145

146-
eventDispatcher.handleEvent(
146+
eventDispatcher.handleExecution(
147147
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
148148
verify(customResourceFacade, never()).replaceWithLock(any());
149149
verify(customResourceFacade, never()).updateStatus(testCustomResource);
@@ -155,7 +155,7 @@ void addsFinalizerIfNotMarkedForDeletionAndEmptyCustomResourceReturned() {
155155
when(controller.createOrUpdateResource(eq(testCustomResource), any()))
156156
.thenReturn(UpdateControl.noUpdate());
157157

158-
eventDispatcher.handleEvent(
158+
eventDispatcher.handleExecution(
159159
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
160160

161161
assertEquals(1, testCustomResource.getMetadata().getFinalizers().size());
@@ -167,7 +167,7 @@ void doesNotCallDeleteIfMarkedForDeletionButNotOurFinalizer() {
167167
removeFinalizers(testCustomResource);
168168
markForDeletion(testCustomResource);
169169

170-
eventDispatcher.handleEvent(
170+
eventDispatcher.handleExecution(
171171
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
172172

173173
verify(customResourceFacade, never()).replaceWithLock(any());
@@ -176,14 +176,41 @@ void doesNotCallDeleteIfMarkedForDeletionButNotOurFinalizer() {
176176

177177
@Test
178178
void executeControllerRegardlessGenerationInNonGenerationAwareMode() {
179-
eventDispatcher.handleEvent(
179+
eventDispatcher.handleExecution(
180180
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
181-
eventDispatcher.handleEvent(
181+
eventDispatcher.handleExecution(
182182
executionScopeWithCREvent(Watcher.Action.MODIFIED, testCustomResource));
183183

184184
verify(controller, times(2)).createOrUpdateResource(eq(testCustomResource), any());
185185
}
186186

187+
@Test
188+
void propagatesRetryInfoToContext() {
189+
eventDispatcher.handleExecution(
190+
new ExecutionScope(
191+
Arrays.asList(),
192+
testCustomResource,
193+
new RetryInfo() {
194+
@Override
195+
public int getAttemptCount() {
196+
return 2;
197+
}
198+
199+
@Override
200+
public boolean isLastAttempt() {
201+
return true;
202+
}
203+
}));
204+
205+
ArgumentCaptor<Context<CustomResource>> contextArgumentCaptor =
206+
ArgumentCaptor.forClass(Context.class);
207+
verify(controller, times(1))
208+
.createOrUpdateResource(eq(testCustomResource), contextArgumentCaptor.capture());
209+
Context<CustomResource> context = contextArgumentCaptor.getValue();
210+
assertThat(context.getRetryInfo().get().getAttemptCount()).isEqualTo(2);
211+
assertThat(context.getRetryInfo().get().isLastAttempt()).isEqualTo(true);
212+
}
213+
187214
private void markForDeletion(CustomResource customResource) {
188215
customResource.getMetadata().setDeletionTimestamp("2019-8-10");
189216
}
@@ -198,6 +225,6 @@ public ExecutionScope executionScopeWithCREvent(
198225
List<Event> eventList = new ArrayList<>(1 + otherEvents.length);
199226
eventList.add(event);
200227
eventList.addAll(Arrays.asList(otherEvents));
201-
return new ExecutionScope(eventList, resource);
228+
return new ExecutionScope(eventList, resource, null);
202229
}
203230
}

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/IntegrationTestSupport.java

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
1717
import io.fabric8.kubernetes.client.utils.Serialization;
1818
import io.javaoperatorsdk.operator.api.ResourceController;
19+
import io.javaoperatorsdk.operator.processing.retry.Retry;
1920
import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
2021
import io.javaoperatorsdk.operator.sample.simple.TestCustomResourceSpec;
2122
import java.io.IOException;
@@ -41,6 +42,11 @@ public class IntegrationTestSupport {
4142

4243
public void initialize(
4344
KubernetesClient k8sClient, ResourceController controller, String crdPath) {
45+
initialize(k8sClient, controller, crdPath, null);
46+
}
47+
48+
public void initialize(
49+
KubernetesClient k8sClient, ResourceController controller, String crdPath, Retry retry) {
4450
log.info("Initializing integration test in namespace {}", TEST_NAMESPACE);
4551
this.k8sClient = k8sClient;
4652
CustomResourceDefinition crd = loadCRDAndApplyToCluster(crdPath);
@@ -62,7 +68,7 @@ public void initialize(
6268
.build());
6369
}
6470
operator = new Operator(k8sClient);
65-
operator.registerController(controller, TEST_NAMESPACE);
71+
operator.registerController(controller, retry, TEST_NAMESPACE);
6672
log.info("Operator is running with {}", controller.getClass().getCanonicalName());
6773
}
6874

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/RetryIT.java

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
package io.javaoperatorsdk.operator;
2+
3+
import static io.javaoperatorsdk.operator.IntegrationTestSupport.TEST_NAMESPACE;
4+
import static io.javaoperatorsdk.operator.sample.event.EventSourceTestCustomResourceController.*;
5+
import static io.javaoperatorsdk.operator.sample.retry.RetryTestCustomResourceStatus.State.SUCCESS;
6+
import static org.assertj.core.api.Assertions.assertThat;
7+
8+
import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
9+
import io.fabric8.kubernetes.client.DefaultKubernetesClient;
10+
import io.fabric8.kubernetes.client.KubernetesClient;
11+
import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
12+
import io.javaoperatorsdk.operator.processing.retry.Retry;
13+
import io.javaoperatorsdk.operator.sample.retry.RetryTestCustomResource;
14+
import io.javaoperatorsdk.operator.sample.retry.RetryTestCustomResourceController;
15+
import io.javaoperatorsdk.operator.sample.retry.RetryTestCustomResourceSpec;
16+
import org.junit.jupiter.api.BeforeEach;
17+
import org.junit.jupiter.api.Test;
18+
19+
public class RetryIT {
20+
21+
public static final int RETRY_INTERVAL = 150;
22+
private IntegrationTestSupport integrationTestSupport = new IntegrationTestSupport();
23+
24+
@BeforeEach
25+
public void initAndCleanup() {
26+
Retry retry =
27+
new GenericRetry().setInitialInterval(RETRY_INTERVAL).withLinearRetry().setMaxAttempts(5);
28+
KubernetesClient k8sClient = new DefaultKubernetesClient();
29+
integrationTestSupport.initialize(
30+
k8sClient, new RetryTestCustomResourceController(), "retry-test-crd.yaml", retry);
31+
integrationTestSupport.cleanup();
32+
}
33+
34+
@Test
35+
public void retryFailedExecution() {
36+
integrationTestSupport.teardownIfSuccess(
37+
() -> {
38+
RetryTestCustomResource resource = createTestCustomResource("1");
39+
integrationTestSupport.getCrOperations().inNamespace(TEST_NAMESPACE).create(resource);
40+
41+
Thread.sleep(
42+
RETRY_INTERVAL * (RetryTestCustomResourceController.NUMBER_FAILED_EXECUTIONS + 2));
43+
44+
assertThat(integrationTestSupport.numberOfControllerExecutions())
45+
.isGreaterThanOrEqualTo(
46+
RetryTestCustomResourceController.NUMBER_FAILED_EXECUTIONS + 1);
47+
48+
RetryTestCustomResource finalResource =
49+
(RetryTestCustomResource)
50+
integrationTestSupport
51+
.getCrOperations()
52+
.inNamespace(TEST_NAMESPACE)
53+
.withName(resource.getMetadata().getName())
54+
.get();
55+
assertThat(finalResource.getStatus().getState()).isEqualTo(SUCCESS);
56+
});
57+
}
58+
59+
public RetryTestCustomResource createTestCustomResource(String id) {
60+
RetryTestCustomResource resource = new RetryTestCustomResource();
61+
resource.setMetadata(
62+
new ObjectMetaBuilder()
63+
.withName("retrysource-" + id)
64+
.withNamespace(TEST_NAMESPACE)
65+
.withFinalizers(RetryTestCustomResourceController.FINALIZER_NAME)
66+
.build());
67+
resource.setKind("retrysample");
68+
resource.setSpec(new RetryTestCustomResourceSpec());
69+
resource.getSpec().setValue(id);
70+
return resource;
71+
}
72+
}

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/DefaultEventHandlerTest.java

Lines changed: 117 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,10 @@
22

33
import static io.javaoperatorsdk.operator.TestUtils.testCustomResource;
44
import static org.assertj.core.api.Assertions.assertThat;
5+
import static org.mockito.ArgumentMatchers.eq;
56
import static org.mockito.Mockito.any;
67
import static org.mockito.Mockito.mock;
8+
import static org.mockito.Mockito.never;
79
import static org.mockito.Mockito.timeout;
810
import static org.mockito.Mockito.times;
911
import static org.mockito.Mockito.verify;
@@ -14,35 +16,54 @@
1416
import io.javaoperatorsdk.operator.processing.event.Event;
1517
import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEvent;
1618
import io.javaoperatorsdk.operator.processing.event.internal.TimerEvent;
19+
import io.javaoperatorsdk.operator.processing.event.internal.TimerEventSource;
20+
import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
1721
import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
22+
import java.util.Arrays;
1823
import java.util.List;
1924
import java.util.UUID;
2025
import org.junit.jupiter.api.BeforeEach;
2126
import org.junit.jupiter.api.Test;
2227
import org.mockito.ArgumentCaptor;
2328
import org.mockito.stubbing.Answer;
29+
import org.slf4j.Logger;
30+
import org.slf4j.LoggerFactory;
2431

2532
class DefaultEventHandlerTest {
2633

34+
private static final Logger log = LoggerFactory.getLogger(DefaultEventHandlerTest.class);
35+
2736
public static final int FAKE_CONTROLLER_EXECUTION_DURATION = 250;
2837
public static final int SEPARATE_EXECUTION_TIMEOUT = 450;
2938
private EventDispatcher eventDispatcherMock = mock(EventDispatcher.class);
3039
private CustomResourceCache customResourceCache = new CustomResourceCache();
31-
private DefaultEventHandler defaultEventHandler =
32-
new DefaultEventHandler(customResourceCache, eventDispatcherMock, "Test");
3340
private DefaultEventSourceManager defaultEventSourceManagerMock =
3441
mock(DefaultEventSourceManager.class);
42+
private TimerEventSource retryTimerEventSourceMock = mock(TimerEventSource.class);
43+
44+
private DefaultEventHandler defaultEventHandler =
45+
new DefaultEventHandler(customResourceCache, eventDispatcherMock, "Test", null);
46+
47+
private DefaultEventHandler defaultEventHandlerWithRetry =
48+
new DefaultEventHandler(
49+
customResourceCache,
50+
eventDispatcherMock,
51+
"Test",
52+
GenericRetry.defaultLimitedExponentialRetry());
3553

3654
@BeforeEach
3755
public void setup() {
38-
defaultEventHandler.setDefaultEventSourceManager(defaultEventSourceManagerMock);
56+
when(defaultEventSourceManagerMock.getRetryTimerEventSource())
57+
.thenReturn(retryTimerEventSourceMock);
58+
defaultEventHandler.setEventSourceManager(defaultEventSourceManagerMock);
59+
defaultEventHandlerWithRetry.setEventSourceManager(defaultEventSourceManagerMock);
3960
}
4061

4162
@Test
4263
public void dispatchesEventsIfNoExecutionInProgress() {
4364
defaultEventHandler.handleEvent(prepareCREvent());
4465

45-
verify(eventDispatcherMock, timeout(50).times(1)).handleEvent(any());
66+
verify(eventDispatcherMock, timeout(50).times(1)).handleExecution(any());
4667
}
4768

4869
@Test
@@ -52,7 +73,7 @@ public void skipProcessingIfLatestCustomResourceNotInCache() {
5273

5374
defaultEventHandler.handleEvent(event);
5475

55-
verify(eventDispatcherMock, timeout(50).times(0)).handleEvent(any());
76+
verify(eventDispatcherMock, timeout(50).times(0)).handleExecution(any());
5677
}
5778

5879
@Test
@@ -61,7 +82,8 @@ public void ifExecutionInProgressWaitsUntilItsFinished() throws InterruptedExcep
6182

6283
defaultEventHandler.handleEvent(nonCREvent(resourceUid));
6384

64-
verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1)).handleEvent(any());
85+
verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
86+
.handleExecution(any());
6587
}
6688

6789
@Test
@@ -73,7 +95,7 @@ public void buffersAllIncomingEventsWhileControllerInExecution() {
7395

7496
ArgumentCaptor<ExecutionScope> captor = ArgumentCaptor.forClass(ExecutionScope.class);
7597
verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(2))
76-
.handleEvent(captor.capture());
98+
.handleExecution(captor.capture());
7799
List<Event> events = captor.getAllValues().get(1).getEvents();
78100
assertThat(events).hasSize(2);
79101
assertThat(events.get(0)).isInstanceOf(TimerEvent.class);
@@ -89,13 +111,99 @@ public void cleanUpAfterDeleteEvent() {
89111
String uid = customResource.getMetadata().getUid();
90112

91113
defaultEventHandler.handleEvent(event);
92-
// todo awaitility?
114+
93115
waitMinimalTime();
94116

95117
verify(defaultEventSourceManagerMock, times(1)).cleanup(uid);
96118
assertThat(customResourceCache.getLatestResource(uid)).isNotPresent();
97119
}
98120

121+
@Test
122+
public void schedulesAnEventRetryOnException() {
123+
Event event = prepareCREvent();
124+
TestCustomResource customResource = testCustomResource();
125+
126+
ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource, null);
127+
PostExecutionControl postExecutionControl =
128+
PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
129+
130+
defaultEventHandlerWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
131+
132+
verify(retryTimerEventSourceMock, times(1))
133+
.scheduleOnce(eq(customResource), eq(GenericRetry.DEFAULT_INITIAL_INTERVAL));
134+
}
135+
136+
@Test
137+
public void executesTheControllerInstantlyAfterErrorIfEventsBuffered() {
138+
Event event = prepareCREvent();
139+
TestCustomResource customResource = testCustomResource();
140+
customResource.getMetadata().setUid(event.getRelatedCustomResourceUid());
141+
ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource, null);
142+
PostExecutionControl postExecutionControl =
143+
PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
144+
145+
// start processing an event
146+
defaultEventHandlerWithRetry.handleEvent(event);
147+
// buffer an another event
148+
defaultEventHandlerWithRetry.handleEvent(event);
149+
verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
150+
.handleExecution(any());
151+
152+
defaultEventHandlerWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
153+
154+
ArgumentCaptor<ExecutionScope> executionScopeArgumentCaptor =
155+
ArgumentCaptor.forClass(ExecutionScope.class);
156+
verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(2))
157+
.handleExecution(executionScopeArgumentCaptor.capture());
158+
List<ExecutionScope> allValues = executionScopeArgumentCaptor.getAllValues();
159+
assertThat(allValues).hasSize(2);
160+
assertThat(allValues.get(1).getEvents()).hasSize(2);
161+
verify(retryTimerEventSourceMock, never())
162+
.scheduleOnce(eq(customResource), eq(GenericRetry.DEFAULT_INITIAL_INTERVAL));
163+
}
164+
165+
@Test
166+
public void successfulExecutionResetsTheRetry() {
167+
log.info("Starting successfulExecutionResetsTheRetry");
168+
169+
Event event = prepareCREvent();
170+
TestCustomResource customResource = testCustomResource();
171+
customResource.getMetadata().setUid(event.getRelatedCustomResourceUid());
172+
ExecutionScope executionScope = new ExecutionScope(Arrays.asList(event), customResource, null);
173+
PostExecutionControl postExecutionControlWithException =
174+
PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
175+
PostExecutionControl defaultDispatchControl = PostExecutionControl.defaultDispatch();
176+
177+
when(eventDispatcherMock.handleExecution(any()))
178+
.thenReturn(postExecutionControlWithException)
179+
.thenReturn(defaultDispatchControl);
180+
181+
ArgumentCaptor<ExecutionScope> executionScopeArgumentCaptor =
182+
ArgumentCaptor.forClass(ExecutionScope.class);
183+
184+
defaultEventHandlerWithRetry.handleEvent(event);
185+
186+
verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(1))
187+
.handleExecution(any());
188+
defaultEventHandlerWithRetry.handleEvent(event);
189+
190+
verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(2))
191+
.handleExecution(any());
192+
defaultEventHandlerWithRetry.handleEvent(event);
193+
194+
verify(eventDispatcherMock, timeout(SEPARATE_EXECUTION_TIMEOUT).times(3))
195+
.handleExecution(executionScopeArgumentCaptor.capture());
196+
log.info("Finished successfulExecutionResetsTheRetry");
197+
198+
List<ExecutionScope> executionScopes = executionScopeArgumentCaptor.getAllValues();
199+
200+
assertThat(executionScopes).hasSize(3);
201+
assertThat(executionScopes.get(0).getRetryInfo()).isNull();
202+
assertThat(executionScopes.get(2).getRetryInfo()).isNull();
203+
assertThat(executionScopes.get(1).getRetryInfo().getAttemptCount()).isEqualTo(1);
204+
assertThat(executionScopes.get(1).getRetryInfo().isLastAttempt()).isEqualTo(false);
205+
}
206+
99207
private void waitMinimalTime() {
100208
try {
101209
Thread.sleep(50);
@@ -105,7 +213,7 @@ private void waitMinimalTime() {
105213
}
106214

107215
private String eventAlreadyUnderProcessing() {
108-
when(eventDispatcherMock.handleEvent(any()))
216+
when(eventDispatcherMock.handleExecution(any()))
109217
.then(
110218
(Answer<PostExecutionControl>)
111219
invocationOnMock -> {

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/event/DefaultEventSourceManagerTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ class DefaultEventSourceManagerTest {
2121

2222
private DefaultEventHandler defaultEventHandlerMock = mock(DefaultEventHandler.class);
2323
private DefaultEventSourceManager defaultEventSourceManager =
24-
new DefaultEventSourceManager(defaultEventHandlerMock);
24+
new DefaultEventSourceManager(defaultEventHandlerMock, false);
2525

2626
@Test
2727
public void registersEventSource() {

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/retry/GenericRetryExecutionTest.java

Lines changed: 20 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.javaoperatorsdk.operator.processing.retry;
22

3+
import static io.javaoperatorsdk.operator.processing.retry.GenericRetry.DEFAULT_INITIAL_INTERVAL;
34
import static org.assertj.core.api.Assertions.assertThat;
45

56
import java.util.Optional;
@@ -8,27 +9,26 @@
89
public class GenericRetryExecutionTest {
910

1011
@Test
11-
public void forFirstBackOffAlwaysReturnsZero() {
12-
assertThat(getDefaultRetryExecution().nextDelay().get()).isEqualTo(0);
12+
public void forFirstBackOffAlwaysReturnsInitialInterval() {
13+
assertThat(getDefaultRetryExecution().nextDelay().get()).isEqualTo(DEFAULT_INITIAL_INTERVAL);
1314
}
1415

1516
@Test
1617
public void delayIsMultipliedEveryNextDelayCall() {
1718
RetryExecution retryExecution = getDefaultRetryExecution();
1819

19-
Optional<Long> res = callNextDelayNTimes(retryExecution, 2);
20-
assertThat(res.get()).isEqualTo(GenericRetry.DEFAULT_INITIAL_INTERVAL);
20+
Optional<Long> res = callNextDelayNTimes(retryExecution, 1);
21+
assertThat(res.get()).isEqualTo(DEFAULT_INITIAL_INTERVAL);
2122

2223
res = retryExecution.nextDelay();
2324
assertThat(res.get())
24-
.isEqualTo(
25-
(long) (GenericRetry.DEFAULT_INITIAL_INTERVAL * GenericRetry.DEFAULT_MULTIPLIER));
25+
.isEqualTo((long) (DEFAULT_INITIAL_INTERVAL * GenericRetry.DEFAULT_MULTIPLIER));
2626

2727
res = retryExecution.nextDelay();
2828
assertThat(res.get())
2929
.isEqualTo(
3030
(long)
31-
(GenericRetry.DEFAULT_INITIAL_INTERVAL
31+
(DEFAULT_INITIAL_INTERVAL
3232
* GenericRetry.DEFAULT_MULTIPLIER
3333
* GenericRetry.DEFAULT_MULTIPLIER));
3434
}
@@ -37,7 +37,7 @@ public void delayIsMultipliedEveryNextDelayCall() {
3737
public void noNextDelayIfMaxAttemptLimitReached() {
3838
RetryExecution retryExecution =
3939
GenericRetry.defaultLimitedExponentialRetry().setMaxAttempts(3).initExecution();
40-
Optional<Long> res = callNextDelayNTimes(retryExecution, 3);
40+
Optional<Long> res = callNextDelayNTimes(retryExecution, 2);
4141
assertThat(res).isNotEmpty();
4242

4343
res = retryExecution.nextDelay();
@@ -61,26 +61,34 @@ public void canLimitMaxIntervalLength() {
6161
@Test
6262
public void supportsNoRetry() {
6363
RetryExecution retryExecution = GenericRetry.noRetry().initExecution();
64-
assertThat(retryExecution.nextDelay().get()).isZero();
6564
assertThat(retryExecution.nextDelay()).isEmpty();
6665
}
6766

6867
@Test
6968
public void supportsIsLastExecution() {
7069
GenericRetryExecution execution = new GenericRetry().setMaxAttempts(2).initExecution();
71-
assertThat(execution.isLastExecution()).isFalse();
70+
assertThat(execution.isLastAttempt()).isFalse();
7271

7372
execution.nextDelay();
7473
execution.nextDelay();
75-
assertThat(execution.isLastExecution()).isTrue();
74+
assertThat(execution.isLastAttempt()).isTrue();
75+
}
76+
77+
@Test
78+
public void returnAttemptIndex() {
79+
RetryExecution retryExecution = GenericRetry.defaultLimitedExponentialRetry().initExecution();
80+
81+
assertThat(retryExecution.getAttemptCount()).isEqualTo(0);
82+
retryExecution.nextDelay();
83+
assertThat(retryExecution.getAttemptCount()).isEqualTo(1);
7684
}
7785

7886
private RetryExecution getDefaultRetryExecution() {
7987
return GenericRetry.defaultLimitedExponentialRetry().initExecution();
8088
}
8189

8290
public Optional<Long> callNextDelayNTimes(RetryExecution retryExecution, int n) {
83-
for (int i = 0; i < n - 1; i++) {
91+
for (int i = 0; i < n; i++) {
8492
retryExecution.nextDelay();
8593
}
8694
return retryExecution.nextDelay();

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResource.java

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
package io.javaoperatorsdk.operator.sample.retry;
2+
3+
import io.fabric8.kubernetes.client.CustomResource;
4+
5+
public class RetryTestCustomResource extends CustomResource {
6+
7+
private RetryTestCustomResourceSpec spec;
8+
9+
private RetryTestCustomResourceStatus status;
10+
11+
public RetryTestCustomResourceSpec getSpec() {
12+
return spec;
13+
}
14+
15+
public void setSpec(RetryTestCustomResourceSpec spec) {
16+
this.spec = spec;
17+
}
18+
19+
public RetryTestCustomResourceStatus getStatus() {
20+
return status;
21+
}
22+
23+
public void setStatus(RetryTestCustomResourceStatus status) {
24+
this.status = status;
25+
}
26+
27+
@Override
28+
public String toString() {
29+
return "TestCustomResource{"
30+
+ "spec="
31+
+ spec
32+
+ ", status="
33+
+ status
34+
+ ", extendedFrom="
35+
+ super.toString()
36+
+ '}';
37+
}
38+
}

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceController.java

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
package io.javaoperatorsdk.operator.sample.retry;
2+
3+
import io.javaoperatorsdk.operator.TestExecutionInfoProvider;
4+
import io.javaoperatorsdk.operator.api.*;
5+
import java.util.concurrent.atomic.AtomicInteger;
6+
import org.slf4j.Logger;
7+
import org.slf4j.LoggerFactory;
8+
9+
@Controller(crdName = RetryTestCustomResourceController.CRD_NAME)
10+
public class RetryTestCustomResourceController
11+
implements ResourceController<RetryTestCustomResource>, TestExecutionInfoProvider {
12+
13+
public static final int NUMBER_FAILED_EXECUTIONS = 2;
14+
15+
public static final String CRD_NAME = "retrysamples.sample.javaoperatorsdk";
16+
public static final String FINALIZER_NAME = CRD_NAME + "/finalizer";
17+
private static final Logger log =
18+
LoggerFactory.getLogger(RetryTestCustomResourceController.class);
19+
private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
20+
21+
@Override
22+
public DeleteControl deleteResource(
23+
RetryTestCustomResource resource, Context<RetryTestCustomResource> context) {
24+
return DeleteControl.DEFAULT_DELETE;
25+
}
26+
27+
@Override
28+
public UpdateControl<RetryTestCustomResource> createOrUpdateResource(
29+
RetryTestCustomResource resource, Context<RetryTestCustomResource> context) {
30+
numberOfExecutions.addAndGet(1);
31+
32+
if (!resource.getMetadata().getFinalizers().contains(FINALIZER_NAME)) {
33+
throw new IllegalStateException("Finalizer is not present.");
34+
}
35+
log.info("Value: " + resource.getSpec().getValue());
36+
37+
if (numberOfExecutions.get() < NUMBER_FAILED_EXECUTIONS + 1) {
38+
throw new RuntimeException("Testing Retry");
39+
}
40+
if (context.getRetryInfo().isEmpty() || context.getRetryInfo().get().isLastAttempt() == true) {
41+
throw new IllegalStateException("Not expected retry info: " + context.getRetryInfo());
42+
}
43+
44+
ensureStatusExists(resource);
45+
resource.getStatus().setState(RetryTestCustomResourceStatus.State.SUCCESS);
46+
47+
return UpdateControl.updateStatusSubResource(resource);
48+
}
49+
50+
private void ensureStatusExists(RetryTestCustomResource resource) {
51+
RetryTestCustomResourceStatus status = resource.getStatus();
52+
if (status == null) {
53+
status = new RetryTestCustomResourceStatus();
54+
resource.setStatus(status);
55+
}
56+
}
57+
58+
public int getNumberOfExecutions() {
59+
return numberOfExecutions.get();
60+
}
61+
}

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceSpec.java

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package io.javaoperatorsdk.operator.sample.retry;
2+
3+
public class RetryTestCustomResourceSpec {
4+
5+
private String value;
6+
7+
public String getValue() {
8+
return value;
9+
}
10+
11+
public RetryTestCustomResourceSpec setValue(String value) {
12+
this.value = value;
13+
return this;
14+
}
15+
}

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/retry/RetryTestCustomResourceStatus.java

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
package io.javaoperatorsdk.operator.sample.retry;
2+
3+
public class RetryTestCustomResourceStatus {
4+
5+
private State state;
6+
7+
public State getState() {
8+
return state;
9+
}
10+
11+
public RetryTestCustomResourceStatus setState(State state) {
12+
this.state = state;
13+
return this;
14+
}
15+
16+
public enum State {
17+
SUCCESS,
18+
ERROR
19+
}
20+
}

‎operator-framework/src/test/resources/io/javaoperatorsdk/operator/retry-test-crd.yaml

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
apiVersion: apiextensions.k8s.io/v1beta1
2+
kind: CustomResourceDefinition
3+
metadata:
4+
name: retrysamples.sample.javaoperatorsdk
5+
spec:
6+
group: sample.javaoperatorsdk
7+
version: v1
8+
subresources:
9+
status: {}
10+
scope: Namespaced
11+
names:
12+
plural: retrysamples
13+
singular: retrysample
14+
kind: retrysample
15+
shortNames:
16+
- rs

0 commit comments

Comments
 (0)
Please sign in to comment.