22
33
44import com .github .containersolutions .operator .processing .retry .Retry ;
5- import com .google .common .util .concurrent .ThreadFactoryBuilder ;
65import io .fabric8 .kubernetes .client .CustomResource ;
76import io .fabric8 .kubernetes .client .KubernetesClientException ;
87import io .fabric8 .kubernetes .client .Watcher ;
1110
1211import java .util .Optional ;
1312import java .util .concurrent .ScheduledThreadPoolExecutor ;
14- import java .util .concurrent .ThreadFactory ;
1513import java .util .concurrent .TimeUnit ;
1614import java .util .concurrent .locks .ReentrantLock ;
1715
@@ -48,11 +46,7 @@ public class EventScheduler implements Watcher<CustomResource> {
4846 public EventScheduler (EventDispatcher eventDispatcher , Retry retry ) {
4947 this .eventDispatcher = eventDispatcher ;
5048 this .retry = retry ;
51- ThreadFactory threadFactory = new ThreadFactoryBuilder ()
52- .setNameFormat ("event-consumer-%d" )
53- .setDaemon (false )
54- .build ();
55- executor = new ScheduledThreadPoolExecutor (1 , threadFactory );
49+ executor = new ScheduledThreadPoolExecutor (1 );
5650 executor .setRemoveOnCancelPolicy (true );
5751 }
5852
@@ -61,14 +55,13 @@ public void eventReceived(Watcher.Action action, CustomResource resource) {
6155 log .debug ("Event received for action: {}, {}: {}" , action .toString ().toLowerCase (), resource .getClass ().getSimpleName (),
6256 resource .getMetadata ().getName ());
6357 CustomResourceEvent event = new CustomResourceEvent (action , resource , retry );
64- scheduleEvent (event );
58+ scheduleEventFromApi (event );
6559 }
6660
67- void scheduleEvent (CustomResourceEvent event ) {
68- log .trace ("Current queue size {}" , executor .getQueue ().size ());
69- log .debug ("Scheduling event: {}" , event );
61+ void scheduleEventFromApi (CustomResourceEvent event ) {
7062 try {
7163 lock .lock ();
64+ log .debug ("Scheduling event from Api: {}" , event );
7265 if (event .getResource ().getMetadata ().getDeletionTimestamp () != null && event .getAction () == Action .DELETED ) {
7366 // Note that we always use finalizers, we want to process delete event just in corner case,
7467 // when we are not able to add finalizer (lets say because of optimistic locking error, and the resource was deleted instantly).
@@ -89,30 +82,39 @@ void scheduleEvent(CustomResourceEvent event) {
8982 eventStore .addOrReplaceEventAsNotScheduled (event );
9083 return ;
9184 }
85+ scheduleEventForExecution (event );
86+ log .trace ("Scheduling event from API finished: {}" , event );
87+ } finally {
88+ lock .unlock ();
89+ }
90+ }
9291
92+ private void scheduleEventForExecution (CustomResourceEvent event ) {
93+ try {
94+ lock .lock ();
95+ log .trace ("Current queue size {}" , executor .getQueue ().size ());
96+ log .debug ("Scheduling event for execution: {}" , event );
9397 Optional <Long > nextBackOff = event .nextBackOff ();
9498 if (!nextBackOff .isPresent ()) {
9599 log .warn ("Event max retry limit reached. Will be discarded. {}" , event );
96100 return ;
97101 }
98- log .debug ("Creating scheduled task for event: {}" , event );
99102 eventStore .addEventUnderProcessing (event );
100103 executor .schedule (new EventConsumer (event , eventDispatcher , this ),
101104 nextBackOff .get (), TimeUnit .MILLISECONDS );
105+ log .trace ("Scheduled task for event: {}" , event );
102106 } finally {
103- log .debug ("Scheduling event finished: {}" , event );
104107 lock .unlock ();
105108 }
106109 }
107110
108111 void eventProcessingFinishedSuccessfully (CustomResourceEvent event ) {
109112 try {
110113 lock .lock ();
111- log .debug ("Event processing successful for event: {}" , event );
112114 eventStore .removeEventUnderProcessing (event .resourceUid ());
113115 if (eventStore .containsNotScheduledEvent (event .resourceUid ())) {
114- log .debug ("Scheduling recent event for processing processing : {}" , event );
115- scheduleEvent ( eventStore . removeEventNotScheduled ( event .resourceUid () ));
116+ log .debug ("Scheduling recent event for processing: {}" , event );
117+ scheduleNotYetScheduledEventForExecution ( event .resourceUid ());
116118 }
117119 } finally {
118120 lock .unlock ();
@@ -124,19 +126,22 @@ void eventProcessingFailed(CustomResourceEvent event) {
124126 lock .lock ();
125127 eventStore .removeEventUnderProcessing (event .resourceUid ());
126128 if (eventStore .containsNotScheduledEvent (event .resourceUid ())) {
127- CustomResourceEvent notScheduledEvent = eventStore .removeEventNotScheduled (event .resourceUid ());
128- log .debug ("Event processing failed. Scheduling the most recent event. Failed event: {}," +
129- " Most recent event: {}" , event , notScheduledEvent );
130- scheduleEvent (notScheduledEvent );
129+ log .debug ("Event processing failed. Scheduling the most recent event. Failed event: {}" , event );
130+ scheduleNotYetScheduledEventForExecution (event .resourceUid ());
131131 } else {
132132 log .debug ("Event processing failed. Attempting to re-schedule the event: {}" , event );
133- scheduleEvent (event );
133+ scheduleEventForExecution (event );
134134 }
135135 } finally {
136136 lock .unlock ();
137137 }
138138 }
139139
140+ private void scheduleNotYetScheduledEventForExecution (String uuid ) {
141+ CustomResourceEvent notScheduledEvent = eventStore .removeEventNotScheduled (uuid );
142+ scheduleEventForExecution (notScheduledEvent );
143+ }
144+
140145 @ Override
141146 public void onClose (KubernetesClientException e ) {
142147 log .error ("Error: " , e );
0 commit comments