intro. SchedulerJobsHelper.executeAndAwaitJob() [FINERACT-922] (#817)
This commit is contained in:
+3
-31
@@ -20,8 +20,6 @@ package org.apache.fineract.integrationtests;
|
||||
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
|
||||
import io.restassured.builder.RequestSpecBuilder;
|
||||
import io.restassured.http.ContentType;
|
||||
@@ -35,7 +33,6 @@ import org.apache.fineract.integrationtests.common.SchedulerJobHelper;
|
||||
import org.apache.fineract.integrationtests.common.Utils;
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
public class SchedulerJobsTest {
|
||||
@@ -129,34 +126,9 @@ public class SchedulerJobsTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
@Ignore // TODO FINERACT-852 & FINERACT-922
|
||||
public void testSchedulerJobs() throws InterruptedException {
|
||||
// For each retrieved scheduled job (by ID)...
|
||||
for (Integer jobId : schedulerJobHelper.getAllSchedulerJobIds()) {
|
||||
// Retrieving Scheduler Job by ID
|
||||
Map<String, Object> schedulerJob = schedulerJobHelper.getSchedulerJobById(jobId);
|
||||
|
||||
// Executing Scheduler Job
|
||||
SchedulerJobHelper.runSchedulerJob(requestSpec, jobId.toString());
|
||||
|
||||
// Retrieving Scheduler Job by ID
|
||||
schedulerJob = schedulerJobHelper.getSchedulerJobById(jobId);
|
||||
assertNotNull(schedulerJob);
|
||||
|
||||
// Waiting for Job to complete
|
||||
while ((Boolean) schedulerJob.get("currentlyRunning") == true) {
|
||||
Thread.sleep(500);
|
||||
schedulerJob = schedulerJobHelper.getSchedulerJobById(jobId);
|
||||
assertNotNull(schedulerJob);
|
||||
}
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
List<Map> jobHistoryData = schedulerJobHelper.getSchedulerJobHistory(jobId);
|
||||
|
||||
// Verifying the Status of the Recently executed Scheduler Job
|
||||
assertFalse("Job History is empty :( Was it too slow? Failures in background job?",
|
||||
jobHistoryData.isEmpty());
|
||||
assertEquals("Verifying Last Scheduler Job Status", "success",
|
||||
jobHistoryData.get(jobHistoryData.size() - 1).get("status"));
|
||||
public void testTriggeringManualExecutionOfAllSchedulerJobs() {
|
||||
for (String jobName : schedulerJobHelper.getAllSchedulerJobNames()) {
|
||||
schedulerJobHelper.executeAndAwaitJob(jobName);
|
||||
}
|
||||
}
|
||||
}
|
||||
+133
-48
@@ -18,23 +18,32 @@
|
||||
*/
|
||||
package org.apache.fineract.integrationtests.common;
|
||||
|
||||
import static java.time.Instant.now;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static org.hamcrest.MatcherAssert.assertThat;
|
||||
import static org.hamcrest.core.Is.is;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
import io.restassured.builder.ResponseSpecBuilder;
|
||||
import io.restassured.specification.RequestSpecification;
|
||||
import io.restassured.specification.ResponseSpecification;
|
||||
import java.util.ArrayList;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.function.ToIntFunction;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
import org.junit.Assert;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
public class SchedulerJobHelper {
|
||||
private final static Logger LOG = LoggerFactory.getLogger(SchedulerJobHelper.class);
|
||||
private final RequestSpecification requestSpec;
|
||||
@@ -53,16 +62,24 @@ public class SchedulerJobHelper {
|
||||
this.response202Spec = responseSpec;
|
||||
}
|
||||
|
||||
private List getAllSchedulerJobs() {
|
||||
private List<Map<String, Object>> getAllSchedulerJobs() {
|
||||
final String GET_ALL_SCHEDULER_JOBS_URL = "/fineract-provider/api/v1/jobs?" + Utils.TENANT_IDENTIFIER;
|
||||
LOG.info("------------------------ RETRIEVING ALL SCHEDULER JOBS -------------------------");
|
||||
final ArrayList response = Utils.performServerGet(requestSpec, response200Spec, GET_ALL_SCHEDULER_JOBS_URL, "");
|
||||
List<Map<String, Object>> response = Utils.performServerGet(requestSpec, response200Spec, GET_ALL_SCHEDULER_JOBS_URL, "");
|
||||
assertNotNull(response);
|
||||
return response;
|
||||
}
|
||||
|
||||
private <T> List<T> getAllSchedulerJobDetails(Function<Map<String, Object>, T> mapper) {
|
||||
return getAllSchedulerJobs().stream().map(mapper).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
public List<Integer> getAllSchedulerJobIds() {
|
||||
ToIntFunction<Map> mapper = map -> (Integer) map.get("jobId");
|
||||
return getAllSchedulerJobs().stream().mapToInt(mapper).boxed().collect(Collectors.toList());
|
||||
return getAllSchedulerJobDetails(map -> (Integer) map.get("jobId"));
|
||||
}
|
||||
|
||||
public List<String> getAllSchedulerJobNames() {
|
||||
return getAllSchedulerJobDetails(map -> (String) map.get("displayName"));
|
||||
}
|
||||
|
||||
public Map<String, Object> getSchedulerJobById(int jobId) {
|
||||
@@ -103,14 +120,15 @@ public class SchedulerJobHelper {
|
||||
return new Gson().toJson(map);
|
||||
}
|
||||
|
||||
public List getSchedulerJobHistory(int jobId) {
|
||||
@SuppressWarnings("unchecked")
|
||||
private List<Map<String, Object>> getSchedulerJobHistory(int jobId) {
|
||||
final String GET_SCHEDULER_STATUS_URL = "/fineract-provider/api/v1/jobs/" + jobId + "/runhistory?" + Utils.TENANT_IDENTIFIER;
|
||||
LOG.info("------------------------ RETRIEVING SCHEDULER JOB HISTORY -------------------------");
|
||||
final Map response = Utils.performServerGet(requestSpec, response200Spec, GET_SCHEDULER_STATUS_URL, "");
|
||||
return (ArrayList) response.get("pageItems");
|
||||
final Map<String, Object> response = Utils.performServerGet(requestSpec, response200Spec, GET_SCHEDULER_STATUS_URL, "");
|
||||
return (List<Map<String, Object>>) response.get("pageItems");
|
||||
}
|
||||
|
||||
public static void runSchedulerJob(final RequestSpecification requestSpec, final String jobId) {
|
||||
private void runSchedulerJob(int jobId) {
|
||||
final ResponseSpecification responseSpec = new ResponseSpecBuilder().expectStatusCode(202).build();
|
||||
final String RUN_SCHEDULER_JOB_URL = "/fineract-provider/api/v1/jobs/" + jobId + "?command=executeJob&" + Utils.TENANT_IDENTIFIER;
|
||||
LOG.info("------------------------ RUN SCHEDULER JOB -------------------------");
|
||||
@@ -124,45 +142,112 @@ public class SchedulerJobHelper {
|
||||
return runSchedulerJob;
|
||||
}
|
||||
|
||||
public void executeJob(String jobName) throws InterruptedException {
|
||||
List<Map> allSchedulerJobsData = getAllSchedulerJobs();
|
||||
Assert.assertNotNull(allSchedulerJobsData);
|
||||
|
||||
private int getSchedulerJobIdByName(String jobName) {
|
||||
List<Map<String, Object>> allSchedulerJobsData = getAllSchedulerJobs();
|
||||
for (Integer jobIndex = 0; jobIndex < allSchedulerJobsData.size(); jobIndex++) {
|
||||
if (allSchedulerJobsData.get(jobIndex).get("displayName").equals(jobName)) {
|
||||
Integer jobId = (Integer) allSchedulerJobsData.get(jobIndex).get("jobId");
|
||||
|
||||
// Executing Scheduler Job
|
||||
runSchedulerJob(this.requestSpec, jobId.toString());
|
||||
|
||||
// Retrieving Scheduler Job by ID
|
||||
Map schedulerJob = getSchedulerJobById(jobId);
|
||||
Assert.assertNotNull(schedulerJob);
|
||||
|
||||
// Waiting for Job to complete
|
||||
while ((Boolean) schedulerJob.get("currentlyRunning") == true) {
|
||||
Thread.sleep(15000);
|
||||
schedulerJob = getSchedulerJobById(jobId);
|
||||
Assert.assertNotNull(schedulerJob);
|
||||
LOG.info("Job is Still Running");
|
||||
}
|
||||
|
||||
List<Map> jobHistoryData = getSchedulerJobHistory(jobId);
|
||||
|
||||
Assert.assertFalse("Job History is empty :( Was it too slow? Failures in background job?", jobHistoryData.isEmpty());
|
||||
|
||||
// print error associated with recent job failure (if any)
|
||||
LOG.info("Job run error message (printed only if the job fails: {}"
|
||||
, jobHistoryData.get(jobHistoryData.size() - 1).get("jobRunErrorMessage"));
|
||||
LOG.info("Job failure error log (printed only if the job fails: {}"
|
||||
, jobHistoryData.get(jobHistoryData.size() - 1).get("jobRunErrorLog"));
|
||||
|
||||
// Verifying the Status of the Recently executed Scheduler Job
|
||||
Assert.assertEquals("Verifying Last Scheduler Job Status", "success",
|
||||
jobHistoryData.get(jobHistoryData.size() - 1).get("status"));
|
||||
|
||||
break;
|
||||
return (Integer) allSchedulerJobsData.get(jobIndex).get("jobId");
|
||||
}
|
||||
}
|
||||
throw new IllegalArgumentException("No such named Job (see org.apache.fineract.infrastructure.jobs.service.JobName enum):" + jobName);
|
||||
}
|
||||
|
||||
// FINERACT-922 TODO Gradually replace use of this method with new executeAndAwaitJob() below, if it proves to be more stable than this one
|
||||
public void executeJob(String jobName) throws InterruptedException {
|
||||
// Stop the Scheduler while we manually trigger execution of job, to avoid side effects and simplify debugging when readings logs
|
||||
updateSchedulerStatus(false);
|
||||
|
||||
int jobId = getSchedulerJobIdByName(jobName);
|
||||
|
||||
// Executing Scheduler Job
|
||||
runSchedulerJob(jobId);
|
||||
|
||||
// Retrieving Scheduler Job by ID
|
||||
Map<String, Object> schedulerJob = getSchedulerJobById(jobId);
|
||||
|
||||
// Waiting for Job to complete
|
||||
while ((Boolean) schedulerJob.get("currentlyRunning") == true) {
|
||||
Thread.sleep(15000);
|
||||
schedulerJob = getSchedulerJobById(jobId);
|
||||
assertNotNull(schedulerJob);
|
||||
LOG.info("Job is Still Running");
|
||||
}
|
||||
|
||||
List<Map<String, Object>> jobHistoryData = getSchedulerJobHistory(jobId);
|
||||
|
||||
// print error associated with recent job failure (if any)
|
||||
LOG.error("Last Job run error message (only relevent if the job fails: {}",
|
||||
jobHistoryData.get(jobHistoryData.size() - 1).get("jobRunErrorMessage"));
|
||||
LOG.error("Lsat Job failure error log (only relevent if the job fails: {}",
|
||||
jobHistoryData.get(jobHistoryData.size() - 1).get("jobRunErrorLog"));
|
||||
|
||||
assertFalse("Job History is empty :( Was it too slow? Failures in background job?", jobHistoryData.isEmpty());
|
||||
|
||||
// Verifying the Status of the Recently executed Scheduler Job
|
||||
assertEquals("Verifying Last Scheduler Job Status", "success",
|
||||
jobHistoryData.get(jobHistoryData.size() - 1).get("status"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Launches a Job and awaits its completion.
|
||||
* @param jobName displayName (see {@link org.apache.fineract.infrastructure.jobs.service.JobName}) of Scheduler Job
|
||||
*
|
||||
* @author Michael Vorburger.ch
|
||||
*/
|
||||
public void executeAndAwaitJob(String jobName) {
|
||||
Duration TIMEOUT = Duration.ofSeconds(30);
|
||||
Duration PAUSE = Duration.ofMillis(500);
|
||||
DateTimeFormatter df = DateTimeFormatter.ISO_INSTANT; // FINERACT-926
|
||||
Instant beforeExecuteTime = now().truncatedTo(ChronoUnit.SECONDS);
|
||||
|
||||
// Stop the Scheduler while we manually trigger execution of job, to avoid side effects and simplify debugging when readings logs
|
||||
updateSchedulerStatus(false);
|
||||
|
||||
// Executing Scheduler Job
|
||||
int jobId = getSchedulerJobIdByName(jobName);
|
||||
runSchedulerJob(jobId);
|
||||
|
||||
// Await JobDetailData.lastRunHistory [JobDetailHistoryData] jobRunStartTime >= beforeExecuteTime (or timeout)
|
||||
await().atMost(TIMEOUT).pollInterval(PAUSE).until(jobLastRunHistorySupplier(jobId), lastRunHistory -> {
|
||||
String jobRunStartText = lastRunHistory.get("jobRunStartTime");
|
||||
if (jobRunStartText == null) {
|
||||
return false;
|
||||
}
|
||||
Instant jobRunStartTime = df.parse(jobRunStartText, Instant::from);
|
||||
return jobRunStartTime.equals(beforeExecuteTime) || jobRunStartTime.isAfter(beforeExecuteTime);
|
||||
});
|
||||
|
||||
// Await JobDetailData.lastRunHistory [JobDetailHistoryData] jobRunEndTime to be both set and >= jobRunStartTime (or timeout)
|
||||
Map<String, String> finalLastRunHistory = await().atMost(TIMEOUT).pollInterval(PAUSE).until(jobLastRunHistorySupplier(jobId), lastRunHistory -> {
|
||||
String jobRunEndText = lastRunHistory.get("jobRunEndTime");
|
||||
if (jobRunEndText == null) {
|
||||
return false;
|
||||
}
|
||||
Instant jobRunEndTime = df.parse(jobRunEndText, Instant::from);
|
||||
Instant jobRunStartTime = df.parse(lastRunHistory.get("jobRunStartTime"), Instant::from);
|
||||
return jobRunEndTime.equals(jobRunStartTime) || jobRunEndTime.isAfter(jobRunStartTime);
|
||||
});
|
||||
|
||||
// Verify triggerType
|
||||
assertThat(finalLastRunHistory.get("triggerType"), is("application"));
|
||||
|
||||
// Verify status & propagate jobRunErrorMessage and/or jobRunErrorLog (if any)
|
||||
String status = finalLastRunHistory.get("status");
|
||||
if (!status.equals("success")) {
|
||||
fail("Job status is not success: " + finalLastRunHistory.toString());
|
||||
}
|
||||
|
||||
// PS: Checking getSchedulerJobHistory() [/runhistory] is pointless, because the lastRunHistory JobDetailHistoryData is already part of JobDetailData anyway.
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private Callable<Map<String, String>> jobLastRunHistorySupplier(int jobId) {
|
||||
return () -> {
|
||||
Map<String, Object> job = getSchedulerJobById(jobId);
|
||||
if (job == null) {
|
||||
return null;
|
||||
}
|
||||
return (Map<String, String>) job.get("lastRunHistory");
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user