![[Jenkins]](/kapua/static/dff81f37/images/svgs/logo.svg){kind=logo}
Package: KapuaJobListener
KapuaJobListener
| name | instruction | branch | complexity | line | method | ||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| KapuaJobListener() |
|
|
|
|
|
||||||||||||||||||||
| afterJob() |
|
|
|
|
|
||||||||||||||||||||
| beforeJob() |
|
|
|
|
|
||||||||||||||||||||
| checkQueuedJobExecutions(KapuaId, KapuaId, KapuaId) |
|
|
|
|
|
||||||||||||||||||||
| createJobExecution(KapuaId, KapuaId, JobTargetSublist, Long) |
|
|
|
|
|
||||||||||||||||||||
| enqueueJobExecution(KapuaId, KapuaId, KapuaId, KapuaId) |
|
|
|
|
|
||||||||||||||||||||
| getAnotherJobExecutionRunning(KapuaId, KapuaId, KapuaId, String, Set) |
|
|
|
|
|
||||||||||||||||||||
| lambda$afterJob$5(JobContextWrapper, KapuaId) |
|
|
|
|
|
||||||||||||||||||||
| lambda$afterJob$6(JobExecution) |
|
|
|
|
|
||||||||||||||||||||
| lambda$beforeJob$0(JobLogger, JobStepProperty) |
|
|
|
|
|
||||||||||||||||||||
| lambda$beforeJob$1(JobContextWrapper) |
|
|
|
|
|
||||||||||||||||||||
| lambda$beforeJob$2(JobExecution) |
|
|
|
|
|
||||||||||||||||||||
| lambda$beforeJob$3(JobExecution, KapuaId) |
|
|
|
|
|
||||||||||||||||||||
| lambda$beforeJob$4(JobTarget) |
|
|
|
|
|
||||||||||||||||||||
| lambda$createJobExecution$7(JobTargetQuery) |
|
|
|
|
|
||||||||||||||||||||
| lambda$createJobExecution$8(Set, JobTarget) |
|
|
|
|
|
||||||||||||||||||||
| lambda$createJobExecution$9(JobExecutionCreator) |
|
|
|
|
|
||||||||||||||||||||
| lambda$enqueueJobExecution$11(QueuedJobExecutionCreator) |
|
|
|
|
|
||||||||||||||||||||
| lambda$getAnotherJobExecutionRunning$10(JobExecutionQuery) |
|
|
|
|
|
||||||||||||||||||||
| static {...} |
|
|
|
|
|
||||||||||||||||||||
Coverage
1: /*******************************************************************************
2: * Copyright (c) 2017, 2022 Eurotech and/or its affiliates and others
3: *
4: * This program and the accompanying materials are made
5: * available under the terms of the Eclipse Public License 2.0
6: * which is available at https://www.eclipse.org/legal/epl-2.0/
7: *
8: * SPDX-License-Identifier: EPL-2.0
9: *
10: * Contributors:
11: * Eurotech - initial API and implementation
12: *******************************************************************************/
13: package org.eclipse.kapua.job.engine.jbatch.listener;
14:
15: import org.eclipse.kapua.KapuaException;
16: import org.eclipse.kapua.model.query.SortOrder;
17: import org.eclipse.kapua.commons.security.KapuaSecurityUtils;
18: import org.eclipse.kapua.job.engine.JobStartOptions;
19: import org.eclipse.kapua.job.engine.commons.logger.JobLogger;
20: import org.eclipse.kapua.job.engine.commons.model.JobTargetSublist;
21: import org.eclipse.kapua.job.engine.commons.wrappers.JobContextWrapper;
22: import org.eclipse.kapua.job.engine.jbatch.driver.JbatchDriver;
23: import org.eclipse.kapua.job.engine.exception.JobAlreadyRunningException;
24: import org.eclipse.kapua.job.engine.exception.JobExecutionEnqueuedException;
25: import org.eclipse.kapua.job.engine.jbatch.setting.JobEngineSetting;
26: import org.eclipse.kapua.job.engine.jbatch.setting.JobEngineSettingKeys;
27: import org.eclipse.kapua.job.engine.queue.QueuedJobExecution;
28: import org.eclipse.kapua.job.engine.queue.QueuedJobExecutionCreator;
29: import org.eclipse.kapua.job.engine.queue.QueuedJobExecutionFactory;
30: import org.eclipse.kapua.job.engine.queue.QueuedJobExecutionService;
31: import org.eclipse.kapua.job.engine.queue.QueuedJobExecutionStatus;
32: import org.eclipse.kapua.locator.KapuaLocator;
33: import org.eclipse.kapua.model.id.KapuaId;
34: import org.eclipse.kapua.model.query.predicate.AttributePredicate;
35: import org.eclipse.kapua.service.job.Job;
36: import org.eclipse.kapua.service.job.execution.JobExecution;
37: import org.eclipse.kapua.service.job.execution.JobExecutionAttributes;
38: import org.eclipse.kapua.service.job.execution.JobExecutionCreator;
39: import org.eclipse.kapua.service.job.execution.JobExecutionFactory;
40: import org.eclipse.kapua.service.job.execution.JobExecutionListResult;
41: import org.eclipse.kapua.service.job.execution.JobExecutionQuery;
42: import org.eclipse.kapua.service.job.execution.JobExecutionService;
43: import org.eclipse.kapua.service.job.targets.JobTarget;
44: import org.eclipse.kapua.service.job.targets.JobTargetAttributes;
45: import org.eclipse.kapua.service.job.targets.JobTargetFactory;
46: import org.eclipse.kapua.service.job.targets.JobTargetListResult;
47: import org.eclipse.kapua.service.job.targets.JobTargetQuery;
48: import org.eclipse.kapua.service.job.targets.JobTargetService;
49: import org.eclipse.kapua.service.job.targets.JobTargetStatus;
50: import org.slf4j.Logger;
51: import org.slf4j.LoggerFactory;
52:
53: import javax.batch.api.listener.AbstractJobListener;
54: import javax.batch.api.listener.JobListener;
55: import javax.batch.runtime.BatchRuntime;
56: import javax.batch.runtime.context.JobContext;
57: import javax.inject.Inject;
58: import java.util.Date;
59: import java.util.HashSet;
60: import java.util.List;
61: import java.util.Properties;
62: import java.util.Set;
63: import java.util.Timer;
64:
65: /**
66: * {@link JobListener} implementations.
67: * <p>
68: * Listener for all {@link Job}.
69: *
70: * @since 1.0.0
71: */
72: public class KapuaJobListener extends AbstractJobListener implements JobListener {
73:
74: private static final Logger LOG = LoggerFactory.getLogger(KapuaJobListener.class);
75:
76: private static final String JBATCH_EXECUTION_ID = "JBATCH_EXECUTION_ID";
77:
78: private static final JobEngineSetting JOB_ENGINE_SETTING = JobEngineSetting.getInstance();
79:
80: private static final KapuaLocator LOCATOR = KapuaLocator.getInstance();
81:
82: private static final JobExecutionService JOB_EXECUTION_SERVICE = LOCATOR.getService(JobExecutionService.class);
83: private static final JobExecutionFactory JOB_EXECUTION_FACTORY = LOCATOR.getFactory(JobExecutionFactory.class);
84:
85: private static final JobTargetService JOB_TARGET_SERVICE = LOCATOR.getService(JobTargetService.class);
86: private static final JobTargetFactory JOB_TARGET_FACTORY = LOCATOR.getFactory(JobTargetFactory.class);
87:
88: private static final QueuedJobExecutionService QUEUED_JOB_SERVICE = LOCATOR.getService(QueuedJobExecutionService.class);
89: private static final QueuedJobExecutionFactory QUEUED_JOB_FACTORY = LOCATOR.getFactory(QueuedJobExecutionFactory.class);
90:
91: @Inject
92: private JobContext jobContext;
93:
94: /**
95: * Before starting the actual {@link org.eclipse.kapua.service.job.Job} processing, create the {@link JobExecution} to track progress and
96: * check if there are other {@link JobExecution}s running with the same {@link JobExecution#getTargetIds()}.
97: * <p>
98: * If there are {@link JobExecution} running with matching {@link JobTargetSublist} the current {@link JobExecution} is stopped.
99: * According to the {@link JobStartOptions#getEnqueue()} parameter, the {@link JobExecution} can be:
100: * <ul>
101: * <li>if ({@code enqueue} = {@code false}) then {@link JobAlreadyRunningException} is {@code throw}n </li>
102: * <li>if ({@code enqueue} = {@code true}) then a new {@link QueuedJobExecution} is created and {@link JobExecutionEnqueuedException} is {@code throw}n</li>
103: * </ul>
104: */
105: @Override
106: public void beforeJob() throws Exception {
107: JobContextWrapper jobContextWrapper = new JobContextWrapper(jobContext);
108:
109: JobLogger jobLogger = jobContextWrapper.getJobLogger();
110: jobLogger.setClassLog(LOG);
111:
112: jobLogger.info("Running before job...");
113: jobLogger.info("Run configuration:");
114:• jobLogger.info("\tTarget count: {}", jobContextWrapper.getTargetSublist().size() != 0 ? jobContextWrapper.getTargetSublist().size() : "all");
115: jobLogger.info("\tReset step index: {}", jobContextWrapper.getResetStepIndex());
116:• jobLogger.info("\tFrom step index: {}", jobContextWrapper.getFromStepIndex() != null ? jobContextWrapper.getFromStepIndex() : 0);
117:• jobLogger.info("\tResuming job execution: {}", jobContextWrapper.getResumedJobExecutionId() != null ? jobContextWrapper.getResumedJobExecutionId() : "none");
118: jobLogger.info("\tEnqueue: {}", jobContextWrapper.getEnqueue());
119:
120:• if (jobContextWrapper.getJobStepPropertiesOverrides().isEmpty()) {
121: jobLogger.info("\tStep Properties Overrides: none");
122: } else {
123: jobLogger.info("\tStep Properties Overrides: ");
124: jobContextWrapper
125: .getJobStepPropertiesOverrides()
126: .getJobStepPropertiesOverrides()
127: .forEach(jsp -> jobLogger.info("\t - {}: {}", jsp.getName(), jsp.getPropertyValue()));
128: }
129:
130: JobExecution jobExecution;
131:• if (jobContextWrapper.getResumedJobExecutionId() != null) {
132: jobLogger.info("Resuming job execution...");
133: try {
134: jobExecution = KapuaSecurityUtils.doPrivileged(() -> JOB_EXECUTION_SERVICE.find(jobContextWrapper.getScopeId(), jobContextWrapper.getResumedJobExecutionId()));
135: } catch (Exception e) {
136: jobLogger.error(e, "Resuming job execution... ERROR!");
137: throw e;
138: }
139: jobLogger.info("Resuming job execution... DONE!");
140: } else {
141: jobLogger.info("Creating job execution...");
142: try {
143: jobExecution = createJobExecution(
144: jobContextWrapper.getScopeId(),
145: jobContextWrapper.getJobId(),
146: jobContextWrapper.getTargetSublist(),
147: jobContextWrapper.getExecutionId());
148: } catch (Exception e) {
149: jobLogger.error(e, "Creating job execution... ERROR!");
150: throw e;
151: }
152: jobLogger.info("Creating job execution... DONE!");
153: }
154:
155: jobLogger.setJobExecutionId(jobExecution.getId());
156:
157: jobContextWrapper.setKapuaExecutionId(jobExecution.getId());
158:
159: JobExecution runningJobExecution = getAnotherJobExecutionRunning(
160: jobExecution.getScopeId(),
161: jobExecution.getJobId(),
162: jobExecution.getId(),
163: jobContextWrapper.getJobName(),
164: jobExecution.getTargetIds());
165:
166:• if (runningJobExecution != null) {
167:
168:• if (jobContextWrapper.getEnqueue()) {
169: jobLogger.warn("Another execution is running! Enqueuing this execution...");
170:
171: QueuedJobExecution queuedJobExecution;
172: try {
173: queuedJobExecution = enqueueJobExecution(
174: jobExecution.getScopeId(),
175: jobExecution.getJobId(),
176: jobExecution.getId(),
177: runningJobExecution.getId());
178: } catch (Exception e) {
179: jobLogger.error(e, "Another execution is running! Enqueuing this execution... ERROR!");
180: throw e;
181: }
182:
183: KapuaSecurityUtils.doPrivileged(() -> JbatchDriver.stopJob(jobExecution.getScopeId(), jobExecution.getJobId(), jobExecution.getId()));
184: jobLogger.warn("Another execution is running! Stopping and enqueuing this execution... DONE! EnqueuedJob id : {}", queuedJobExecution.getJobId());
185:
186: } else {
187: jobLogger.error("Another execution is running! Aborting this execution...");
188: throw new JobAlreadyRunningException(jobExecution.getScopeId(), jobExecution.getJobId(), jobExecution.getId(), jobExecution.getTargetIds());
189: }
190: }
191:
192:• if (jobContextWrapper.getResetStepIndex()) {
193:• int resetToStepIndex = jobContextWrapper.getFromStepIndex() != null ? jobContextWrapper.getFromStepIndex() : 0;
194:
195: try {
196: jobLogger.info("Resetting {} targets to step index: {}...", jobExecution.getTargetIds().size(), resetToStepIndex);
197:
198:• for (KapuaId jobTargetId : jobExecution.getTargetIds()) {
199: JobTarget jobTarget = KapuaSecurityUtils.doPrivileged(() -> JOB_TARGET_SERVICE.find(jobExecution.getScopeId(), jobTargetId));
200:
201: jobTarget.setStepIndex(resetToStepIndex);
202: jobTarget.setStatus(JobTargetStatus.PROCESS_AWAITING);
203: jobTarget.setStatusMessage(null);
204: jobTarget.setException(null);
205:
206: KapuaSecurityUtils.doPrivileged(() -> JOB_TARGET_SERVICE.update(jobTarget));
207: }
208: } catch (KapuaException e) {
209: jobLogger.error(e, "Resetting {} targets to step index: {}... ERROR!", jobExecution.getTargetIds().size(), resetToStepIndex);
210: }
211: jobLogger.info("Resetting {} targets to step index: {}... DONE!", jobExecution.getTargetIds().size(), resetToStepIndex);
212: }
213:
214: jobLogger.info("Running before job... DONE!");
215: }
216:
217: /**
218: * Close the {@link JobExecution} setting the {@link JobExecution#getEndedOn()}.
219: */
220: @Override
221: public void afterJob() throws Exception {
222: JobContextWrapper jobContextWrapper = new JobContextWrapper(jobContext);
223: JobLogger jobLogger = jobContextWrapper.getJobLogger();
224:
225: jobLogger.info("Running after job...");
226:
227: KapuaId kapuaExecutionId = jobContextWrapper.getKapuaExecutionId();
228:• if (kapuaExecutionId == null) {
229: LOG.error("Cannot update job execution (internal reference [{}]). Cannot find 'executionId' in JobContext", jobContextWrapper.getExecutionId());
230: // Don't send any exception to prevent the job engine to set the job exit status as failed!
231: } else {
232: JobExecution jobExecution = KapuaSecurityUtils.doPrivileged(() -> JOB_EXECUTION_SERVICE.find(jobContextWrapper.getScopeId(), kapuaExecutionId));
233:
234:• if (jobExecution != null) {
235: jobExecution.setLog(jobLogger.flush());
236: jobExecution.setEndedOn(new Date());
237:
238: KapuaSecurityUtils.doPrivileged(() -> JOB_EXECUTION_SERVICE.update(jobExecution));
239:
240: checkQueuedJobExecutions(
241: jobContextWrapper.getScopeId(),
242: jobContextWrapper.getJobId(),
243: jobContextWrapper.getKapuaExecutionId());
244: } else {
245: LOG.warn("Cannot find job execution with id: {}. This is likely to happen with the Job has been forcibly deleted.", kapuaExecutionId);
246: }
247: }
248: jobLogger.info("Running after job... DONE!");
249: }
250:
251: /**
252: * Creates the {@link JobExecution} from the data in the {@link JobContextWrapper}.
253: * <p>
254: * If the {@link org.eclipse.kapua.service.job.Job} has started without a defined set of {@link JobStartOptions#getTargetIdSublist()}
255: * all {@link org.eclipse.kapua.service.job.targets.JobTarget}s will be added to the {@link JobExecution#getTargetIds()}.
256: *
257: * @param scopeId The {@link Job#getScopeId()}
258: * @param jobId The {@link Job#getId()}
259: * @param jobTargetSublist The {@link JobStartOptions#getTargetIdSublist()} of the targeted items
260: * @param jBatchExecutionId The {@link JobContext#getExecutionId()}
261: * @return The newly created {@link JobExecution}
262: * @throws KapuaException If any error happens during the processing
263: * @since 1.1.0
264: */
265: private JobExecution createJobExecution(KapuaId scopeId, KapuaId jobId, JobTargetSublist jobTargetSublist, Long jBatchExecutionId) throws KapuaException {
266:
267: Properties jobExecutionProperties = new Properties();
268: jobExecutionProperties.put(JBATCH_EXECUTION_ID, Long.toString(jBatchExecutionId));
269:
270: JobExecutionCreator jobExecutionCreator = JOB_EXECUTION_FACTORY.newCreator(scopeId);
271:
272: jobExecutionCreator.setJobId(jobId);
273: jobExecutionCreator.setStartedOn(new Date());
274: jobExecutionCreator.setEntityAttributes(jobExecutionProperties);
275:
276:• if (jobTargetSublist.isEmpty()) {
277: JobTargetQuery jobTargetQuery = JOB_TARGET_FACTORY.newQuery(scopeId);
278:
279: jobTargetQuery.setPredicate(
280: jobTargetQuery.andPredicate(
281: jobTargetQuery.attributePredicate(JobTargetAttributes.JOB_ID, jobId)
282: )
283: );
284:
285: JobTargetListResult jobTargets = KapuaSecurityUtils.doPrivileged(() -> JOB_TARGET_SERVICE.query(jobTargetQuery));
286:
287: Set<KapuaId> targetIds = new HashSet<>();
288: jobTargets.getItems().forEach(jt -> targetIds.add(jt.getId()));
289:
290: jobExecutionCreator.setTargetIds(new HashSet<>(targetIds));
291: } else {
292: jobExecutionCreator.setTargetIds(jobTargetSublist.getTargetIds());
293: }
294:
295: return KapuaSecurityUtils.doPrivileged(() -> JOB_EXECUTION_SERVICE.create(jobExecutionCreator));
296: }
297:
298: /**
299: * Checks if there are other {@link JobExecution}s running in this moment.
300: * <p>
301: * First it checks for an execution running from the {@link BatchRuntime}.
302: * This will return the jBatch execution ids that are currently active.
303: * <p>
304: * If 0, no {@link JobExecution} is currently running, returns {@code null}.
305: * <p>
306: * If greater than 0, it checks if the running {@link JobExecution} has a subset of {@link org.eclipse.kapua.service.job.targets.JobTarget}s compatible with the current {@link JobTargetSublist}.
307: * If the current {@link JobTargetSublist} doesn't match {@link org.eclipse.kapua.service.job.targets.JobTarget}s of any other running JobExecution, returns the current running {@link JobExecution}.
308: * <p>
309: * In other all other cases returns {@code null}.
310: *
311: * @param scopeId The current {@link JobExecution#getScopeId()}.
312: * @param jobId The current {@link JobExecution#getJobId()}.
313: * @param currentJobExecutionId The current {@link JobExecution#getId()}.
314: * @param jobName The current {@link JobContextWrapper#getJobName()}.
315: * @param jobTargetIdSubset The current {@link JobExecution#getTargetIds()} }.
316: * @return The other running {@link JobExecution} or {@code null} if there is no other running {@link JobExecution}.
317: * @throws KapuaException If any error happens during the processing.
318: * @since 1.1.0
319: */
320: private JobExecution getAnotherJobExecutionRunning(KapuaId scopeId, KapuaId jobId, KapuaId currentJobExecutionId, String jobName, Set<KapuaId> jobTargetIdSubset) throws KapuaException {
321: List<Long> runningExecutionsIds = BatchRuntime.getJobOperator().getRunningExecutions(jobName);
322:• if (runningExecutionsIds.size() > 1) {
323:
324: JobExecutionQuery jobExecutionQuery = JOB_EXECUTION_FACTORY.newQuery(scopeId);
325:
326: jobExecutionQuery.setPredicate(
327: jobExecutionQuery.andPredicate(
328: jobExecutionQuery.attributePredicate(JobExecutionAttributes.JOB_ID, jobId),
329: jobExecutionQuery.attributePredicate(JobExecutionAttributes.ENTITY_ID, currentJobExecutionId, AttributePredicate.Operator.NOT_EQUAL),
330: jobExecutionQuery.attributePredicate(JobExecutionAttributes.ENDED_ON, null),
331: jobExecutionQuery.attributePredicate(JobExecutionAttributes.TARGET_IDS, jobTargetIdSubset.toArray())
332: )
333: );
334:
335: jobExecutionQuery.setSortCriteria(jobExecutionQuery.fieldSortCriteria(JobExecutionAttributes.STARTED_ON, SortOrder.ASCENDING));
336:
337: JobExecutionListResult jobExecutions = KapuaSecurityUtils.doPrivileged(() -> JOB_EXECUTION_SERVICE.query(jobExecutionQuery));
338:
339: return jobExecutions.getFirstItem();
340: }
341:
342: return null;
343: }
344:
345: /**
346: * Creates a new {@link QueuedJobExecution} when the current {@link JobExecution} cannot be started
347: * due to another running {@link JobExecution} and the {@link JobStartOptions#getEnqueue()} is {@code true}.
348: *
349: * @param scopeId The current {@link JobExecution#getScopeId()}
350: * @param jobId The current {@link JobExecution#getJobId()}
351: * @param jobExecutionId The current {@link JobExecution#getId()}
352: * @param runningJobExecutionId The current running {@link JobExecution#getId()}
353: * @return The newly created {@link QueuedJobExecution}
354: * @throws KapuaException If any error happens during the processing.
355: */
356: private QueuedJobExecution enqueueJobExecution(KapuaId scopeId, KapuaId jobId, KapuaId jobExecutionId, KapuaId runningJobExecutionId) throws KapuaException {
357:
358: QueuedJobExecutionCreator queuedJobExecutionCreator = QUEUED_JOB_FACTORY.newCreator(scopeId);
359: queuedJobExecutionCreator.setJobId(jobId);
360: queuedJobExecutionCreator.setJobExecutionId(jobExecutionId);
361: queuedJobExecutionCreator.setWaitForJobExecutionId(runningJobExecutionId);
362: queuedJobExecutionCreator.setStatus(QueuedJobExecutionStatus.QUEUED);
363:
364: return KapuaSecurityUtils.doPrivileged(() -> QUEUED_JOB_SERVICE.create(queuedJobExecutionCreator));
365: }
366:
367: private void checkQueuedJobExecutions(KapuaId scopeId, KapuaId jobId, KapuaId jobExecutionId) {
368:
369: String timerName = new StringBuilder()
370: .append(KapuaJobListener.class.getSimpleName())
371: .append("-DelayedCheckTimer-")
372: .append(jobExecutionId)
373: .toString();
374:
375: Timer queuedCheckTimer = new Timer(timerName, true);
376:
377: queuedCheckTimer.schedule(new QueuedJobExecutionCheckTask(scopeId, jobId, jobExecutionId), JOB_ENGINE_SETTING.getLong(JobEngineSettingKeys.JOB_ENGINE_QUEUE_CHECK_DELAY));
378: }
379: }