| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.batchee.container.services.persistence; |
| |
| import org.apache.batchee.container.impl.JobExecutionImpl; |
| import org.apache.batchee.container.impl.JobInstanceImpl; |
| import org.apache.batchee.container.impl.MetricImpl; |
| import org.apache.batchee.container.impl.StepContextImpl; |
| import org.apache.batchee.container.impl.StepExecutionImpl; |
| import org.apache.batchee.container.impl.controller.PartitionedStepBuilder; |
| import org.apache.batchee.container.impl.controller.chunk.CheckpointData; |
| import org.apache.batchee.container.impl.controller.chunk.CheckpointDataKey; |
| import org.apache.batchee.container.impl.jobinstance.RuntimeFlowInSplitExecution; |
| import org.apache.batchee.container.impl.jobinstance.RuntimeJobExecution; |
| import org.apache.batchee.container.services.InternalJobExecution; |
| import org.apache.batchee.container.status.JobStatus; |
| import org.apache.batchee.container.status.StepStatus; |
| import org.apache.batchee.spi.PersistenceManagerService; |
| |
| import javax.batch.operations.NoSuchJobExecutionException; |
| import javax.batch.runtime.BatchStatus; |
| import javax.batch.runtime.JobInstance; |
| import javax.batch.runtime.Metric; |
| import javax.batch.runtime.StepExecution; |
| import java.io.Serializable; |
| import java.sql.Timestamp; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.Date; |
| import java.util.HashSet; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CopyOnWriteArrayList; |
| import java.util.concurrent.atomic.AtomicLong; |
| |
| public class MemoryPersistenceManagerService implements PersistenceManagerService { |
| private static final Collection<BatchStatus> RUNNING_STATUSES = new CopyOnWriteArrayList<BatchStatus>() {{ |
| add(BatchStatus.STARTED); |
| add(BatchStatus.STARTING); |
| add(BatchStatus.STOPPING); |
| }}; |
| |
| protected static interface Structures extends Serializable { |
| static class JobInstanceData implements Structures { |
| protected JobInstanceImpl instance; |
| protected JobStatus status; |
| protected final List<ExecutionInstanceData> executions = new LinkedList<ExecutionInstanceData>(); |
| protected final Collection<CheckpointDataKey> checkpoints = new LinkedList<CheckpointDataKey>(); |
| } |
| |
| static class ExecutionInstanceData implements Structures { |
| protected final List<StepExecution> stepExecutions = new LinkedList<StepExecution>(); |
| protected JobExecutionImpl execution; |
| } |
| |
| static class StepExecutionInstanceData implements Structures { |
| protected ExecutionInstanceData jobExec; |
| protected StepExecutionImpl execution; |
| protected StepStatus status; |
| } |
| } |
| |
| static class Data { |
| protected final AtomicLong lastCleanedJobInstanceId = new AtomicLong(0); |
| protected final AtomicLong jobInstanceIdGenerator = new AtomicLong(); |
| protected final AtomicLong executionInstanceIdGenerator = new AtomicLong(); |
| protected final AtomicLong stepExecutionIdGenerator = new AtomicLong(); |
| |
| protected final Map<CheckpointDataKey, CheckpointData> checkpointData = new ConcurrentHashMap<CheckpointDataKey, CheckpointData>(); |
| protected final Map<Long, Structures.JobInstanceData> jobInstanceData = new ConcurrentHashMap<Long, Structures.JobInstanceData>(); |
| protected final Map<Long, Structures.ExecutionInstanceData> executionInstanceData = new ConcurrentHashMap<Long, Structures.ExecutionInstanceData>(); |
| protected final Map<Long, Structures.StepExecutionInstanceData> stepExecutionInstanceData = new ConcurrentHashMap<Long, Structures.StepExecutionInstanceData>(); |
| } |
| |
| private static final Data GLOBAL_DATA = new Data(); |
| |
| private Data data; |
| private int maxSize; |
| |
| // an extension point to be able to use whatever map you want, a distributed/filtered/size-limited one for instance |
| protected Data newData() { |
| return new Data(); |
| } |
| |
| @Override |
| public void init(final Properties batchConfig) { |
| if ("true".equalsIgnoreCase(batchConfig.getProperty("persistence.memory.global", "false"))) { |
| data = GLOBAL_DATA; |
| } else { |
| data = newData(); |
| } |
| maxSize = Integer.parseInt(batchConfig.getProperty("persistence.memory.max-jobs-instances", "1000")); |
| } |
| |
| @Override |
| public int jobOperatorGetJobInstanceCount(final String jobName) { |
| int i = 0; |
| for (final Structures.JobInstanceData jobInstanceData : data.jobInstanceData.values()) { |
| if (jobInstanceData.instance.getJobName().equals(jobName)) { |
| i++; |
| } |
| } |
| return i; |
| } |
| |
| @Override |
| public Set<String> getJobNames() { |
| Set<String> jobNames = new HashSet<String>(); |
| for (final Structures.JobInstanceData jobInstanceData : data.jobInstanceData.values()) { |
| if (jobInstanceData.instance.getJobName() != null && !jobInstanceData.instance.getJobName().startsWith(PartitionedStepBuilder.JOB_ID_SEPARATOR)) { |
| jobNames.add(jobInstanceData.instance.getJobName()); |
| } |
| } |
| return jobNames; |
| } |
| |
| @Override |
| public List<Long> jobOperatorGetJobInstanceIds(final String jobName, final int start, final int count) { |
| final List<Long> out = new LinkedList<Long>(); |
| for (final Structures.JobInstanceData jobInstanceData : data.jobInstanceData.values()) { |
| if (jobInstanceData.instance.getJobName() != null && jobInstanceData.instance.getJobName().equals(jobName)) { |
| out.add(jobInstanceData.instance.getInstanceId()); |
| } |
| } |
| |
| // sorting could be optimized a bit but is it necessary for this impl? |
| Collections.sort(out); |
| // Collections.reverse(out); |
| |
| if (out.size() > 0) { |
| try { |
| return out.subList(start, start + count); |
| } catch (final IndexOutOfBoundsException oobEx) { |
| return out.subList(start, out.size()); |
| } |
| } |
| |
| return out; |
| } |
| |
| @Override |
| public Timestamp jobOperatorQueryJobExecutionTimestamp(final long key, final TimestampType timestampType) { |
| return null; // avoid infinite loops |
| } |
| |
| @Override |
| public String jobOperatorQueryJobExecutionBatchStatus(final long key) { |
| return null; // avoid infinite loops |
| } |
| |
| @Override |
| public String jobOperatorQueryJobExecutionExitStatus(final long key) { |
| return null; // avoid infinite loops |
| } |
| |
| @Override |
| public List<StepExecution> getStepExecutionsForJobExecution(final long execid) { |
| final Structures.ExecutionInstanceData executionInstanceData = data.executionInstanceData.get(execid); |
| if (executionInstanceData == null) { |
| return Collections.emptyList(); |
| } |
| |
| synchronized (executionInstanceData.stepExecutions) { |
| return executionInstanceData.stepExecutions; |
| } |
| } |
| |
| @Override |
| public void updateBatchStatusOnly(final long executionId, final BatchStatus batchStatus, final Timestamp timestamp) { |
| final Structures.ExecutionInstanceData toUpdate = data.executionInstanceData.get(executionId); |
| toUpdate.execution.setBatchStatus(batchStatus.name()); |
| toUpdate.execution.setLastUpdateTime(timestamp); |
| } |
| |
| @Override |
| public void markJobStarted(final long key, final Timestamp startTS) { |
| final Structures.ExecutionInstanceData toUpdate = data.executionInstanceData.get(key); |
| toUpdate.execution.setBatchStatus(BatchStatus.STARTED.name()); |
| toUpdate.execution.setLastUpdateTime(startTS); |
| toUpdate.execution.setStartTime(startTS); |
| } |
| |
| @Override |
| public void updateWithFinalExecutionStatusesAndTimestamps(final long key, final BatchStatus batchStatus, final String exitStatus, final Timestamp updatets) { |
| final Structures.ExecutionInstanceData toUpdate = data.executionInstanceData.get(key); |
| if (batchStatus != null) { |
| toUpdate.execution.setBatchStatus(batchStatus.name()); |
| } else { |
| toUpdate.execution.setBatchStatus(null); |
| } |
| toUpdate.execution.setExitStatus(exitStatus); |
| toUpdate.execution.setLastUpdateTime(updatets); |
| toUpdate.execution.setEndTime(updatets); |
| } |
| |
| @Override |
| public InternalJobExecution jobOperatorGetJobExecution(final long jobExecutionId) { |
| final Structures.ExecutionInstanceData executionInstanceData = data.executionInstanceData.get(jobExecutionId); |
| if (executionInstanceData == null) { |
| return null; |
| } |
| return executionInstanceData.execution; |
| } |
| |
| @Override |
| public Properties getParameters(final long executionId) throws NoSuchJobExecutionException { |
| return data.executionInstanceData.get(executionId).execution.getJobParameters(); |
| } |
| |
| @Override |
| public List<InternalJobExecution> jobOperatorGetJobExecutions(final long jobInstanceId) { |
| final List<InternalJobExecution> list = new LinkedList<InternalJobExecution>(); |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(jobInstanceId); |
| if (jobInstanceData == null) { |
| return list; |
| } |
| synchronized (jobInstanceData.executions) { |
| for (final Structures.ExecutionInstanceData executionInstanceData : jobInstanceData.executions) { |
| list.add(executionInstanceData.execution); |
| } |
| } |
| return list; |
| } |
| |
| @Override |
| public Set<Long> jobOperatorGetRunningExecutions(final String jobName) { |
| final Set<Long> set = new HashSet<Long>(); |
| |
| for (final Structures.JobInstanceData instanceData : data.jobInstanceData.values()) { |
| if (instanceData.instance.getJobName().equals(jobName)) { |
| synchronized (instanceData.executions) { |
| for (final Structures.ExecutionInstanceData executionInstanceData : instanceData.executions) { |
| if (RUNNING_STATUSES.contains(executionInstanceData.execution.getBatchStatus())) { |
| set.add(executionInstanceData.execution.getExecutionId()); |
| } |
| } |
| } |
| } |
| } |
| |
| return set; |
| } |
| |
| @Override |
| public JobStatus getJobStatusFromExecution(final long executionId) { |
| final Structures.ExecutionInstanceData executionInstanceData = data.executionInstanceData.get(executionId); |
| if (executionInstanceData == null) { |
| return null; |
| } |
| |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(executionInstanceData.execution.getInstanceId()); |
| if (jobInstanceData == null) { |
| return null; |
| } |
| return jobInstanceData.status; |
| } |
| |
| @Override |
| public long getJobInstanceIdByExecutionId(final long executionId) throws NoSuchJobExecutionException { |
| final Structures.ExecutionInstanceData executionInstanceData = data.executionInstanceData.get(executionId); |
| if (executionInstanceData == null) { |
| throw new NoSuchJobExecutionException("Execution #" + executionId); |
| } |
| return executionInstanceData.execution.getInstanceId(); |
| } |
| |
| @Override |
| public JobInstance createJobInstance(final String name, final String jobXml) { |
| final JobInstanceImpl jobInstance = new JobInstanceImpl(data.jobInstanceIdGenerator.getAndIncrement(), jobXml); |
| jobInstance.setJobName(name); |
| |
| final Structures.JobInstanceData jobInstanceData = new Structures.JobInstanceData(); |
| jobInstanceData.instance = jobInstance; |
| if (maxSize > 0 && data.jobInstanceData.size() >= maxSize) { |
| synchronized (this) { |
| while (data.jobInstanceData.size() >= maxSize) { |
| cleanUp(data.lastCleanedJobInstanceId.getAndIncrement()); |
| } |
| } |
| } |
| data.jobInstanceData.put(jobInstance.getInstanceId(), jobInstanceData); |
| |
| return jobInstance; |
| } |
| |
| @Override |
| public RuntimeJobExecution createJobExecution(final JobInstance jobInstance, final Properties jobParameters, final BatchStatus batchStatus) { |
| final Timestamp now = new Timestamp(System.currentTimeMillis()); |
| |
| final Structures.ExecutionInstanceData executionInstanceData = createRuntimeJobExecutionEntry(jobInstance, jobParameters, batchStatus, now); |
| executionInstanceData.execution.setJobName(jobInstance.getJobName()); |
| |
| final RuntimeJobExecution jobExecution = new RuntimeJobExecution(jobInstance, executionInstanceData.execution.getExecutionId(), this); |
| jobExecution.setBatchStatus(batchStatus.name()); |
| jobExecution.setCreateTime(now); |
| jobExecution.setLastUpdateTime(now); |
| |
| return jobExecution; |
| } |
| |
| private Structures.ExecutionInstanceData createRuntimeJobExecutionEntry(final JobInstance jobInstance, final Properties jobParameters, |
| final BatchStatus batchStatus, final Timestamp now) { |
| final Structures.ExecutionInstanceData executionInstanceData = new Structures.ExecutionInstanceData(); |
| final long id = data.executionInstanceIdGenerator.getAndIncrement(); |
| executionInstanceData.execution = new JobExecutionImpl(id, jobInstance.getInstanceId(), this); |
| executionInstanceData.execution.setExecutionId(id); |
| executionInstanceData.execution.setInstanceId(jobInstance.getInstanceId()); |
| executionInstanceData.execution.setBatchStatus(batchStatus.name()); |
| executionInstanceData.execution.setCreateTime(now); |
| executionInstanceData.execution.setLastUpdateTime(now); |
| executionInstanceData.execution.setJobParameters(jobParameters); |
| |
| data.executionInstanceData.put(id, executionInstanceData); |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(jobInstance.getInstanceId()); |
| synchronized (jobInstanceData.executions) { |
| jobInstanceData.executions.add(executionInstanceData); |
| } |
| return executionInstanceData; |
| } |
| |
| @Override |
| public StepExecutionImpl createStepExecution(final long rootJobExecId, final StepContextImpl stepContext) { |
| final String batchStatus = stepContext.getBatchStatus() == null ? BatchStatus.STARTING.name() : stepContext.getBatchStatus().name(); |
| final String exitStatus = stepContext.getExitStatus(); |
| final String stepName = stepContext.getStepName(); |
| |
| long readCount = 0; |
| long writeCount = 0; |
| long commitCount = 0; |
| long rollbackCount = 0; |
| long readSkipCount = 0; |
| long processSkipCount = 0; |
| long filterCount = 0; |
| long writeSkipCount = 0; |
| Timestamp startTime = stepContext.getStartTimeTS(); |
| Timestamp endTime = stepContext.getEndTimeTS(); |
| |
| final Metric[] metrics = stepContext.getMetrics(); |
| for (final Metric metric : metrics) { |
| if (metric.getType().equals(MetricImpl.MetricType.READ_COUNT)) { |
| readCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.WRITE_COUNT)) { |
| writeCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.PROCESS_SKIP_COUNT)) { |
| processSkipCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.COMMIT_COUNT)) { |
| commitCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.ROLLBACK_COUNT)) { |
| rollbackCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.READ_SKIP_COUNT)) { |
| readSkipCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.FILTER_COUNT)) { |
| filterCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.WRITE_SKIP_COUNT)) { |
| writeSkipCount = metric.getValue(); |
| } |
| } |
| final Serializable persistentData = stepContext.getPersistentUserData(); |
| |
| return createStepExecution(rootJobExecId, batchStatus, exitStatus, stepName, readCount, |
| writeCount, commitCount, rollbackCount, readSkipCount, processSkipCount, filterCount, writeSkipCount, startTime, |
| endTime, persistentData); |
| } |
| |
| //CHECKSTYLE:OFF |
| private StepExecutionImpl createStepExecution(final long rootJobExecId, final String batchStatus, final String exitStatus, final String stepName, |
| final long readCount, final long writeCount, final long commitCount, final long rollbackCount, |
| final long readSkipCount, final long processSkipCount, final long filterCount, final long writeSkipCount, |
| final Timestamp startTime, final Timestamp endTime, final Serializable persistentData) { |
| //CHECKSTYLE:ON |
| |
| final StepExecutionImpl stepExecution = new StepExecutionImpl(rootJobExecId, data.stepExecutionIdGenerator.getAndIncrement()); |
| stepExecution.setStepName(stepName); |
| final Structures.ExecutionInstanceData executionInstanceData = data.executionInstanceData.get(rootJobExecId); |
| if (executionInstanceData == null) { |
| return null; |
| } |
| synchronized (executionInstanceData.stepExecutions) { |
| executionInstanceData.stepExecutions.add(stepExecution); |
| } |
| |
| final Structures.StepExecutionInstanceData stepExecutionInstanceData = new Structures.StepExecutionInstanceData(); |
| stepExecutionInstanceData.execution = stepExecution; |
| updateStepExecutionInstanceData(executionInstanceData, batchStatus, exitStatus, stepName, |
| readCount, writeCount, commitCount, rollbackCount, |
| readSkipCount, processSkipCount, filterCount, writeSkipCount, |
| startTime, endTime, persistentData, |
| stepExecutionInstanceData); |
| data.stepExecutionInstanceData.put(stepExecution.getStepExecutionId(), stepExecutionInstanceData); |
| |
| return stepExecution; |
| } |
| |
| //CHECKSTYLE:OFF |
| private void updateStepExecutionInstanceData(final Structures.ExecutionInstanceData exec, |
| final String batchStatus,final String exitStatus, final String stepName, |
| final long readCount, final long writeCount, final long commitCount, final long rollbackCount, |
| final long readSkipCount, final long processSkipCount, final long filterCount, final long writeSkipCount, |
| final Timestamp startTime, final Timestamp endTime, |
| final Serializable persistentData, |
| final Structures.StepExecutionInstanceData stepExecutionInstanceData) { |
| //CHECKSTYLE:ON |
| stepExecutionInstanceData.jobExec = exec; |
| stepExecutionInstanceData.execution.setExitStatus(exitStatus); |
| stepExecutionInstanceData.execution.setBatchStatus(BatchStatus.valueOf(batchStatus)); |
| stepExecutionInstanceData.execution.setRollbackCount(rollbackCount); |
| stepExecutionInstanceData.execution.setStepName(stepName); |
| stepExecutionInstanceData.execution.setReadCount(readCount); |
| stepExecutionInstanceData.execution.setWriteCount(writeCount); |
| stepExecutionInstanceData.execution.setCommitCount(commitCount); |
| stepExecutionInstanceData.execution.setRollbackCount(rollbackCount); |
| stepExecutionInstanceData.execution.setReadSkipCount(readSkipCount); |
| stepExecutionInstanceData.execution.setProcessSkipCount(processSkipCount); |
| stepExecutionInstanceData.execution.setWriteSkipCount(writeSkipCount); |
| stepExecutionInstanceData.execution.setFilterCount(filterCount); |
| stepExecutionInstanceData.execution.setWriteSkipCount(writeSkipCount); |
| stepExecutionInstanceData.execution.setStartTime(startTime); |
| stepExecutionInstanceData.execution.setEndTime(endTime); |
| stepExecutionInstanceData.execution.setPersistentUserData(persistentData); |
| } |
| |
| @Override |
| public void updateStepExecution(final long jobExecId, final StepContextImpl stepContext) { |
| final long stepExecutionId = stepContext.getStepInternalExecID(); |
| final String batchStatus = stepContext.getBatchStatus() == null ? BatchStatus.STARTING.name() : stepContext.getBatchStatus().name(); |
| final String exitStatus = stepContext.getExitStatus(); |
| final String stepName = stepContext.getStepName(); |
| |
| long readCount = 0; |
| long writeCount = 0; |
| long commitCount = 0; |
| long rollbackCount = 0; |
| long readSkipCount = 0; |
| long processSkipCount = 0; |
| long filterCount = 0; |
| long writeSkipCount = 0; |
| Timestamp startTime = stepContext.getStartTimeTS(); |
| Timestamp endTime = stepContext.getEndTimeTS(); |
| |
| final Metric[] metrics = stepContext.getMetrics(); |
| for (final Metric metric : metrics) { |
| if (metric.getType().equals(MetricImpl.MetricType.READ_COUNT)) { |
| readCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.WRITE_COUNT)) { |
| writeCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.PROCESS_SKIP_COUNT)) { |
| processSkipCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.COMMIT_COUNT)) { |
| commitCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.ROLLBACK_COUNT)) { |
| rollbackCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.READ_SKIP_COUNT)) { |
| readSkipCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.FILTER_COUNT)) { |
| filterCount = metric.getValue(); |
| } else if (metric.getType().equals(MetricImpl.MetricType.WRITE_SKIP_COUNT)) { |
| writeSkipCount = metric.getValue(); |
| } |
| } |
| final Serializable persistentData = stepContext.getPersistentUserData(); |
| |
| updateStepExecution(stepExecutionId, jobExecId, batchStatus, exitStatus, stepName, readCount, |
| writeCount, commitCount, rollbackCount, readSkipCount, processSkipCount, filterCount, |
| writeSkipCount, startTime, endTime, persistentData); |
| } |
| |
| //CHECKSTYLE:OFF |
| private void updateStepExecution(final long stepExecutionId, final long jobExecId, final String batchStatus, final String exitStatus, final String stepName, |
| final long readCount, final long writeCount, final long commitCount, final long rollbackCount, |
| final long readSkipCount, final long processSkipCount, final long filterCount, final long writeSkipCount, |
| final Timestamp startTime, final Timestamp endTime, final Serializable persistentData) { |
| //CHECKSTYLE:ON |
| final Structures.ExecutionInstanceData executionInstanceData = data.executionInstanceData.get(jobExecId); |
| if (executionInstanceData == null) { |
| return; |
| } |
| |
| synchronized (executionInstanceData.stepExecutions) { |
| for (final StepExecution execution : executionInstanceData.stepExecutions) { |
| if (execution.getStepExecutionId() == stepExecutionId) { |
| updateStepExecutionInstanceData(executionInstanceData, batchStatus, exitStatus, stepName, |
| readCount, writeCount, commitCount, rollbackCount, |
| readSkipCount, processSkipCount, filterCount, writeSkipCount, |
| startTime, endTime, persistentData, |
| data.stepExecutionInstanceData.get(stepExecutionId)); |
| break; |
| } |
| } |
| } |
| } |
| |
| @Override |
| public JobStatus createJobStatus(final long jobInstanceId) { |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(jobInstanceId); |
| jobInstanceData.status = new JobStatus(jobInstanceId); |
| return jobInstanceData.status; |
| } |
| |
| @Override |
| public JobStatus getJobStatus(final long instanceId) { |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(instanceId); |
| if (jobInstanceData == null) { |
| return null; |
| } |
| return jobInstanceData.status; |
| } |
| |
| @Override |
| public void updateJobStatus(final long instanceId, final JobStatus jobStatus) { |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(instanceId); |
| if (jobInstanceData != null) { |
| jobInstanceData.status = jobStatus; |
| } |
| } |
| |
| @Override |
| public StepStatus createStepStatus(final long stepExecId) { |
| final StepStatus stepStatus = new StepStatus(stepExecId); |
| data.stepExecutionInstanceData.get(stepExecId).status = stepStatus; |
| return stepStatus; |
| } |
| |
| @Override |
| public StepStatus getStepStatus(final long instanceId, final String stepName) { |
| final Map<Date, StepStatus> statusMap = new TreeMap<Date, StepStatus>(ReverseDateComparator.INSTANCE); |
| for (final Structures.StepExecutionInstanceData stepExecutionInstanceData : data.stepExecutionInstanceData.values()) { |
| if (stepExecutionInstanceData.jobExec.execution.getInstanceId() == instanceId |
| && stepExecutionInstanceData.execution.getStepName().equals(stepName) |
| && stepExecutionInstanceData.status != null) { |
| statusMap.put(stepExecutionInstanceData.jobExec.execution.getCreateTime(), stepExecutionInstanceData.status); |
| } |
| } |
| if (statusMap.isEmpty()) { |
| return null; |
| } |
| return statusMap.values().iterator().next(); |
| } |
| |
| @Override |
| public void updateStepStatus(final long stepExecutionId, final StepStatus stepStatus) { |
| final Structures.StepExecutionInstanceData stepExecutionInstanceData = data.stepExecutionInstanceData.get(stepExecutionId); |
| if (stepExecutionInstanceData != null) { |
| stepExecutionInstanceData.status = stepStatus; |
| } |
| } |
| |
| @Override |
| public void setCheckpointData(final CheckpointDataKey key, final CheckpointData value) { |
| data.checkpointData.put(key, value); |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(key.getJobInstanceId()); |
| synchronized (jobInstanceData.checkpoints) { |
| jobInstanceData.checkpoints.add(key); |
| } |
| } |
| |
| @Override |
| public CheckpointData getCheckpointData(final CheckpointDataKey key) { |
| return data.checkpointData.get(key); |
| } |
| |
| @Override |
| public long getMostRecentExecutionId(final long jobInstanceId) { |
| final Map<Date, Structures.ExecutionInstanceData> filter = new TreeMap<Date, Structures.ExecutionInstanceData>(ReverseDateComparator.INSTANCE); |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(jobInstanceId); |
| synchronized (jobInstanceData.executions) { |
| for (final Structures.ExecutionInstanceData exec : jobInstanceData.executions) { |
| if (exec.execution.getInstanceId() == jobInstanceId) { |
| filter.put(exec.execution.getCreateTime(), exec); |
| } |
| } |
| } |
| if (filter.isEmpty()) { |
| return -1; |
| } |
| return filter.values().iterator().next().execution.getExecutionId(); |
| } |
| |
| @Override |
| public JobInstance createSubJobInstance(final String name) { |
| return createJobInstance(name, null); |
| } |
| |
| @Override |
| public RuntimeFlowInSplitExecution createFlowInSplitExecution(final JobInstance jobInstance, final BatchStatus batchStatus) { |
| final Timestamp now = new Timestamp(System.currentTimeMillis()); |
| final Structures.ExecutionInstanceData executionInstanceData = createRuntimeJobExecutionEntry(jobInstance, null, batchStatus, now); |
| final RuntimeFlowInSplitExecution flowExecution = new RuntimeFlowInSplitExecution(jobInstance, executionInstanceData.execution.getExecutionId(), this); |
| flowExecution.setBatchStatus(batchStatus.name()); |
| flowExecution.setCreateTime(now); |
| flowExecution.setLastUpdateTime(now); |
| return flowExecution; |
| } |
| |
| @Override |
| public StepExecution getStepExecutionByStepExecutionId(final long stepExecId) { |
| return data.stepExecutionInstanceData.get(stepExecId).execution; |
| } |
| |
| @Override |
| public void cleanUp(final long instanceId) { |
| final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.remove(instanceId); |
| if (jobInstanceData == null) { |
| return; |
| } |
| |
| synchronized (jobInstanceData.executions) { |
| for (final Structures.ExecutionInstanceData executionInstanceData : jobInstanceData.executions) { |
| data.executionInstanceData.remove(executionInstanceData.execution.getExecutionId()); |
| synchronized (executionInstanceData.stepExecutions) { |
| for (final StepExecution stepExecution : executionInstanceData.stepExecutions) { |
| data.stepExecutionInstanceData.remove(stepExecution.getStepExecutionId()); |
| } |
| } |
| } |
| } |
| synchronized (jobInstanceData.checkpoints) { |
| for (final CheckpointDataKey key : jobInstanceData.checkpoints) { |
| data.checkpointData.remove(key); |
| } |
| } |
| } |
| |
| @Override |
| public void cleanUp(final Date until) { |
| final Collection<Long> instanceIdToRemove = new ArrayList<Long>(); |
| for (final Map.Entry<Long, Structures.JobInstanceData> entry : data.jobInstanceData.entrySet()) { |
| boolean match = true; |
| for (final Structures.ExecutionInstanceData exec : entry.getValue().executions) { |
| if (exec.execution.getEndTime() == null || exec.execution.getEndTime().after(until)) { |
| match = false; |
| break; |
| } |
| } |
| if (match) { |
| instanceIdToRemove.add(entry.getKey()); |
| } |
| } |
| |
| for (final Long id : instanceIdToRemove) { |
| cleanUp(id); |
| } |
| } |
| |
| private static class ReverseDateComparator implements Comparator<Date> { |
| public static final ReverseDateComparator INSTANCE = new ReverseDateComparator(); |
| |
| @Override |
| public int compare(final Date o1, final Date o2) { |
| return o2.compareTo(o1); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| return getClass().getName(); |
| } |
| } |
