You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by cz...@apache.org on 2014/10/14 16:22:33 UTC
svn commit: r1631770 - in
/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl:
jobs/BackgroundLoader.java jobs/JobManagerImpl.java jobs/Utility.java
support/ResourceHelper.java
Author: cziegeler
Date: Tue Oct 14 14:22:33 2014
New Revision: 1631770
URL: http://svn.apache.org/r1631770
Log:
SLING-4048 : Avoid keeping jobs in memory. Move job reading to a utility class
Modified:
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/BackgroundLoader.java
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/Utility.java
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/BackgroundLoader.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/BackgroundLoader.java?rev=1631770&r1=1631769&r2=1631770&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/BackgroundLoader.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/BackgroundLoader.java Tue Oct 14 14:22:33 2014
@@ -249,7 +249,7 @@ public class BackgroundLoader implements
} else {
if (ResourceHelper.RESOURCE_TYPE_JOB.equals(resource.getResourceType()) ) {
this.logger.debug("Reading local job from {}", path);
- final JobImpl job = this.jobManager.readJob(resource);
+ final JobImpl job = Utility.readJob(logger, resource);
if ( job != null ) {
if ( job.hasReadErrors() ) {
synchronized ( this.unloadedJobs ) {
@@ -479,7 +479,7 @@ public class BackgroundLoader implements
while ( this.isRunning() && jobIter.hasNext() ) {
final Resource jobResource = jobIter.next();
- final JobImpl job = this.jobManager.readJob(jobResource);
+ final JobImpl job = Utility.readJob(logger, jobResource);
if ( job != null && job.getCreated().compareTo(now) <= 0 ) {
logger.debug("Found job {}", jobResource.getName());
jobs.add(job);
@@ -528,7 +528,7 @@ public class BackgroundLoader implements
private boolean loadJobInTheBackground(final Resource jobResource) {
// sanity check for the path
if ( this.configuration.isLocalJob(jobResource.getPath()) ) {
- final JobImpl job = this.jobManager.readJob(jobResource);
+ final JobImpl job = Utility.readJob(logger, jobResource);
if ( job != null ) {
// check if the job is currently running
if ( this.firstRun || job.getProcessingStarted() == null ) {
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java?rev=1631770&r1=1631769&r2=1631770&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobManagerImpl.java Tue Oct 14 14:22:33 2014
@@ -494,66 +494,6 @@ public class JobManagerImpl
}
}
- /**
- * Read a job
- */
- JobImpl readJob(final Resource resource) {
- JobImpl job = null;
- if ( resource != null ) {
- try {
- final ValueMap vm = ResourceHelper.getValueMap(resource);
-
- // check job topic and job id
- final String errorMessage = Utility.checkJobTopic(vm.get(ResourceHelper.PROPERTY_JOB_TOPIC));
- final String jobId = vm.get(ResourceHelper.PROPERTY_JOB_ID, String.class);
- if ( errorMessage == null && jobId != null ) {
- final String topic = vm.get(ResourceHelper.PROPERTY_JOB_TOPIC, String.class);
- final Map<String, Object> jobProperties = ResourceHelper.cloneValueMap(vm);
-
- jobProperties.put(JobImpl.PROPERTY_RESOURCE_PATH, resource.getPath());
- // convert to integers (JCR supports only long...)
- jobProperties.put(Job.PROPERTY_JOB_RETRIES, vm.get(Job.PROPERTY_JOB_RETRIES, Integer.class));
- jobProperties.put(Job.PROPERTY_JOB_RETRY_COUNT, vm.get(Job.PROPERTY_JOB_RETRY_COUNT, Integer.class));
- if ( vm.get(Job.PROPERTY_JOB_PROGRESS_STEPS) != null ) {
- jobProperties.put(Job.PROPERTY_JOB_PROGRESS_STEPS, vm.get(Job.PROPERTY_JOB_PROGRESS_STEPS, Integer.class));
- }
- if ( vm.get(Job.PROPERTY_JOB_PROGRESS_STEP) != null ) {
- jobProperties.put(Job.PROPERTY_JOB_PROGRESS_STEP, vm.get(Job.PROPERTY_JOB_PROGRESS_STEP, Integer.class));
- }
- @SuppressWarnings("unchecked")
- final List<Exception> readErrorList = (List<Exception>) jobProperties.get(ResourceHelper.PROPERTY_MARKER_READ_ERROR_LIST);
- if ( readErrorList != null ) {
- for(final Exception e : readErrorList) {
- logger.warn("Unable to read job from " + resource.getPath(), e);
- }
- }
- job = new JobImpl(topic,
- (String)jobProperties.get(ResourceHelper.PROPERTY_JOB_NAME),
- jobId,
- jobProperties);
- } else {
- if ( errorMessage != null ) {
- logger.warn("{} : {}", errorMessage, resource.getPath());
- } else if ( jobId == null ) {
- logger.warn("Discarding job - no job id found : {}", resource.getPath());
- }
- // remove the job as the topic is invalid anyway
- try {
- resource.getResourceResolver().delete(resource);
- resource.getResourceResolver().commit();
- } catch ( final PersistenceException ignore) {
- this.ignoreException(ignore);
- }
- }
- } catch (final InstantiationException ie) {
- // something happened with the resource in the meantime
- this.ignoreException(ie);
- }
-
- }
- return job;
- }
-
private void stopProcessing() {
this.backgroundLoader.stop();
@@ -847,7 +787,7 @@ public class JobManagerImpl
final Resource jobResource = result.next();
// sanity check for the path
if ( this.configuration.isJob(jobResource.getPath()) ) {
- final JobImpl job = this.readJob(jobResource);
+ final JobImpl job = Utility.readJob(logger, jobResource);
if ( job != null ) {
return job;
}
@@ -893,7 +833,7 @@ public class JobManagerImpl
final Resource jobResource = result.next();
// sanity check for the path
if ( this.configuration.isJob(jobResource.getPath()) ) {
- final JobImpl job = this.readJob(jobResource);
+ final JobImpl job = Utility.readJob(logger, jobResource);
if ( job != null ) {
if ( logger.isDebugEnabled() ) {
logger.debug("Found job with id {} = {}", id, job);
@@ -1109,7 +1049,7 @@ public class JobManagerImpl
final Resource jobResource = iter.next();
// sanity check for the path
if ( this.configuration.isJob(jobResource.getPath()) ) {
- final JobImpl job = readJob(jobResource);
+ final JobImpl job = Utility.readJob(logger, jobResource);
if ( job != null ) {
count++;
result.add(job);
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/Utility.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/Utility.java?rev=1631770&r1=1631769&r2=1631770&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/Utility.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/Utility.java Tue Oct 14 14:22:33 2014
@@ -23,8 +23,12 @@ import java.util.Calendar;
import java.util.Dictionary;
import java.util.HashMap;
import java.util.Hashtable;
+import java.util.List;
import java.util.Map;
+import org.apache.sling.api.resource.PersistenceException;
+import org.apache.sling.api.resource.Resource;
+import org.apache.sling.api.resource.ValueMap;
import org.apache.sling.event.impl.support.ResourceHelper;
import org.apache.sling.event.jobs.Job;
import org.apache.sling.event.jobs.JobUtil;
@@ -33,6 +37,7 @@ import org.apache.sling.event.jobs.consu
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.osgi.service.event.EventConstants;
+import org.slf4j.Logger;
public abstract class Utility {
@@ -253,4 +258,64 @@ public abstract class Utility {
sb.append("]");
return sb.toString();
}
+
+ /**
+ * Read a job
+ */
+ public static JobImpl readJob(final Logger logger, final Resource resource) {
+ JobImpl job = null;
+ if ( resource != null ) {
+ try {
+ final ValueMap vm = ResourceHelper.getValueMap(resource);
+
+ // check job topic and job id
+ final String errorMessage = Utility.checkJobTopic(vm.get(ResourceHelper.PROPERTY_JOB_TOPIC));
+ final String jobId = vm.get(ResourceHelper.PROPERTY_JOB_ID, String.class);
+ if ( errorMessage == null && jobId != null ) {
+ final String topic = vm.get(ResourceHelper.PROPERTY_JOB_TOPIC, String.class);
+ final Map<String, Object> jobProperties = ResourceHelper.cloneValueMap(vm);
+
+ jobProperties.put(JobImpl.PROPERTY_RESOURCE_PATH, resource.getPath());
+ // convert to integers (JCR supports only long...)
+ jobProperties.put(Job.PROPERTY_JOB_RETRIES, vm.get(Job.PROPERTY_JOB_RETRIES, Integer.class));
+ jobProperties.put(Job.PROPERTY_JOB_RETRY_COUNT, vm.get(Job.PROPERTY_JOB_RETRY_COUNT, Integer.class));
+ if ( vm.get(Job.PROPERTY_JOB_PROGRESS_STEPS) != null ) {
+ jobProperties.put(Job.PROPERTY_JOB_PROGRESS_STEPS, vm.get(Job.PROPERTY_JOB_PROGRESS_STEPS, Integer.class));
+ }
+ if ( vm.get(Job.PROPERTY_JOB_PROGRESS_STEP) != null ) {
+ jobProperties.put(Job.PROPERTY_JOB_PROGRESS_STEP, vm.get(Job.PROPERTY_JOB_PROGRESS_STEP, Integer.class));
+ }
+ @SuppressWarnings("unchecked")
+ final List<Exception> readErrorList = (List<Exception>) jobProperties.get(ResourceHelper.PROPERTY_MARKER_READ_ERROR_LIST);
+ if ( readErrorList != null ) {
+ for(final Exception e : readErrorList) {
+ logger.warn("Unable to read job from " + resource.getPath(), e);
+ }
+ }
+ job = new JobImpl(topic,
+ (String)jobProperties.get(ResourceHelper.PROPERTY_JOB_NAME),
+ jobId,
+ jobProperties);
+ } else {
+ if ( errorMessage != null ) {
+ logger.warn("{} : {}", errorMessage, resource.getPath());
+ } else if ( jobId == null ) {
+ logger.warn("Discarding job - no job id found : {}", resource.getPath());
+ }
+ // remove the job as the topic is invalid anyway
+ try {
+ resource.getResourceResolver().delete(resource);
+ resource.getResourceResolver().commit();
+ } catch ( final PersistenceException ignore) {
+ logger.debug("Unable to remove job resource.", ignore);
+ }
+ }
+ } catch (final InstantiationException ie) {
+ // something happened with the resource in the meantime
+ logger.debug("Unable to instantiate resource.", ie);
+ }
+
+ }
+ return job;
+ }
}
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java?rev=1631770&r1=1631769&r2=1631770&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/support/ResourceHelper.java Tue Oct 14 14:22:33 2014
@@ -140,7 +140,7 @@ public abstract class ResourceHelper {
return ResourceHelper.filterName(queueName);
}
}
-
+
/**
* Filter the node name for not allowed characters and replace them.
* @param resourceName The suggested resource name.
@@ -266,8 +266,7 @@ public abstract class ResourceHelper {
public static void getOrCreateBasePath(final ResourceResolver resolver,
final String path)
throws PersistenceException {
- // TODO - we should rather fix ResourceUtil.getOrCreateResource:
- // on concurrent writes, create might fail!
+ PersistenceException mostRecentPE = null;
for(int i=0;i<5;i++) {
try {
ResourceUtil.getOrCreateResource(resolver,
@@ -277,17 +276,17 @@ public abstract class ResourceHelper {
true);
return;
} catch ( final PersistenceException pe ) {
- // ignore
+ //in case of exception, revert to last clean state and retry SLING-4014
+ resolver.revert();
+ mostRecentPE = pe;
}
}
- throw new PersistenceException("Unable to create resource with path " + path);
+ throw new PersistenceException("Unable to create resource with path " + path, mostRecentPE);
}
public static Resource getOrCreateResource(final ResourceResolver resolver,
final String path, final Map<String, Object> props)
throws PersistenceException {
- // TODO - we should rather fix ResourceUtil.getOrCreateResource:
- // on concurrent writes, create might fail!
PersistenceException mostRecentPE = null;
for(int i=0;i<5;i++) {
try {