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 2010/02/10 17:34:47 UTC
svn commit: r908572 - in
/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event:
EventUtil.java impl/JobEventHandler.java
Author: cziegeler
Date: Wed Feb 10 16:34:47 2010
New Revision: 908572
URL: http://svn.apache.org/viewvc?rev=908572&view=rev
Log:
SLING-1365 : Limit the number of parallel jobs
Add more debug logging.
Modified:
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/EventUtil.java
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/JobEventHandler.java
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/EventUtil.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/EventUtil.java?rev=908572&r1=908571&r2=908572&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/EventUtil.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/EventUtil.java Wed Feb 10 16:34:47 2010
@@ -156,6 +156,7 @@
public static final String TOPIC_JOB_FINISHED = "org/apache/sling/event/notification/job/FINISHED";
/** Asynchronous notification event when a job failed.
+ * If a job execution fails, it is rescheduled for another try.
* The property {@link #PROPERTY_NOTIFICATION_JOB} contains the job event and the
* property {@link org.osgi.service.event.EventConstants#TIMESTAMP} contains the
* timestamp of the event (as a Long).
@@ -163,6 +164,7 @@
public static final String TOPIC_JOB_FAILED = "org/apache/sling/event/notification/job/FAILED";
/** Asynchronous notification event when a job is cancelled.
+ * If a job execution is cancelled it is not rescheduled.
* The property {@link #PROPERTY_NOTIFICATION_JOB} contains the job event and the
* property {@link org.osgi.service.event.EventConstants#TIMESTAMP} contains the
* timestamp of the event (as a Long).
@@ -661,7 +663,9 @@
return "<null>";
}
final StringBuilder buffer = new StringBuilder(e.getClass().getName());
- buffer.append(" [topic=");
+ buffer.append('(');
+ buffer.append(e.hashCode());
+ buffer.append(") [topic=");
buffer.append(e.getTopic());
buffer.append(", properties=");
final String[] names = e.getPropertyNames();
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/JobEventHandler.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/JobEventHandler.java?rev=908572&r1=908571&r2=908572&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/JobEventHandler.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/JobEventHandler.java Wed Feb 10 16:34:47 2010
@@ -404,7 +404,7 @@
process = this.processingEventsList.remove(info.nodePath) != null;
}
if ( process ) {
- this.logger.info("No acknowledge received for job {} stored at {}. Requeueing job.", info.event, info.nodePath);
+ this.logger.info("No acknowledge received for job {} stored at {}. Requeueing job.", EventUtil.toString(info.event), info.nodePath);
this.finishedJob(info.event, info.nodePath, true);
}
}
@@ -451,7 +451,9 @@
this.ignoreException(e);
}
if ( event != null && this.running ) {
- logger.debug("Persisting job {}", event);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Persisting job {}", EventUtil.toString(event));
+ }
final EventInfo info = new EventInfo();
info.event = event;
final String jobId = (String)event.getProperty(EventUtil.PROPERTY_JOB_ID);
@@ -512,7 +514,7 @@
}
} catch (RepositoryException re ) {
// something went wrong, so let's log it
- this.logger.error("Exception during writing new job '" + event + "' to repository at " + nodePath, re);
+ this.logger.error("Exception during writing new job '" + EventUtil.toString(event) + "' to repository at " + nodePath, re);
}
}
}
@@ -570,13 +572,17 @@
}
if ( info != null && this.running ) {
- logger.debug("Processing new job {}", info.event);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Received new job {}", EventUtil.toString(info.event));
+ }
// check for local only jobs and remove them from the queue if they're meant
// for another application node
final String appId = (String)info.event.getProperty(EventUtil.PROPERTY_APPLICATION);
if ( info.event.getProperty(EventUtil.PROPERTY_JOB_RUN_LOCAL) != null
&& appId != null && !this.applicationId.equals(appId) ) {
- logger.debug("Discarding job {} : local job for a different application node.", info.event);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Discarding job {} : local job for a different application node.", EventUtil.toString(info.event));
+ }
info = null;
}
@@ -584,7 +590,9 @@
if ( info != null && info.event.getProperty(EventUtil.PROPERTY_JOB_QUEUE_NAME) != null ) {
final String queueName = (String)info.event.getProperty(EventUtil.PROPERTY_JOB_QUEUE_NAME);
synchronized ( this.jobQueues ) {
- logger.debug("Queuing job {} into queue {}.", info.event, queueName);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Queuing job {} into queue {}.", EventUtil.toString(info.event), queueName);
+ }
BlockingQueue<EventInfo> jobQueue = this.jobQueues.get(queueName);
if ( jobQueue == null ) {
final boolean orderedQueue = info.event.getProperty(EventUtil.PROPERTY_JOB_QUEUE_ORDERED) != null;
@@ -699,7 +707,9 @@
boolean putback = false;
boolean wait = false;
synchronized (this.backgroundLock) {
- logger.debug("Executing job {}.", info.event);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Executing job {}.", EventUtil.toString(info.event));
+ }
try {
this.backgroundSession.refresh(false);
// check if the node still exists
@@ -738,7 +748,9 @@
}
// check number of parallel jobs for main queue
if ( process && jobQueue == null && this.parallelJobCount >= this.maximumParallelJobs ) {
- logger.debug("Rescheduling job {} - maximum parallel job count of {} reached!", info.event, this.maximumParallelJobs);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Rescheduling job {} - maximum parallel job count of {} reached!", EventUtil.toString(info.event), this.maximumParallelJobs);
+ }
process = false;
wait = true;
}
@@ -826,12 +838,16 @@
* @see org.osgi.service.event.EventHandler#handleEvent(org.osgi.service.event.Event)
*/
public void handleEvent(final Event event) {
- logger.debug("Receiving event {}", event);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Receiving event {}", EventUtil.toString(event));
+ }
// we ignore remote job events
if ( EventUtil.isLocal(event) ) {
// check for bundle event
if ( event.getTopic().equals(EventUtil.TOPIC_JOB)) {
- logger.debug("Handling local job {}", event);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Handling local job {}", EventUtil.toString(event));
+ }
// job event
final String jobTopic = (String)event.getProperty(EventUtil.PROPERTY_JOB_TOPIC);
@@ -956,7 +972,9 @@
final ParallelInfo parInfo = ParallelInfo.getParallelInfo(event);
final boolean parallelProcessing = parInfo.processParallel;
final String jobTopic = (String)event.getProperty(EventUtil.PROPERTY_JOB_TOPIC);
- logger.debug("Starting job {}", event);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Starting job {}", EventUtil.toString(event));
+ }
boolean unlock = true;
try {
if ( isMainQueue ) {
@@ -1231,6 +1249,9 @@
* @see org.apache.sling.event.EventUtil.JobStatusNotifier#finishedJob(org.osgi.service.event.Event, String, boolean)
*/
public boolean finishedJob(Event job, String eventNodePath, boolean shouldReschedule) {
+ if ( this.logger.isDebugEnabled() ) {
+ this.logger.debug("Received finish for job {}, shouldReschedule={}", EventUtil.toString(job), shouldReschedule);
+ }
// let's remove the event from our processing list
// this is just a sanity check, as usually the job should have been
// removed during sendAcknowledge.
@@ -1259,11 +1280,20 @@
newProperties.put(EventUtil.PROPERTY_JOB_RETRY_COUNT, retryCount);
newProperties.put(EventUtil.PROPERTY_JOB_RETRIES, retries);
job = new Event(job.getTopic(), newProperties);
- this.sendNotification(EventUtil.TOPIC_JOB_CANCELLED, job);
- } else {
+ if ( this.logger.isDebugEnabled() ) {
+ this.logger.debug("Failed job {}", EventUtil.toString(job));
+ }
this.sendNotification(EventUtil.TOPIC_JOB_FAILED, job);
+ } else {
+ if ( this.logger.isDebugEnabled() ) {
+ this.logger.debug("Cancelled job {}", EventUtil.toString(job));
+ }
+ this.sendNotification(EventUtil.TOPIC_JOB_CANCELLED, job);
}
} else {
+ if ( this.logger.isDebugEnabled() ) {
+ this.logger.debug("Finished job {}", EventUtil.toString(job));
+ }
this.sendNotification(EventUtil.TOPIC_JOB_FINISHED, job);
}
final ParallelInfo parInfo = ParallelInfo.getParallelInfo(job);
@@ -1401,7 +1431,9 @@
}
private void putBackIntoMainQueue(final EventInfo info, final boolean useSleepTime) {
- logger.debug("Putting job {} back into the queue.", info.event);
+ if ( logger.isDebugEnabled() ) {
+ logger.debug("Putting job {} back into the queue.", EventUtil.toString(info.event));
+ }
final Date fireDate = new Date();
if ( useSleepTime ) {
fireDate.setTime(System.currentTimeMillis() + this.sleepTime * 1000);