code.onehippo.org is currently readonly. We are migrating to code.bloomreach.com, please continue working there on Monday 14/12. See: https://docs.bloomreach.com/display/engineering/GitLab

Commit 7e91d6c3 authored by Ate Douma's avatar Ate Douma

REPO-1874 Leverage LockManager for the RepositoryScheduler replacing the deprecated HippoLock usage

parent c64b8742
......@@ -52,10 +52,12 @@ public class RepositorySchedulerImpl implements RepositoryScheduler {
private final Session session;
private final JCRScheduler scheduler;
private final String moduleConfigPath;
RepositorySchedulerImpl(Session session, JCRScheduler scheduler) {
RepositorySchedulerImpl(final Session session, final JCRScheduler scheduler, final String moduleConfigPath) {
this.session = session;
this.scheduler = scheduler;
this.moduleConfigPath = moduleConfigPath;
}
@Override
......@@ -153,7 +155,7 @@ public class RepositorySchedulerImpl implements RepositoryScheduler {
private Node newJobNode(final RepositoryJobInfo info) throws RepositoryException {
final String name = info.getName();
final String group = info.getGroup();
final Node moduleConfig = session.getNode(SchedulerModule.getModuleConfigPath());
final Node moduleConfig = session.getNode(moduleConfigPath);
final Node jobGroup;
if (moduleConfig.hasNode(group)) {
jobGroup = moduleConfig.getNode(group);
......@@ -168,7 +170,7 @@ public class RepositorySchedulerImpl implements RepositoryScheduler {
private Node getJobNode(final String jobName, final String groupName) throws RepositoryException {
synchronized (session) {
final Node moduleConfig = session.getNode(SchedulerModule.getModuleConfigPath());
final Node moduleConfig = session.getNode(moduleConfigPath);
final Node groupNode = JcrUtils.getNodeIfExists(moduleConfig, getGroupName(groupName));
if (groupNode != null) {
return JcrUtils.getNodeIfExists(groupNode, jobName);
......
/*
* Copyright 2008-2013 Hippo B.V. (http://www.onehippo.com)
* Copyright 2008-2017 Hippo B.V. (http://www.onehippo.com)
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
......@@ -51,8 +51,6 @@ public class SchedulerModule implements DaemonModule, ConfigurableDaemonModule {
SCHEDULER_FACTORY_PROPERTIES.put(JcrSchedulerFactory.PROP_JOB_STORE_CLASS, JCRJobStore.class.getName());
}
private static SchedulerModule instance;
private Session session;
private JCRScheduler scheduler = null;
private RepositoryScheduler service;
......@@ -69,7 +67,6 @@ public class SchedulerModule implements DaemonModule, ConfigurableDaemonModule {
@Override
public void initialize(Session session) throws RepositoryException {
instance = this;
this.session = session;
try {
final JcrSchedulerFactory schedFactory = new JcrSchedulerFactory(SCHEDULER_FACTORY_PROPERTIES);
......@@ -84,18 +81,10 @@ public class SchedulerModule implements DaemonModule, ConfigurableDaemonModule {
log.error("Failed to initialize quartz scheduler", e);
return;
}
service = new RepositorySchedulerImpl(session, scheduler);
service = new RepositorySchedulerImpl(session, scheduler, moduleConfigPath);
HippoServiceRegistry.registerService(service, RepositoryScheduler.class);
}
static Session getSession() {
return instance.session;
}
static String getModuleConfigPath() {
return instance.moduleConfigPath;
}
@Override
public void shutdown() {
if (service != null) {
......@@ -112,13 +101,14 @@ public class SchedulerModule implements DaemonModule, ConfigurableDaemonModule {
private static final String PROP_THREAD_POOL_THREADCOUNT = "org.quartz.threadPool.threadCount";
private static final String PROP_THREAD_POOL_THREADPRIORITY = "org.quartz.threadPool.threadPriority";
public JcrSchedulerFactory(Properties properties) throws SchedulerException {
public JcrSchedulerFactory(final Properties properties) throws SchedulerException {
super(properties);
}
@Override
protected Scheduler instantiate(QuartzSchedulerResources rcs, QuartzScheduler qs) {
try {
((JCRJobStore)rcs.getJobStore()).init(session, moduleConfigPath);
qs.getSchedulerContext().put(Session.class.getName(), session);
return new JCRScheduler(qs);
} catch (SchedulerException e) {
......
/*
* Copyright 2012-2013 Hippo B.V. (http://www.onehippo.com)
* Copyright 2012-2017 Hippo B.V. (http://www.onehippo.com)
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
......@@ -40,7 +40,6 @@ import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertFalse;
import static junit.framework.Assert.assertNotNull;
import static junit.framework.Assert.assertTrue;
import static junit.framework.Assert.fail;
import static org.hippoecm.repository.quartz.HippoSchedJcrConstants.HIPPOSCHED_JOBGROUP;
import static org.hippoecm.repository.quartz.HippoSchedJcrConstants.HIPPOSCHED_NEXTFIRETIME;
import static org.hippoecm.repository.quartz.HippoSchedJcrConstants.HIPPOSCHED_REPOSITORY_JOB;
......@@ -66,13 +65,22 @@ public class JCRJobStoreTest extends RepositoryTestCase {
session.save();
session.getWorkspace().getObservationManager().addEventListener(listener, ALL_EVENTS, "/test", true, null, null, false);
storeSession = session.impersonate(new SimpleCredentials("admin", new char[]{}));
store = new JCRJobStore(10, storeSession, "/test");
store = new JCRJobStore();
store.init(storeSession, "/test");
store.initialize(null, null);
}
@Override
public void tearDown() throws Exception {
store.shutdown();
// give store executor max 1 sec. to terminate running threads
for (int i = 0; i < 10; i++) {
if (!store.isTerminated()) {
Thread.sleep(100);
} else {
break;
}
}
storeSession.logout();
session.refresh(false);
super.tearDown();
......@@ -94,7 +102,7 @@ public class JCRJobStoreTest extends RepositoryTestCase {
assertNotNull(triggers);
assertFalse(triggers.isEmpty());
assertEquals(1, triggers.size());
assertTrue(jobNode.getNode("hipposched:triggers/trigger").isLocked());
assertTrue(store.isLocked(jobNode, jobNode.getNode("hipposched:triggers/trigger")));
assertFalse(listener.hasTriggerUpdateEvents);
}
......@@ -106,7 +114,7 @@ public class JCRJobStoreTest extends RepositoryTestCase {
assumeNotNull(triggers);
assumeTrue(!triggers.isEmpty());
store.releaseAcquiredTrigger(triggers.get(0));
assertFalse(jobNode.getNode("hipposched:triggers/trigger").isLocked());
assertFalse(store.isLocked(jobNode, jobNode.getNode("hipposched:triggers/trigger")));
assertFalse(listener.hasTriggerUpdateEvents);
}
......@@ -162,29 +170,13 @@ public class JCRJobStoreTest extends RepositoryTestCase {
}
@Test
public void testTriggerLockKeepAlive() throws Exception {
public void testAquireTriggerKeepsItLocked() throws Exception {
final Node jobNode = createAndStoreJobAndSimpleTrigger(store);
final List<OperableTrigger> triggers = store.acquireNextTriggers(System.currentTimeMillis(), 1, -1l);
Thread.sleep(1000*12); // sleep longer than lock timeout
assertTrue(jobNode.getNode("hipposched:triggers/trigger").isLocked());
Thread.sleep(1000); // sleep for a bit time, just to make sure
assertTrue(store.isLocked(jobNode, jobNode.getNode("hipposched:triggers/trigger")));
store.releaseAcquiredTrigger(triggers.get(0));
assertFalse(jobNode.getNode("hipposched:triggers/trigger").isLocked());
}
@Test
public void testTriggerLockKeepAliveIsCancelledWhenRefreshingLockFails() throws Exception {
final Node jobNode = createAndStoreJobAndSimpleTrigger(store);
final List<OperableTrigger> triggers = store.acquireNextTriggers(System.currentTimeMillis(), 1, -1l);
final Node triggerNode = jobNode.getNode("hipposched:triggers/trigger");
triggerNode.unlock();
Thread.sleep(1000*12); // sleep longer than twice the refresh interval
for (int i = 0; i < 10; i++) {
if (!store.getLockKeepAlives().containsKey(triggerNode.getIdentifier())) {
return;
}
Thread.sleep(1000l);
}
fail("Keep alive was not cancelled");
assertFalse(store.isLocked(jobNode, jobNode.getNode("hipposched:triggers/trigger")));
}
@Test
......
......@@ -32,6 +32,7 @@ import org.onehippo.repository.scheduling.RepositoryJobSimpleTrigger;
import org.onehippo.repository.scheduling.RepositoryJobTrigger;
import org.onehippo.repository.scheduling.RepositoryScheduler;
import org.onehippo.repository.testutils.RepositoryTestCase;
import org.onehippo.testutils.log4j.Log4jInterceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -96,8 +97,10 @@ public class SchedulerTest extends RepositoryTestCase {
final RepositoryJobTrigger testJobTrigger = new RepositoryJobSimpleTrigger("test", new Date(System.currentTimeMillis() + 2500));
scheduler.scheduleJob(testJobInfo, testJobTrigger);
scheduler.deleteJob(testJobInfo.getName(), testJobInfo.getGroup());
if (waitUntilExecuted()) {
fail("Deleted job was still executed.");
try (Log4jInterceptor ignore = Log4jInterceptor.onError().deny(JCRJobStore.class).build()) {
if (waitUntilExecuted()) {
fail("Deleted job was still executed.");
}
}
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment