2929import java .util .Optional ;
3030import java .util .concurrent .TimeUnit ;
3131
32-
3332@ Service
3433public class FlowJobService extends ResourceManager {
3534 private static final String ML_JOBS_NAME = "ml:jobs" ;
@@ -46,7 +45,7 @@ public class FlowJobService extends ResourceManager {
4645
4746 //use a cache with ttl 5 minutes to reduce frequency for fetching data from DB
4847 public final Cache <String , FlowJobs > cachedJobsByFlowName = CacheBuilder .newBuilder ().expireAfterWrite (
49- 5 , TimeUnit .MINUTES ).build ();
48+ 5 , TimeUnit .SECONDS ).build ();
5049
5150 public FlowJobService () {
5251 super ();
@@ -56,127 +55,132 @@ public FlowJobService() {
5655 }
5756
5857 private void setupClient () {
59- this . client = hubConfig .newJobDbClient ();
58+ client = hubConfig .newJobDbClient ();
6059 }
6160
62- public FlowJobs getJobs (String flowName ) {
63- if (this .client == null ) {
64- this .setupClient ();
61+ public FlowJobs getJobs (String flowName , boolean forceRefresh ) {
62+ if (forceRefresh ) {
63+ FlowJobs flowJobs = retrieveJobsByFlowName (flowName );
64+ cachedJobsByFlowName .put (flowName , flowJobs );
65+ return flowJobs ;
6566 }
6667 try {
67- return cachedJobsByFlowName .get (flowName , () -> {
68- client .init (ML_JOBS_NAME , this );
69- RequestParameters params = new RequestParameters ();
70- if (StringUtils .isNotEmpty (flowName )) {
71- params .add ("flow-name" , flowName );
72- }
68+ return cachedJobsByFlowName .get (flowName , () -> retrieveJobsByFlowName (flowName ));
69+ } catch (Exception e ) {
70+ logger .error (e .getMessage ());
71+ } finally {
72+ if (cachedJobsByFlowName .getIfPresent (flowName ) == null ) {
73+ cachedJobsByFlowName .invalidate (flowName );
74+ }
75+ }
7376
74- ServiceResultIterator resultItr = this .getServices ().get (params );
75- if (resultItr == null || !resultItr .hasNext ()) {
76- throw new RuntimeException ("Unable to get job document" );
77- }
78- ServiceResult res = resultItr .next ();
79- JsonNode jsonNode = res .getContent (new JacksonHandle ()).get ();
80-
81- Flow flow = flowManager .getFlow (flowName );
82- Map <String , Step > steps = flow .getSteps ();
83-
84- List <String > jobs = new ArrayList <>();
85- LatestJob latestJob = new LatestJob ();
86- FlowJobs flowJobs = new FlowJobs (jobs , latestJob );
87-
88- final JsonNode [] lastJob = {null };
89- final String [] lastTime = {null , null }; //store last start and end time
90- if (jsonNode .isArray ()) {
91- jsonNode .forEach (job -> {
92- JSONObject jobJson = new JSONObject (job .get ("job" ));
93- String jobId = jobJson .getString ("jobId" );
94- jobs .add (jobId );
95- String currStartTime = jobJson .getString ("timeStarted" , "" );
96- String currEndTime = jobJson .getString ("timeEnded" , "" );
97- if (StringUtils .isEmpty (lastTime [0 ])) {
77+ return cachedJobsByFlowName .getIfPresent (flowName );
78+ }
79+
80+ private FlowJobs retrieveJobsByFlowName (String flowName ) {
81+ if (client == null ) {
82+ setupClient ();
83+ }
84+ client .init (ML_JOBS_NAME , this );
85+ RequestParameters params = new RequestParameters ();
86+ if (StringUtils .isNotEmpty (flowName )) {
87+ params .add ("flow-name" , flowName );
88+ }
89+
90+ ServiceResultIterator resultItr = this .getServices ().get (params );
91+ if (resultItr == null || !resultItr .hasNext ()) {
92+ throw new RuntimeException ("No jobs found for flow with name: " + flowName );
93+ }
94+ ServiceResult res = resultItr .next ();
95+ JsonNode jsonNode = res .getContent (new JacksonHandle ()).get ();
96+
97+ Flow flow = flowManager .getFlow (flowName );
98+ Map <String , Step > steps = flow .getSteps ();
99+
100+ List <String > jobs = new ArrayList <>();
101+ LatestJob latestJob = new LatestJob ();
102+ FlowJobs flowJobs = new FlowJobs (jobs , latestJob );
103+
104+ final JsonNode [] lastJob = {null };
105+ final String [] lastTime = {null , null }; //store last start and end time
106+ if (jsonNode .isArray ()) {
107+ jsonNode .forEach (job -> {
108+ JSONObject jobJson = new JSONObject (job .get ("job" ));
109+ String jobId = jobJson .getString ("jobId" );
110+ jobs .add (jobId );
111+ String currStartTime = jobJson .getString ("timeStarted" , "" );
112+ String currEndTime = jobJson .getString ("timeEnded" , "" );
113+ if (StringUtils .isEmpty (lastTime [0 ])) {
114+ lastTime [0 ] = currStartTime ;
115+ lastTime [1 ] = currEndTime ;
116+ lastJob [0 ] = jobJson .jsonNode ();
117+ } else {
118+ try {
119+ int cmp = DatatypeConverter .parseDateTime (lastTime [0 ]).getTime ().compareTo (DatatypeConverter .parseDateTime (currStartTime ).getTime ());
120+ if (cmp < 0 ) {
98121 lastTime [0 ] = currStartTime ;
99122 lastTime [1 ] = currEndTime ;
123+ //lastJobId[0] = jobId;
100124 lastJob [0 ] = jobJson .jsonNode ();
101- }
102- else {
103- try {
104- int cmp = DatatypeConverter .parseDateTime (lastTime [0 ]).getTime ().compareTo (DatatypeConverter .parseDateTime (currStartTime ).getTime ());
105- if (cmp < 0 ) {
106- lastTime [0 ] = currStartTime ;
107- lastTime [1 ] = currEndTime ;
108- //lastJobId[0] = jobId;
109- lastJob [0 ] = jobJson .jsonNode ();
110- }
111- else if (cmp == 0 ) { //compare end time just in case
112- if (DatatypeConverter .parseDateTime (lastTime [1 ]).getTime ().compareTo (DatatypeConverter .parseDateTime (currEndTime ).getTime ()) < 0 ) {
113- lastTime [0 ] = currStartTime ;
114- lastTime [1 ] = currEndTime ;
115- lastJob [0 ] = jobJson .jsonNode ();
116- }
117- }
118- }
119- catch (Exception e ) {
120- //ignore, maybe log
125+ } else if (cmp == 0 ) { //compare end time just in case
126+ if (DatatypeConverter .parseDateTime (lastTime [1 ]).getTime ().compareTo (DatatypeConverter .parseDateTime (currEndTime ).getTime ()) < 0 ) {
127+ lastTime [0 ] = currStartTime ;
128+ lastTime [1 ] = currEndTime ;
129+ lastJob [0 ] = jobJson .jsonNode ();
121130 }
122131 }
123- });
124- }
125-
126- if (jobs .isEmpty () || lastJob [0 ] == null ) {
127- return flowJobs ;
132+ } catch (Exception e ) {
133+ logger .error (e .getMessage ());
134+ }
128135 }
129- JSONObject jobJson = new JSONObject (lastJob [0 ]);
130-
131- latestJob .id = jobJson .getString ("jobId" );
132- latestJob .startTime = jobJson .getString ("timeStarted" , "" );
133- latestJob .endTime = jobJson .getString ("timeEnded" , "" );
134- latestJob .status = jobJson .getString ("jobStatus" );
135-
136- String completedKey = jobJson .getString ("lastCompletedStep" , "0" );
137- String attemptedKey = jobJson .getString ("lastAttemptedStep" , "0" );
138- String stepKey = Integer .compare (Integer .valueOf (completedKey ), Integer .valueOf (attemptedKey )) < 0 ? attemptedKey : completedKey ;
139- Optional .ofNullable (steps .get (stepKey )).ifPresent (s -> {
140- latestJob .stepName = s .getName ();
141- latestJob .stepId = s .getName () + "-" + s .getStepDefinitionType ();
142- });
143-
144- JsonNode stepRes = jobJson .getNode ("stepResponses" );
145-
146- if (stepRes != null ) {
147- stepRes .forEach (s -> {
148- latestJob .successfulEvents += s .get ("successfulEvents" ).asLong ();
149- latestJob .failedEvents += s .get ("failedEvents" ).asLong ();
150- latestJob .output = s .get ("stepOutput" ); //last step output ?
151- });
152- }
153-
154- return flowJobs ;
155136 });
156137 }
157- catch (Exception e ) {
158138
139+ if (jobs .isEmpty () || lastJob [0 ] == null ) {
140+ return flowJobs ;
159141 }
160- finally {
161- if (cachedJobsByFlowName .getIfPresent (flowName ) == null ) {
162- cachedJobsByFlowName .invalidate (flowName );
163- }
164- this .release ();
142+ JSONObject jobJson = new JSONObject (lastJob [0 ]);
143+
144+ latestJob .id = jobJson .getString ("jobId" );
145+ latestJob .startTime = jobJson .getString ("timeStarted" , "" );
146+ latestJob .endTime = jobJson .getString ("timeEnded" , "" );
147+ if ("N/A" .equals (latestJob .endTime )) {
148+ latestJob .endTime = "" ;
165149 }
166- return cachedJobsByFlowName .getIfPresent (flowName );
150+ latestJob .status = jobJson .getString ("jobStatus" );
151+
152+ String completedKey = jobJson .getString ("lastCompletedStep" , "0" );
153+ String attemptedKey = jobJson .getString ("lastAttemptedStep" , "0" );
154+ String stepKey = Integer .compare (Integer .valueOf (completedKey ), Integer .valueOf (attemptedKey )) < 0 ? attemptedKey : completedKey ;
155+ Optional .ofNullable (steps .get (stepKey )).ifPresent (s -> {
156+ latestJob .stepName = s .getName ();
157+ latestJob .stepId = s .getName () + "-" + s .getStepDefinitionType ();
158+ });
159+
160+ JsonNode stepRes = jobJson .getNode ("stepResponses" );
161+
162+ if (stepRes != null ) {
163+ stepRes .forEach (s -> {
164+ latestJob .successfulEvents += s .get ("successfulEvents" ) != null ? s .get ("successfulEvents" ).asLong () : 0 ;
165+ latestJob .failedEvents += s .get ("failedEvents" ) != null ? s .get ("failedEvents" ).asLong () : 0 ;
166+ latestJob .output = s .get ("stepOutput" ); //last step output ?
167+ });
168+ }
169+ this .release ();
170+
171+ return flowJobs ;
167172 }
168173
169- //TODO: fix this whole client mess and properly report exceptions
170174 private void release () {
171- if (this . client != null ) {
175+ if (client != null ) {
172176 try {
173- this . client .release ();
177+ client .release ();
174178 }
175179 catch (Exception e ) {
176180 logger .error (e .getMessage ());
177181 }
178182 finally {
179- this . client = null ;
183+ client = null ;
180184 }
181185 }
182186 }
0 commit comments