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/19 16:28:20 UTC
svn commit: r1632913 - in /sling/trunk/bundles/extensions/event/src:
main/java/org/apache/sling/event/impl/jobs/
main/java/org/apache/sling/event/impl/jobs/config/
main/java/org/apache/sling/event/impl/jobs/queues/
test/java/org/apache/sling/event/it/
Author: cziegeler
Date: Sun Oct 19 14:28:20 2014
New Revision: 1632913
URL: http://svn.apache.org/r1632913
Log:
SLING-4048 : Avoid keeping jobs in memory. Fix rescheduling bug when topology changes in between
Modified:
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/ParallelJobQueue.java
sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java
sling/trunk/bundles/extensions/event/src/test/java/org/apache/sling/event/it/ChaosTest.java
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java?rev=1632913&r1=1632912&r2=1632913&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/JobHandler.java Sun Oct 19 14:28:20 2014
@@ -250,8 +250,8 @@ public class JobHandler {
}
- public void removeFromRetryList() {
- this.configuration.removeJobFromRetryList(this.job);
+ public boolean removeFromRetryList() {
+ return this.configuration.removeJobFromRetryList(this.job);
}
@Override
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java?rev=1632913&r1=1632912&r2=1632913&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/config/JobManagerConfiguration.java Sun Oct 19 14:28:20 2014
@@ -536,9 +536,18 @@ public class JobManagerConfiguration imp
}
}
- public void removeJobFromRetryList(final Job job) {
+ public List<Job> clearJobRetryList() {
+ final List<Job> result = new ArrayList<Job>();
+ synchronized ( this.retryList ) {
+ result.addAll(retryList.values());
+ retryList.clear();
+ }
+ return result;
+ }
+
+ public boolean removeJobFromRetryList(final Job job) {
synchronized ( retryList ) {
- retryList.remove(job.getId());
+ return retryList.remove(job.getId()) != null;
}
}
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/ParallelJobQueue.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/ParallelJobQueue.java?rev=1632913&r1=1632912&r2=1632913&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/ParallelJobQueue.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/ParallelJobQueue.java Sun Oct 19 14:28:20 2014
@@ -126,8 +126,9 @@ public final class ParallelJobQueue exte
final Runnable t = new Runnable() {
@Override
public void run() {
- handler.removeFromRetryList();
- ParallelJobQueue.super.reschedule(handler);
+ if ( handler.removeFromRetryList() ) {
+ ParallelJobQueue.super.reschedule(handler);
+ }
}
};
services.scheduler.schedule(t, services.scheduler.AT(fireDate).name(jobName));
Modified: sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java?rev=1632913&r1=1632912&r2=1632913&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java (original)
+++ sling/trunk/bundles/extensions/event/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java Sun Oct 19 14:28:20 2014
@@ -41,6 +41,8 @@ import org.apache.sling.api.resource.Res
import org.apache.sling.commons.scheduler.Scheduler;
import org.apache.sling.commons.threads.ThreadPoolManager;
import org.apache.sling.event.impl.jobs.JobConsumerManager;
+import org.apache.sling.event.impl.jobs.JobHandler;
+import org.apache.sling.event.impl.jobs.JobImpl;
import org.apache.sling.event.impl.jobs.config.ConfigurationChangeListener;
import org.apache.sling.event.impl.jobs.config.InternalQueueConfiguration;
import org.apache.sling.event.impl.jobs.config.JobManagerConfiguration;
@@ -51,6 +53,7 @@ import org.apache.sling.event.impl.jobs.
import org.apache.sling.event.impl.jobs.stats.StatisticsManager;
import org.apache.sling.event.impl.support.Environment;
import org.apache.sling.event.impl.support.ResourceHelper;
+import org.apache.sling.event.jobs.Job;
import org.apache.sling.event.jobs.NotificationConstants;
import org.apache.sling.event.jobs.Queue;
import org.apache.sling.event.jobs.QueueConfiguration;
@@ -282,6 +285,11 @@ public class QueueManager
this.outdateQueue(queue);
}
}
+ final List<Job> rescheduleList = this.configuration.clearJobRetryList();
+ for(final Job j : rescheduleList) {
+ final JobHandler jh = new JobHandler((JobImpl)j, this.configuration);
+ jh.reschedule();
+ }
}
/**
@@ -328,6 +336,7 @@ public class QueueManager
@Override
public void configurationChanged(final boolean active) {
logger.debug("Topology changed {}", active);
+ this.isActive.set(active);
if ( active ) {
final Set<String> topics = this.initialScan();
final Map<QueueInfo, Set<String>> mapping = this.updateTopicMapping(topics);
@@ -335,9 +344,7 @@ public class QueueManager
for(final Map.Entry<QueueInfo, Set<String>> entry : mapping.entrySet() ) {
this.start(entry.getKey(), entry.getValue());
}
- this.isActive.set(true);
} else {
- this.isActive.set(false);
this.restart();
}
}
Modified: sling/trunk/bundles/extensions/event/src/test/java/org/apache/sling/event/it/ChaosTest.java
URL: http://svn.apache.org/viewvc/sling/trunk/bundles/extensions/event/src/test/java/org/apache/sling/event/it/ChaosTest.java?rev=1632913&r1=1632912&r2=1632913&view=diff
==============================================================================
--- sling/trunk/bundles/extensions/event/src/test/java/org/apache/sling/event/it/ChaosTest.java (original)
+++ sling/trunk/bundles/extensions/event/src/test/java/org/apache/sling/event/it/ChaosTest.java Sun Oct 19 14:28:20 2014
@@ -305,7 +305,7 @@ public class ChaosTest extends AbstractJ
}
}
- @Test(timeout=DURATION * 3000)
+ @Test(timeout=DURATION * 2000)
public void testDoChaos() throws Exception {
final JobManager jobManager = this.getJobManager();
@@ -333,13 +333,15 @@ public class ChaosTest extends AbstractJ
final List<Thread> threads = new ArrayList<Thread>();
final AtomicLong finishedThreads = new AtomicLong();
- final ServiceRegistration eventHandler = this.registerEventHandler(NotificationConstants.TOPIC_JOB_FINISHED,
+ final ServiceRegistration eventHandler = this.registerEventHandler("org/apache/sling/event/notification/job/*",
new EventHandler() {
@Override
public void handleEvent(final Event event) {
- final String topic = (String) event.getProperty(NotificationConstants.NOTIFICATION_PROPERTY_JOB_TOPIC);
- finished.get(topic).incrementAndGet();
+ if ( NotificationConstants.TOPIC_JOB_FINISHED.equals(event.getTopic())) {
+ final String topic = (String) event.getProperty(NotificationConstants.NOTIFICATION_PROPERTY_JOB_TOPIC);
+ finished.get(topic).incrementAndGet();
+ }
}
});
try {