You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lens.apache.org by pr...@apache.org on 2016/07/25 12:06:28 UTC
[1/3] lens git commit: LENS-128: SchedulerServiceImpl for scheduler.
Repository: lens
Updated Branches:
refs/heads/master acb32d540 -> 1a96948ee
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/test/java/org/apache/lens/server/scheduler/AlarmServiceTest.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/scheduler/AlarmServiceTest.java b/lens-server/src/test/java/org/apache/lens/server/scheduler/AlarmServiceTest.java
new file mode 100644
index 0000000..7b610de
--- /dev/null
+++ b/lens-server/src/test/java/org/apache/lens/server/scheduler/AlarmServiceTest.java
@@ -0,0 +1,177 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.server.scheduler;
+
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+import java.util.*;
+
+import org.apache.lens.api.scheduler.SchedulerJobHandle;
+import org.apache.lens.api.scheduler.XFrequency;
+import org.apache.lens.api.scheduler.XFrequencyEnum;
+import org.apache.lens.server.EventServiceImpl;
+import org.apache.lens.server.LensServerConf;
+import org.apache.lens.server.LensServices;
+import org.apache.lens.server.api.LensConfConstants;
+import org.apache.lens.server.api.error.LensException;
+import org.apache.lens.server.api.events.LensEventListener;
+import org.apache.lens.server.api.events.LensEventService;
+import org.apache.lens.server.api.events.SchedulerAlarmEvent;
+
+import org.joda.time.DateTime;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+import org.quartz.*;
+import org.quartz.impl.StdSchedulerFactory;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.Test;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Tests for AlarmService.
+ */
+@Slf4j
+@Test(groups = "unit-test")
+public class AlarmServiceTest {
+
+ private static List<SchedulerAlarmEvent> events = new LinkedList<>();
+ private static volatile int counter = 0;
+ private final LensEventListener<SchedulerAlarmEvent> alarmEventListener;
+ private AlarmService alarmService;
+ private EventServiceImpl eventService;
+
+ {
+ alarmEventListener = new LensEventListener<SchedulerAlarmEvent>() {
+
+ @Override
+ public void onEvent(SchedulerAlarmEvent event) {
+ events.add(event);
+ }
+ };
+ }
+
+ @BeforeMethod
+ public void initializeEventsList() {
+ events = new LinkedList<>();
+ }
+
+ @BeforeTest
+ public void setUp() {
+ System.setProperty(LensConfConstants.CONFIG_LOCATION, "target/test-classes/");
+ LensServices.get().init(LensServerConf.getHiveConf());
+ LensServices.get().start();
+ eventService = LensServices.get().getService(LensEventService.NAME);
+ assertNotNull(eventService);
+ alarmService = LensServices.get().getService(AlarmService.NAME);
+ assertNotNull(alarmService);
+ eventService.addListenerForType(alarmEventListener, SchedulerAlarmEvent.class);
+ }
+
+ /**
+ * This test generally tests the basic understanding and assumptions about quartz framework with a sample job.
+ *
+ * @throws SchedulerException
+ * @throws InterruptedException
+ */
+ @Test
+ public void testCoreExecution() throws SchedulerException, InterruptedException {
+ SchedulerFactory sf = new StdSchedulerFactory();
+ Scheduler scheduler = sf.getScheduler();
+ scheduler.start();
+ // prepare a sample Job
+ JobDetail job = JobBuilder.newJob(TestJob.class).withIdentity("random", "group").build();
+ scheduler.scheduleJob(job, getPastPerSecondsTrigger());
+ Thread.sleep(2000);
+ Assert.assertEquals(counter, 10);
+ scheduler.deleteJob(job.getKey());
+ }
+
+ private CalendarIntervalScheduleBuilder getPerSecondCalendar() {
+ return CalendarIntervalScheduleBuilder.calendarIntervalSchedule().withInterval(1, DateBuilder.IntervalUnit.SECOND)
+ .withMisfireHandlingInstructionFireAndProceed();
+ }
+
+ private Trigger getPastPerSecondsTrigger() {
+ CalendarIntervalScheduleBuilder scheduleBuilder = getPerSecondCalendar();
+ Date start = new Date();
+ start = new Date(start.getTime() - 10000);
+ Date end = new Date();
+
+ return TriggerBuilder.newTrigger().withIdentity("trigger1", "group1").startAt(start).endAt(end)
+ .withSchedule(scheduleBuilder).build();
+ }
+
+ @Test
+ public void testAlarmServiceEnums() throws InterruptedException, LensException {
+ DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-mm-dd");
+ DateTime start = formatter.parseDateTime("2016-03-03");
+ DateTime end = formatter.parseDateTime("2016-03-06");
+ SchedulerJobHandle jobHandle = new SchedulerJobHandle(UUID.randomUUID());
+ XFrequency frequency = new XFrequency();
+ frequency.setEnum(XFrequencyEnum.DAILY);
+ alarmService.schedule(start, end, frequency, jobHandle.toString());
+ Thread.sleep(2000); // give chance to the event listener to process the data
+ int count = 0;
+ for (SchedulerAlarmEvent event : events) {
+ if (event.getJobHandle().equals(jobHandle)) {
+ count++;
+ }
+ }
+ // 3 scheduled events, and one expired event.
+ Assert.assertEquals(count, 4);
+ DateTime expectedDate = start;
+ Set<DateTime> actualSet = new HashSet<>();
+ for (SchedulerAlarmEvent e : events) {
+ actualSet.add(e.getNominalTime());
+ }
+
+ for (int i = 0; i < actualSet.size(); i++) {
+ Assert.assertTrue(actualSet.contains(expectedDate));
+ expectedDate = expectedDate.plusDays(1);
+ }
+ }
+
+ @Test
+ public void testAlarmServiceCronExpressions() throws InterruptedException, LensException {
+ DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-mm-dd");
+ DateTime start = formatter.parseDateTime("2016-03-03");
+ DateTime end = formatter.parseDateTime("2016-03-06");
+ SchedulerJobHandle jobHandle = new SchedulerJobHandle(UUID.randomUUID());
+ System.out.println("jobHandle = " + jobHandle);
+ XFrequency frequency = new XFrequency();
+ frequency.setCronExpression("0/1 * * * * ?");
+ alarmService.schedule(start, end, frequency, jobHandle.toString());
+ Thread.sleep(2000);
+ // Assert that the events are fired and at per second interval.
+ assertTrue(events.size() > 1);
+ }
+
+ @PersistJobDataAfterExecution
+ @DisallowConcurrentExecution
+ public static class TestJob implements Job {
+ @Override
+ public void execute(JobExecutionContext context) throws JobExecutionException {
+ AlarmServiceTest.counter++;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerDAOTest.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerDAOTest.java b/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerDAOTest.java
index d76a586..16a6c78 100644
--- a/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerDAOTest.java
+++ b/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerDAOTest.java
@@ -25,10 +25,11 @@ import javax.xml.datatype.DatatypeFactory;
import javax.xml.datatype.XMLGregorianCalendar;
import org.apache.lens.api.LensSessionHandle;
+import org.apache.lens.api.query.QueryHandle;
import org.apache.lens.api.scheduler.*;
import org.apache.lens.server.LensServerConf;
import org.apache.lens.server.api.LensConfConstants;
-import org.apache.lens.server.scheduler.util.UtilityMethods;
+import org.apache.lens.server.util.UtilityMethods;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.hadoop.conf.Configuration;
@@ -45,12 +46,13 @@ public class SchedulerDAOTest {
SchedulerDAO schedulerDAO;
Map<SchedulerJobInstanceHandle, SchedulerJobInstanceInfo> instances = new HashMap<>();
+ SchedulerJobHandle jobHandle = null;
@BeforeClass
public void setup() throws Exception {
System.setProperty(LensConfConstants.CONFIG_LOCATION, "target/test-classes/");
Configuration conf = LensServerConf.getHiveConf();
- QueryRunner runner = new QueryRunner(UtilityMethods.getDataSourceFromConf(conf));
+ QueryRunner runner = new QueryRunner(UtilityMethods.getDataSourceFromConfForScheduler(conf));
// Cleanup all tables
runner.update("DROP TABLE IF EXISTS job_table");
runner.update("DROP TABLE IF EXISTS job_instance_table");
@@ -95,12 +97,13 @@ public class SchedulerDAOTest {
return job;
}
- @Test
+ @Test(priority = 1)
public void testStoreJob() throws Exception {
XJob job = getTestJob();
long currentTime = System.currentTimeMillis();
- SchedulerJobInfo info = new SchedulerJobInfo(SchedulerJobHandle.fromString(UUID.randomUUID().toString()), job,
- "lens", SchedulerJobStatus.NEW, currentTime, currentTime);
+ jobHandle = new SchedulerJobHandle(UUID.randomUUID());
+ SchedulerJobInfo info = new SchedulerJobInfo(jobHandle, job, "lens", SchedulerJobState.NEW, currentTime,
+ currentTime);
// Store the job
schedulerDAO.storeJob(info);
// Retrive the stored job
@@ -108,43 +111,49 @@ public class SchedulerDAOTest {
Assert.assertEquals(job, outJob);
}
- @Test
+ @Test(priority = 2)
public void testStoreInstance() throws Exception {
long currentTime = System.currentTimeMillis();
- SchedulerJobHandle jobHandle = SchedulerJobHandle.fromString(UUID.randomUUID().toString());
- SchedulerJobInstanceInfo firstInstance = new SchedulerJobInstanceInfo(
- SchedulerJobInstanceHandle.fromString(UUID.randomUUID().toString()), jobHandle,
+ SchedulerJobInstanceHandle instanceHandle = new SchedulerJobInstanceHandle(UUID.randomUUID());
+ SchedulerJobInstanceInfo firstInstance = new SchedulerJobInstanceInfo(instanceHandle, jobHandle, currentTime,
+ new ArrayList<SchedulerJobInstanceRun>());
+ SchedulerJobInstanceRun run1 = new SchedulerJobInstanceRun(instanceHandle, 1,
new LensSessionHandle(UUID.randomUUID(), UUID.randomUUID()), currentTime, currentTime, "/tmp/",
- "select * form yoda_cube", SchedulerJobInstanceStatus.WAITING, currentTime);
+ QueryHandle.fromString(UUID.randomUUID().toString()), SchedulerJobInstanceState.WAITING);
instances.put(firstInstance.getId(), firstInstance);
schedulerDAO.storeJobInstance(firstInstance);
+ schedulerDAO.storeJobInstanceRun(run1);
+ // Put run in the instance
+ firstInstance.getInstanceRunList().add(run1);
currentTime = System.currentTimeMillis();
- SchedulerJobInstanceInfo secondInstance = new SchedulerJobInstanceInfo(
- SchedulerJobInstanceHandle.fromString(UUID.randomUUID().toString()), jobHandle,
+ instanceHandle = new SchedulerJobInstanceHandle(UUID.randomUUID());
+ SchedulerJobInstanceInfo secondInstance = new SchedulerJobInstanceInfo(instanceHandle, jobHandle, currentTime,
+ new ArrayList<SchedulerJobInstanceRun>());
+ SchedulerJobInstanceRun run2 = new SchedulerJobInstanceRun(instanceHandle, 1,
new LensSessionHandle(UUID.randomUUID(), UUID.randomUUID()), currentTime, currentTime, "/tmp/",
- "select * form yoda_cube", SchedulerJobInstanceStatus.WAITING, currentTime);
+ QueryHandle.fromString(UUID.randomUUID().toString()), SchedulerJobInstanceState.WAITING);
instances.put(secondInstance.getId(), secondInstance);
schedulerDAO.storeJobInstance(secondInstance);
- List<SchedulerJobInstanceHandle> handleList = schedulerDAO.getJobInstances(jobHandle);
+ schedulerDAO.storeJobInstanceRun(run2);
+ secondInstance.getInstanceRunList().add(run2);
+
+ List<SchedulerJobInstanceInfo> handleList = schedulerDAO.getJobInstances(jobHandle);
// Size should be 2
Assert.assertEquals(handleList.size(), 2);
// Get the definition of instance from the store.
- SchedulerJobInstanceInfo instance1 = schedulerDAO.getSchedulerJobInstanceInfo(handleList.get(0));
- Assert.assertEquals(instances.get(handleList.get(0)), instance1);
+ SchedulerJobInstanceInfo instance1 = handleList.get(0);
+ Assert.assertEquals(instances.get(handleList.get(0).getId()), instance1);
- SchedulerJobInstanceInfo instance2 = schedulerDAO.getSchedulerJobInstanceInfo(handleList.get(1));
- Assert.assertEquals(instances.get(handleList.get(1)), instance2);
+ SchedulerJobInstanceInfo instance2 = handleList.get(1);
+ Assert.assertEquals(instances.get(handleList.get(1).getId()), instance2);
}
- @Test(dependsOnMethods = { "testStoreJob" })
+ @Test(priority = 2)
public void testUpdateJob() throws Exception {
// Get all the stored jobs.
// update one and check if it successful.
- List<SchedulerJobHandle> jobHandles = schedulerDAO.getJobs(null, null, null, null);
- Assert.assertEquals(jobHandles.size() > 0, true);
-
- SchedulerJobInfo jobInfo = schedulerDAO.getSchedulerJobInfo(jobHandles.get(0));
+ SchedulerJobInfo jobInfo = schedulerDAO.getSchedulerJobInfo(jobHandle);
XJob newJob = getTestJob();
jobInfo.setJob(newJob);
schedulerDAO.updateJob(jobInfo);
@@ -152,37 +161,38 @@ public class SchedulerDAOTest {
XJob storedJob = schedulerDAO.getJob(jobInfo.getId());
Assert.assertEquals(storedJob, newJob);
- // Change State
- jobInfo.setState(SchedulerJobStatus.SCHEDULED);
- schedulerDAO.updateJobState(jobInfo);
- Assert.assertEquals(schedulerDAO.getJobState(jobInfo.getId()), SchedulerJobStatus.SCHEDULED);
+ // Change SchedulerJobInstanceState
+ jobInfo.setJobState(jobInfo.getJobState().nextTransition(SchedulerJobEvent.ON_SCHEDULE));
+ schedulerDAO.updateJobStatus(jobInfo);
+ Assert.assertEquals(schedulerDAO.getJobState(jobInfo.getId()), SchedulerJobState.SCHEDULED);
}
- @Test(dependsOnMethods = { "testStoreInstance" })
+ @Test(priority = 3)
public void testUpdateJobInstance() {
SchedulerJobInstanceHandle handle = instances.keySet().iterator().next();
SchedulerJobInstanceInfo info = instances.get(handle);
- info.setState(SchedulerJobInstanceStatus.LAUNCHED);
- schedulerDAO.updateJobInstanceState(info);
+ SchedulerJobInstanceRun run = info.getInstanceRunList().get(0);
+ run.setInstanceState(SchedulerJobInstanceState.LAUNCHED);
+ schedulerDAO.updateJobInstanceRun(run);
// Get the instance
Assert.assertEquals(schedulerDAO.getSchedulerJobInstanceInfo(handle), info);
}
- @Test(dependsOnMethods = { "testUpdateJob" })
+ @Test(priority = 3)
public void testSearchStoreJob() throws Exception {
// Store more jobs with the one user and search
XJob job = getTestJob();
long currentTime = System.currentTimeMillis();
SchedulerJobInfo info = new SchedulerJobInfo(SchedulerJobHandle.fromString(UUID.randomUUID().toString()), job,
- "lens", SchedulerJobStatus.NEW, currentTime, currentTime);
+ "lens", SchedulerJobState.NEW, currentTime, currentTime);
// Store the job
schedulerDAO.storeJob(info);
info = new SchedulerJobInfo(SchedulerJobHandle.fromString(UUID.randomUUID().toString()), job, "lens",
- SchedulerJobStatus.NEW, currentTime, currentTime);
+ SchedulerJobState.NEW, currentTime, currentTime);
schedulerDAO.storeJob(info);
// There should be 3 jobs till now.
Assert.assertEquals(schedulerDAO.getJobs("lens", null, null, null).size(), 3);
- Assert.assertEquals(schedulerDAO.getJobs("lens", SchedulerJobStatus.NEW, 1L, System.currentTimeMillis()).size(), 2);
- Assert.assertEquals(schedulerDAO.getJobs("Alice", SchedulerJobStatus.NEW, null, null).size(), 0);
+ Assert.assertEquals(schedulerDAO.getJobs("lens", SchedulerJobState.NEW, 1L, System.currentTimeMillis()).size(), 2);
+ Assert.assertEquals(schedulerDAO.getJobs("Alice", SchedulerJobState.NEW, null, null).size(), 0);
}
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/test/java/org/apache/lens/server/scheduler/TestSchedulerServiceImpl.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/scheduler/TestSchedulerServiceImpl.java b/lens-server/src/test/java/org/apache/lens/server/scheduler/TestSchedulerServiceImpl.java
new file mode 100644
index 0000000..ce744af
--- /dev/null
+++ b/lens-server/src/test/java/org/apache/lens/server/scheduler/TestSchedulerServiceImpl.java
@@ -0,0 +1,236 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.server.scheduler;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+
+import java.util.GregorianCalendar;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+
+import javax.xml.datatype.DatatypeFactory;
+import javax.xml.datatype.XMLGregorianCalendar;
+
+import org.apache.lens.api.LensConf;
+import org.apache.lens.api.LensSessionHandle;
+import org.apache.lens.api.query.QueryHandle;
+import org.apache.lens.api.query.QueryStatus;
+import org.apache.lens.api.scheduler.*;
+import org.apache.lens.server.EventServiceImpl;
+import org.apache.lens.server.LensServerConf;
+import org.apache.lens.server.LensServices;
+import org.apache.lens.server.api.LensConfConstants;
+import org.apache.lens.server.api.query.QueryContext;
+import org.apache.lens.server.api.query.QueryEnded;
+import org.apache.lens.server.api.query.QueryExecutionService;
+import org.apache.lens.server.api.scheduler.SchedulerService;
+import org.apache.lens.server.query.QueryExecutionServiceImpl;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.powermock.api.mockito.PowerMockito;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Test(groups = "unit-test")
+public class TestSchedulerServiceImpl {
+
+ SchedulerServiceImpl scheduler;
+ EventServiceImpl eventService;
+ LensSessionHandle sessionHandle = null;
+
+ @BeforeMethod
+ public void setup() throws Exception {
+ System.setProperty(LensConfConstants.CONFIG_LOCATION, "target/test-classes/");
+ }
+
+ private void setupQueryService() throws Exception {
+ QueryExecutionService queryExecutionService = PowerMockito.mock(QueryExecutionService.class);
+ scheduler.setQueryService(queryExecutionService);
+ PowerMockito.when(scheduler.getQueryService()
+ .estimate(anyString(), any(LensSessionHandle.class), anyString(), any(LensConf.class))).thenReturn(null);
+ PowerMockito.when(scheduler.getQueryService()
+ .executeAsync(any(LensSessionHandle.class), anyString(), any(LensConf.class), anyString()))
+ .thenReturn(new QueryHandle(UUID.randomUUID()));
+ PowerMockito.when(scheduler.getQueryService().cancelQuery(any(LensSessionHandle.class), any(QueryHandle.class)))
+ .thenReturn(true);
+ scheduler.getSchedulerEventListener().setQueryService(queryExecutionService);
+ }
+
+ private QueryEnded mockQueryEnded(SchedulerJobInstanceHandle instanceHandle, QueryStatus.Status status) {
+ QueryContext mockContext = PowerMockito.mock(QueryContext.class);
+ PowerMockito.when(mockContext.getDriverResultPath()).thenReturn("/tmp/query1/result");
+ Configuration conf = new Configuration();
+ // set the instance handle
+ conf.set("job_instance_key", instanceHandle.getHandleIdString());
+ PowerMockito.when(mockContext.getConf()).thenReturn(conf);
+ // Get the queryHandle.
+ PowerMockito.when(mockContext.getQueryHandle()).thenReturn(new QueryHandle(UUID.randomUUID()));
+ QueryEnded queryEnded = PowerMockito.mock(QueryEnded.class);
+ PowerMockito.when(queryEnded.getQueryContext()).thenReturn(mockContext);
+ PowerMockito.when(queryEnded.getCurrentValue()).thenReturn(status);
+ return queryEnded;
+ }
+
+ @Test(priority = 1)
+ public void testScheduler() throws Exception {
+ LensServices.get().init(LensServerConf.getHiveConf());
+ LensServices.get().start();
+ scheduler = LensServices.get().getService(SchedulerService.NAME);
+ eventService = LensServices.get().getService(EventServiceImpl.NAME);
+ setupQueryService();
+ sessionHandle = ((QueryExecutionServiceImpl) LensServices.get().getService(QueryExecutionService.NAME))
+ .openSession("someuser", "test", new HashMap<String, String>(), false);
+ long currentTime = System.currentTimeMillis();
+ XJob job = getTestJob("0/5 * * * * ?", currentTime, currentTime + 15000);
+ SchedulerJobHandle jobHandle = scheduler.submitAndScheduleJob(sessionHandle, job);
+ Assert.assertNotNull(jobHandle);
+ Assert.assertEquals(scheduler.getSchedulerDAO().getJobState(jobHandle), SchedulerJobState.SCHEDULED);
+ // Wait for job to finish
+ Thread.sleep(30000);
+ List<SchedulerJobInstanceInfo> instanceHandleList = scheduler.getSchedulerDAO().getJobInstances(jobHandle);
+ Assert.assertEquals(instanceHandleList.size() >= 3, true);
+ Assert.assertEquals(scheduler.getSchedulerDAO().getJobState(jobHandle), SchedulerJobState.EXPIRED);
+ // SuccessFul query
+ eventService.notifyEvent(mockQueryEnded(instanceHandleList.get(0).getId(), QueryStatus.Status.SUCCESSFUL));
+ // Wait, for event to get processed
+ Thread.sleep(2000);
+ // Check the instance value
+ SchedulerJobInstanceInfo info = scheduler.getSchedulerDAO()
+ .getSchedulerJobInstanceInfo(instanceHandleList.get(0).getId());
+ Assert.assertEquals(info.getInstanceRunList().size(), 1);
+ Assert.assertEquals(info.getInstanceRunList().get(0).getResultPath(), "/tmp/query1/result");
+ Assert.assertEquals(info.getInstanceRunList().get(0).getInstanceState(), SchedulerJobInstanceState.SUCCEEDED);
+ }
+
+ @Test(priority = 2)
+ public void testSuspendResume() throws Exception {
+ long currentTime = System.currentTimeMillis();
+ XJob job = getTestJob("0/10 * * * * ?", currentTime, currentTime + 180000);
+ SchedulerJobHandle jobHandle = scheduler.submitAndScheduleJob(sessionHandle, job);
+ Assert.assertNotNull(jobHandle);
+ Assert.assertTrue(scheduler.suspendJob(sessionHandle, jobHandle));
+ Assert.assertEquals(scheduler.getSchedulerDAO().getJobState(jobHandle), SchedulerJobState.SUSPENDED);
+ Assert.assertTrue(scheduler.resumeJob(sessionHandle, jobHandle));
+ Assert.assertEquals(scheduler.getSchedulerDAO().getJobState(jobHandle), SchedulerJobState.SCHEDULED);
+ Thread.sleep(10000);
+ Assert.assertTrue(scheduler.expireJob(sessionHandle, jobHandle));
+ Assert.assertEquals(scheduler.getSchedulerDAO().getJobState(jobHandle), SchedulerJobState.EXPIRED);
+ }
+
+ @Test(priority = 2)
+ public void testRerunInstance() throws Exception {
+ long currentTime = System.currentTimeMillis();
+
+ XJob job = getTestJob("0/10 * * * * ?", currentTime, currentTime + 180000);
+ SchedulerJobHandle jobHandle = scheduler.submitAndScheduleJob(sessionHandle, job);
+ // Wait for some instances.
+ Thread.sleep(15000);
+ List<SchedulerJobInstanceInfo> instanceHandleList = scheduler.getSchedulerDAO().getJobInstances(jobHandle);
+ // Mark fail
+ eventService.notifyEvent(mockQueryEnded(instanceHandleList.get(0).getId(), QueryStatus.Status.FAILED));
+ Thread.sleep(1000);
+ SchedulerJobInstanceInfo info = scheduler.getSchedulerDAO()
+ .getSchedulerJobInstanceInfo(instanceHandleList.get(0).getId());
+ // First run
+ Assert.assertEquals(info.getInstanceRunList().size(), 1);
+ Assert.assertEquals(info.getInstanceRunList().get(0).getInstanceState(), SchedulerJobInstanceState.FAILED);
+
+ // Rerun
+ Assert.assertTrue(scheduler.rerunInstance(sessionHandle, instanceHandleList.get(0).getId()));
+ Thread.sleep(5000);
+ eventService.notifyEvent(mockQueryEnded(instanceHandleList.get(0).getId(), QueryStatus.Status.SUCCESSFUL));
+ Thread.sleep(1000);
+ info = scheduler.getSchedulerDAO().getSchedulerJobInstanceInfo(instanceHandleList.get(0).getId());
+ // There should be 2 reruns.
+ Assert.assertEquals(info.getInstanceRunList().size(), 2);
+ Assert.assertEquals(info.getInstanceRunList().get(1).getResultPath(), "/tmp/query1/result");
+ Assert.assertEquals(info.getInstanceRunList().get(1).getInstanceState(), SchedulerJobInstanceState.SUCCEEDED);
+ Assert.assertTrue(scheduler.expireJob(sessionHandle, jobHandle));
+ Assert.assertEquals(scheduler.getSchedulerDAO().getJobState(jobHandle), SchedulerJobState.EXPIRED);
+ }
+
+ @Test(priority = 2)
+ public void testKillRunningInstance() throws Exception {
+ long currentTime = System.currentTimeMillis();
+
+ XJob job = getTestJob("0/5 * * * * ?", currentTime, currentTime + 180000);
+ SchedulerJobHandle jobHandle = scheduler.submitAndScheduleJob(sessionHandle, job);
+ // Let it run
+ Thread.sleep(6000);
+ List<SchedulerJobInstanceInfo> instanceHandleList = scheduler.getSchedulerDAO().getJobInstances(jobHandle);
+ Assert.assertTrue(scheduler.killInstance(sessionHandle, instanceHandleList.get(0).getId()));
+ Thread.sleep(2000);
+ SchedulerJobInstanceInfo info = scheduler.getSchedulerDAO()
+ .getSchedulerJobInstanceInfo(instanceHandleList.get(0).getId());
+ Assert.assertEquals(info.getInstanceRunList().size(), 1);
+ Assert.assertEquals(info.getInstanceRunList().get(0).getInstanceState(), SchedulerJobInstanceState.RUNNING);
+ // Query End event
+ eventService.notifyEvent(mockQueryEnded(instanceHandleList.get(0).getId(), QueryStatus.Status.CANCELED));
+ Thread.sleep(2000);
+ info = scheduler.getSchedulerDAO().getSchedulerJobInstanceInfo(instanceHandleList.get(0).getId());
+ Assert.assertEquals(info.getInstanceRunList().get(0).getInstanceState(), SchedulerJobInstanceState.KILLED);
+ Assert.assertTrue(scheduler.expireJob(sessionHandle, jobHandle));
+ Assert.assertEquals(scheduler.getSchedulerDAO().getJobState(jobHandle), SchedulerJobState.EXPIRED);
+ }
+
+ private XTrigger getTestTrigger(String cron) {
+ XTrigger trigger = new XTrigger();
+ XFrequency frequency = new XFrequency();
+ frequency.setCronExpression(cron);
+ frequency.setTimezone("UTC");
+ trigger.setFrequency(frequency);
+ return trigger;
+ }
+
+ private XExecution getTestExecution() {
+ XExecution execution = new XExecution();
+ XJobQuery query = new XJobQuery();
+ query.setQuery("select ID from test_table");
+ execution.setQuery(query);
+ XSessionType sessionType = new XSessionType();
+ sessionType.setDb("test");
+ execution.setSession(sessionType);
+ return execution;
+ }
+
+ private XJob getTestJob(String cron, long start, long end) throws Exception {
+ XJob job = new XJob();
+ job.setTrigger(getTestTrigger(cron));
+ job.setName("Test lens Job");
+ GregorianCalendar startTime = new GregorianCalendar();
+ startTime.setTimeInMillis(start);
+ XMLGregorianCalendar startCal = DatatypeFactory.newInstance().newXMLGregorianCalendar(startTime);
+
+ GregorianCalendar endTime = new GregorianCalendar();
+ endTime.setTimeInMillis(end);
+ XMLGregorianCalendar endCal = DatatypeFactory.newInstance().newXMLGregorianCalendar(endTime);
+
+ job.setStartTime(startCal);
+ job.setEndTime(endCal);
+ job.setExecution(getTestExecution());
+ return job;
+ }
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/test/java/org/apache/lens/server/scheduler/notification/services/AlarmServiceTest.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/scheduler/notification/services/AlarmServiceTest.java b/lens-server/src/test/java/org/apache/lens/server/scheduler/notification/services/AlarmServiceTest.java
deleted file mode 100644
index 06883ae..0000000
--- a/lens-server/src/test/java/org/apache/lens/server/scheduler/notification/services/AlarmServiceTest.java
+++ /dev/null
@@ -1,180 +0,0 @@
-/**
- * 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.lens.server.scheduler.notification.services;
-
-import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
-
-import java.util.*;
-
-import org.apache.lens.api.scheduler.SchedulerJobHandle;
-import org.apache.lens.api.scheduler.XFrequency;
-import org.apache.lens.api.scheduler.XFrequencyEnum;
-import org.apache.lens.server.EventServiceImpl;
-import org.apache.lens.server.LensServerConf;
-import org.apache.lens.server.LensServices;
-import org.apache.lens.server.api.LensConfConstants;
-import org.apache.lens.server.api.error.LensException;
-import org.apache.lens.server.api.events.LensEventListener;
-import org.apache.lens.server.api.events.LensEventService;
-import org.apache.lens.server.api.events.SchedulerAlarmEvent;
-
-import org.joda.time.DateTime;
-import org.joda.time.format.DateTimeFormat;
-import org.joda.time.format.DateTimeFormatter;
-
-import org.quartz.*;
-import org.quartz.impl.StdSchedulerFactory;
-
-import org.testng.Assert;
-import org.testng.annotations.*;
-
-import lombok.extern.slf4j.Slf4j;
-/**
- * Tests for AlarmService.
- */
-@Slf4j
-@Test(groups = "unit-test")
-public class AlarmServiceTest {
-
- private AlarmService alarmService;
-
- private EventServiceImpl eventService;
-
- private static List<SchedulerAlarmEvent> events = new LinkedList<>();
-
- private static volatile int counter = 0;
-
- private final LensEventListener<SchedulerAlarmEvent> alarmEventListener = new
- LensEventListener<SchedulerAlarmEvent>() {
-
- @Override
- public void onEvent(SchedulerAlarmEvent event) {
- events.add(event);
- }
- };
-
- @BeforeMethod
- public void initializeEventsList() {
- events = new LinkedList<>();
- }
-
- @BeforeTest
- public void setUp() {
- System.setProperty(LensConfConstants.CONFIG_LOCATION, "target/test-classes/");
- LensServices.get().init(LensServerConf.getHiveConf());
- LensServices.get().start();
- eventService = LensServices.get().getService(LensEventService.NAME);
- assertNotNull(eventService);
- alarmService = LensServices.get().getService(AlarmService.NAME);
- assertNotNull(alarmService);
- eventService.addListenerForType(alarmEventListener, SchedulerAlarmEvent.class);
- }
-
- /**
- * This test generally tests the basic understanding and assumptions about quartz framework with a sample job.
- *
- * @throws SchedulerException
- * @throws InterruptedException
- */
- @Test
- public void testCoreExecution() throws SchedulerException, InterruptedException {
- SchedulerFactory sf = new StdSchedulerFactory();
- Scheduler scheduler = sf.getScheduler();
- scheduler.start();
- try {
- // prepare a sample Job
- JobDetail job = JobBuilder.newJob(TestJob.class).withIdentity("random", "group").build();
- scheduler.scheduleJob(job, getPastPerSecondsTrigger());
- Thread.sleep(2000);
- Assert.assertEquals(counter, 10);
- } finally {
- scheduler.shutdown();
- }
- }
-
- private CalendarIntervalScheduleBuilder getPerSecondCalendar() {
- return CalendarIntervalScheduleBuilder.calendarIntervalSchedule()
- .withInterval(1, DateBuilder.IntervalUnit.SECOND)
- .withMisfireHandlingInstructionFireAndProceed();
- }
- private Trigger getPastPerSecondsTrigger() {
- CalendarIntervalScheduleBuilder scheduleBuilder = getPerSecondCalendar();
- Date start = new Date();
- start = new Date(start.getTime() - 10000);
- Date end = new Date();
-
- return TriggerBuilder.newTrigger().withIdentity("trigger1", "group1")
- .startAt(start).endAt(end).withSchedule(scheduleBuilder).build();
- }
-
-
- @PersistJobDataAfterExecution
- @DisallowConcurrentExecution
- public static class TestJob implements Job {
- @Override
- public void execute(JobExecutionContext context) throws JobExecutionException {
- AlarmServiceTest.counter++;
- }
- }
-
- @Test
- public void testAlarmServiceEnums() throws InterruptedException, LensException {
- DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-mm-dd");
- DateTime start = formatter.parseDateTime("2016-03-03");
- DateTime end = formatter.parseDateTime("2016-03-06");
- SchedulerJobHandle jobHandle = new SchedulerJobHandle(UUID.randomUUID());
- XFrequency frequency = new XFrequency();
- frequency.setEnum(XFrequencyEnum.DAILY);
- alarmService.schedule(start, end, frequency, jobHandle.toString());
- Thread.sleep(2000); // give chance to the event listener to process the data
- int count = 0;
- for (SchedulerAlarmEvent event : events) {
- if (event.getJobHandle().equals(jobHandle)) {
- count++;
- }
- }
- Assert.assertEquals(count, 3);
- DateTime expectedDate = start;
- Set<DateTime> actualSet = new HashSet<>();
- for (SchedulerAlarmEvent e : events) {
- actualSet.add(e.getNominalTime());
- }
-
- for (int i = 0; i < 3; i++) {
- Assert.assertTrue(actualSet.contains(expectedDate));
- expectedDate = expectedDate.plusDays(1);
- }
- }
-
- @Test
- public void testAlarmServiceCronExpressions() throws InterruptedException, LensException {
- DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-mm-dd");
- DateTime start = formatter.parseDateTime("2016-03-03");
- DateTime end = formatter.parseDateTime("2016-03-06");
- SchedulerJobHandle jobHandle = new SchedulerJobHandle(UUID.randomUUID());
- System.out.println("jobHandle = " + jobHandle);
- XFrequency frequency = new XFrequency();
- frequency.setCronExpression("0/1 * * * * ?");
- alarmService.schedule(start, end, frequency, jobHandle.toString());
- Thread.sleep(2000);
- // Assert that the events are fired and at per second interval.
- assertTrue(events.size() > 1);
- }
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/src/site/apt/admin/config.apt
----------------------------------------------------------------------
diff --git a/src/site/apt/admin/config.apt b/src/site/apt/admin/config.apt
index d9b2a19..db859e6 100644
--- a/src/site/apt/admin/config.apt
+++ b/src/site/apt/admin/config.apt
@@ -40,235 +40,229 @@ Lens server configuration
*--+--+---+--+
|8|hive.server2.logging.operation.enabled|false|Disable the operation logging, since there are no execution operation happening in lens session. Keeping it true might conflict with Hive Server if both are running on the same machine causing disappearance of the operation log directory.|
*--+--+---+--+
-|9|lens.server.appevent.ws.listener.impl|org.apache.lens.server.LensApplicationListener|Implementation class for Lens Application Listener|
+|9|lens.query.current.time.millis|0|Query current time in millis. This is used to resolve 'now'. If value is set to zero, 'now' is resolved to current value|
*--+--+---+--+
-|10|lens.server.base.url|http://0.0.0.0:9999/lensapi/|The base url for the lens server|
+|10|lens.server.appevent.ws.listener.impl|org.apache.lens.server.LensApplicationListener|Implementation class for Lens Application Listener|
*--+--+---+--+
-|11|lens.server.consistentState.ws.filter.impl|org.apache.lens.server.ConsistentStateFilter|Implementation class for ConsistentState Filter|
+|11|lens.server.base.url|http://0.0.0.0:9999/lensapi/|The base url for the lens server|
*--+--+---+--+
-|12|lens.server.database.resource.dir|/tmp/lens/resources|Top level directory to store database specific static resources which are added to each database. To add a static resource for a DB, create a directory with the same name as that of the DB under this directory, and place the resources inside the DB directory. Any subdirectories under the DB directory and the resources under them will be ignored|
+|12|lens.server.consistentState.ws.filter.impl|org.apache.lens.server.ConsistentStateFilter|Implementation class for ConsistentState Filter|
*--+--+---+--+
-|13|lens.server.db.driver.name|org.hsqldb.jdbcDriver|Database driver for database where Finished queries have to be stored. Also used for database based user config loaders|
+|13|lens.server.database.resource.dir|/tmp/lens/resources|Top level directory to store database specific static resources which are added to each database. To add a static resource for a DB, create a directory with the same name as that of the DB under this directory, and place the resources inside the DB directory. Any subdirectories under the DB directory and the resources under them will be ignored|
*--+--+---+--+
-|14|lens.server.db.jdbc.pass| |JDBC Password for Finished queries table. Also used for database based user config loaders|
+|14|lens.server.db.driver.name|org.hsqldb.jdbcDriver|Database driver for database where Finished queries have to be stored. Also used for database based user config loaders|
*--+--+---+--+
-|15|lens.server.db.jdbc.url|jdbc:hsqldb:/tmp/lensserver/queries.db|JDBC URL where the database for storing finished queries is located. Also used for database based user config loaders|
+|15|lens.server.db.jdbc.pass| |JDBC Password for Finished queries table. Also used for database based user config loaders|
*--+--+---+--+
-|16|lens.server.db.jdbc.user|SA|JDBC User for Finished queries table. Also used for database based user config loaders|
+|16|lens.server.db.jdbc.url|jdbc:hsqldb:/tmp/lensserver/queries.db|JDBC URL where the database for storing finished queries is located. Also used for database based user config loaders|
*--+--+---+--+
-|17|lens.server.db.validation.query|select 1 from INFORMATION_SCHEMA.SYSTEM_USERS|BasicDatasource has a config parameter called validationQuery. This query is run before passing the Datasource to the application to verify if connection is valid.\ |
+|17|lens.server.db.jdbc.user|SA|JDBC User for Finished queries table. Also used for database based user config loaders|
+*--+--+---+--+
+|18|lens.server.db.validation.query|select 1 from INFORMATION_SCHEMA.SYSTEM_USERS|BasicDatasource has a config parameter called validationQuery. This query is run before passing the Datasource to the application to verify if connection is valid.\ |
| | | |{{{http://commons.apache.org/proper/commons-dbcp/configuration.html}dbcp configuration}} \ |
| | | |{{{http://stackoverflow.com/questions/10684244/dbcp-validationquery-for-different-databases}validationquery for different databases}} \ |
| | | | |
*--+--+---+--+
-|18|lens.server.domain|company.com|Server domain. This will be used in authentication and Sending emails to users. For the server, the username will be without domain and domain will come through conf.|
+|19|lens.server.domain|company.com|Server domain. This will be used in authentication and Sending emails to users. For the server, the username will be without domain and domain will come through conf.|
*--+--+---+--+
-|19|lens.server.driver.selector.class|org.apache.lens.server.api.driver.MinQueryCostSelector|Class for selecting best driver given the query context|
+|20|lens.server.driver.selector.class|org.apache.lens.server.api.driver.MinQueryCostSelector|Class for selecting best driver given the query context|
*--+--+---+--+
-|20|lens.server.drivers|hive:org.apache.lens.driver.hive.HiveDriver|Drivers enabled for this lens server instance|
+|21|lens.server.drivers|hive:org.apache.lens.driver.hive.HiveDriver|Drivers enabled for this lens server instance|
*--+--+---+--+
-|21|lens.server.duplicate.query.allowed|false|If this is false and same query is submitted by a user in the same session and with the same configuration while earlier query is not completed then lens server will return the handle of the previous query|
+|22|lens.server.duplicate.query.allowed|false|If this is false and same query is submitted by a user in the same session and with the same configuration while earlier query is not completed then lens server will return the handle of the previous query|
*--+--+---+--+
-|22|lens.server.enable.console.metrics|false|Enable metrics to be reported on console|
+|23|lens.server.enable.console.metrics|false|Enable metrics to be reported on console|
*--+--+---+--+
-|23|lens.server.enable.csv.metrics|false|Enable metrics to be reported as csv files in a directory|
+|24|lens.server.enable.csv.metrics|false|Enable metrics to be reported as csv files in a directory|
*--+--+---+--+
-|24|lens.server.enable.ganglia.metrics|false|Enable metrics to be reported on ganglia|
+|25|lens.server.enable.ganglia.metrics|false|Enable metrics to be reported on ganglia|
*--+--+---+--+
-|25|lens.server.enable.graphite.metrics|false|Enable metrics to be reported on graphite|
+|26|lens.server.enable.graphite.metrics|false|Enable metrics to be reported on graphite|
*--+--+---+--+
-|26|lens.server.enable.resource.method.metering|false|Whether to Enable metering for all resource methods.|
+|27|lens.server.enable.resource.method.metering|false|Whether to Enable metering for all resource methods.|
*--+--+---+--+
-|27|lens.server.estimate.pool.keepalive.millis|60000|Thread keep alive time in milliseconds for the estimate thread pool. If there are no estimate requests for this period,then cached threads are released from the pool.|
+|28|lens.server.estimate.pool.keepalive.millis|60000|Thread keep alive time in milliseconds for the estimate thread pool. If there are no estimate requests for this period,then cached threads are released from the pool.|
*--+--+---+--+
-|28|lens.server.estimate.pool.max.threads|100|Maximum number of threads in the estimate thread pool|
+|29|lens.server.estimate.pool.max.threads|100|Maximum number of threads in the estimate thread pool|
*--+--+---+--+
-|29|lens.server.estimate.pool.min.threads|3|Minimum number of threads in the estimate thread pool|
+|30|lens.server.estimate.pool.min.threads|3|Minimum number of threads in the estimate thread pool|
*--+--+---+--+
-|30|lens.server.estimate.timeout.millis|300000|Timeout for parallel query estimate calls in milliseconds. A driver needs to comeback with a query estimate within this timeout. If the timeout is reached, only the drivers that have provided an estimate would be considered for query selection. If the timeout is reached and none of the drivers have provided an estimate then estimate calls fails with an exception.|
+|31|lens.server.estimate.timeout.millis|300000|Timeout for parallel query estimate calls in milliseconds. A driver needs to comeback with a query estimate within this timeout. If the timeout is reached, only the drivers that have provided an estimate would be considered for query selection. If the timeout is reached and none of the drivers have provided an estimate then estimate calls fails with an exception.|
*--+--+---+--+
-|31|lens.server.event.service.thread.pool.size| |The size of thread pool for notifying events. The no value is specified, it uses the available processors as the number.|
+|32|lens.server.event.service.thread.pool.size| |The size of thread pool for notifying events. The no value is specified, it uses the available processors as the number.|
*--+--+---+--+
-|32|lens.server.hdfs.output.retention|1 day|hdfs output retention period. Default 1 day|
+|33|lens.server.hdfs.output.retention|1 day|hdfs output retention period. Default 1 day|
*--+--+---+--+
-|33|lens.server.index.ws.resource.impl|org.apache.lens.server.IndexResource|Implementation class for Index Resource|
+|34|lens.server.index.ws.resource.impl|org.apache.lens.server.IndexResource|Implementation class for Index Resource|
*--+--+---+--+
-|34|lens.server.inmemory.resultset.ttl.secs|300|This property defines the TTL(time to live) in seconds for all result sets of type InMemoryResultSet beyond which they are eligible for purging irrespective of whether the result set has been read or not. The default value is 300 seconds (5 minutes).|
+|35|lens.server.inmemory.resultset.ttl.secs|300|This property defines the TTL(time to live) in seconds for all result sets of type InMemoryResultSet beyond which they are eligible for purging irrespective of whether the result set has been read or not. The default value is 300 seconds (5 minutes).|
*--+--+---+--+
-|35|lens.server.launcher.pool.keepalive.millis|60000|Thread keep alive time in milliseconds for the query launcher thread pool. If there are no query launches for this period,then cached threads will be released from the pool.|
+|36|lens.server.launcher.pool.keepalive.millis|60000|Thread keep alive time in milliseconds for the query launcher thread pool. If there are no query launches for this period,then cached threads will be released from the pool.|
*--+--+---+--+
-|36|lens.server.launcher.pool.max.threads|100|Maximum number of threads in the query launcher thread pool. Keeping the default to hundred, we may never grow till there, it would go to max for concurrrent queries allowed on all drivers together. This value should be greater than the max concurrent queries allowed on all drivers.|
+|37|lens.server.launcher.pool.max.threads|100|Maximum number of threads in the query launcher thread pool. Keeping the default to hundred, we may never grow till there, it would go to max for concurrrent queries allowed on all drivers together. This value should be greater than the max concurrent queries allowed on all drivers.|
*--+--+---+--+
-|37|lens.server.launcher.pool.min.threads|3|Minimum number of threads in the query launcher thread pool|
+|38|lens.server.launcher.pool.min.threads|3|Minimum number of threads in the query launcher thread pool|
*--+--+---+--+
-|38|lens.server.log.ws.resource.impl|org.apache.lens.server.LogResource|Implementation class for Log Resource|
+|39|lens.server.log.ws.resource.impl|org.apache.lens.server.LogResource|Implementation class for Log Resource|
*--+--+---+--+
-|39|lens.server.mail.from.address|blah@company.com|The from field in the notifier mail to the submitter.|
+|40|lens.server.mail.from.address|blah@company.com|The from field in the notifier mail to the submitter.|
*--+--+---+--+
-|40|lens.server.mail.host|mail-host.company.com|SMTP Host for sending mail|
+|41|lens.server.mail.host|mail-host.company.com|SMTP Host for sending mail|
*--+--+---+--+
-|41|lens.server.mail.port|25|SMTP Port|
+|42|lens.server.mail.port|25|SMTP Port|
*--+--+---+--+
-|42|lens.server.mail.smtp.connectiontimeout|15000|Socket connection timeout value in milliseconds. This timeout is implemented by java.net.Socket. Default is 15 seconds.|
+|43|lens.server.mail.smtp.connectiontimeout|15000|Socket connection timeout value in milliseconds. This timeout is implemented by java.net.Socket. Default is 15 seconds.|
*--+--+---+--+
-|43|lens.server.mail.smtp.timeout|30000|Socket read timeout value in milliseconds. This timeout is implemented by java.net.Socket. Default is 30 seconds.|
+|44|lens.server.mail.smtp.timeout|30000|Socket read timeout value in milliseconds. This timeout is implemented by java.net.Socket. Default is 30 seconds.|
*--+--+---+--+
-|44|lens.server.max.sessions.per.user|10|Number of sessions can be allowed for each user. User has to close one of the active sessions to open a new session once limit is reached. Otherwise Server throws an exception by saying that opened session limit has been already reached for user.|
+|45|lens.server.max.sessions.per.user|10|Number of sessions can be allowed for each user. User has to close one of the active sessions to open a new session once limit is reached. Otherwise Server throws an exception by saying that opened session limit has been already reached for user.|
*--+--+---+--+
-|45|lens.server.metastore.service.impl|org.apache.lens.server.metastore.CubeMetastoreServiceImpl|Implementation class for metastore service|
+|46|lens.server.metastore.service.impl|org.apache.lens.server.metastore.CubeMetastoreServiceImpl|Implementation class for metastore service|
*--+--+---+--+
-|46|lens.server.metastore.ws.resource.impl|org.apache.lens.server.metastore.MetastoreResource|Implementation class for Metastore Resource|
+|47|lens.server.metastore.ws.resource.impl|org.apache.lens.server.metastore.MetastoreResource|Implementation class for Metastore Resource|
*--+--+---+--+
-|47|lens.server.metrics.csv.directory.path|metrics/|Path of the directory in which to report metrics as separate csv files.|
+|48|lens.server.metrics.csv.directory.path|metrics/|Path of the directory in which to report metrics as separate csv files.|
*--+--+---+--+
-|48|lens.server.metrics.ganglia.host| |The ganglia host name|
+|49|lens.server.metrics.ganglia.host| |The ganglia host name|
*--+--+---+--+
-|49|lens.server.metrics.ganglia.port| |The ganglia port|
+|50|lens.server.metrics.ganglia.port| |The ganglia port|
*--+--+---+--+
-|50|lens.server.metrics.graphite.host| |The graphite host name|
+|51|lens.server.metrics.graphite.host| |The graphite host name|
*--+--+---+--+
-|51|lens.server.metrics.graphite.port| |The graphite port|
+|52|lens.server.metrics.graphite.port| |The graphite port|
*--+--+---+--+
-|52|lens.server.metrics.reporting.period|10|The reporting period for metrics. The value is in seconds|
+|53|lens.server.metrics.reporting.period|10|The reporting period for metrics. The value is in seconds|
*--+--+---+--+
-|53|lens.server.mode|OPEN|The mode in which server should run. Allowed values are OPEN, READ_ONLY, METASTORE_READONLY, METASTORE_NODROP. OPEN mode will allow all requests. READ_ONLY mode will allow all requests on session resouce and only GET requests on all other resources. METASTORE_READONLY will allow GET on metastore and all other requests in other services. METASTORE_NODROP will not allow DELETE on metastore, will allow all other requests.|
+|54|lens.server.mode|OPEN|The mode in which server should run. Allowed values are OPEN, READ_ONLY, METASTORE_READONLY, METASTORE_NODROP. OPEN mode will allow all requests. READ_ONLY mode will allow all requests on session resouce and only GET requests on all other resources. METASTORE_READONLY will allow GET on metastore and all other requests in other services. METASTORE_NODROP will not allow DELETE on metastore, will allow all other requests.|
*--+--+---+--+
-|54|lens.server.moxyjson.ws.feature.impl|org.glassfish.jersey.moxy.json.MoxyJsonFeature|Enable Moxy json feature|
+|55|lens.server.moxyjson.ws.feature.impl|org.glassfish.jersey.moxy.json.MoxyJsonFeature|Enable Moxy json feature|
*--+--+---+--+
-|55|lens.server.moxyjsonconfigresovler.ws.feature.impl|org.apache.lens.api.util.MoxyJsonConfigurationContextResolver|Moxy json configuration resolver|
+|56|lens.server.moxyjsonconfigresovler.ws.feature.impl|org.apache.lens.api.util.MoxyJsonConfigurationContextResolver|Moxy json configuration resolver|
*--+--+---+--+
-|56|lens.server.multipart.ws.feature.impl|org.glassfish.jersey.media.multipart.MultiPartFeature|Implementation class for query scheduler resource|
+|57|lens.server.multipart.ws.feature.impl|org.glassfish.jersey.media.multipart.MultiPartFeature|Implementation class for query scheduler resource|
*--+--+---+--+
-|57|lens.server.persist.location|file:///tmp/lensserver|The directory in which lens server will persist its state when it is going down. The location be on any Hadoop compatible file system. Server will read from the location when it is restarted and recovery is enabled. So, Server should have both read and write permissions to the location|
+|58|lens.server.persist.location|file:///tmp/lensserver|The directory in which lens server will persist its state when it is going down. The location be on any Hadoop compatible file system. Server will read from the location when it is restarted and recovery is enabled. So, Server should have both read and write permissions to the location|
*--+--+---+--+
-|58|lens.server.query.acceptors| |Query Acceptors configured. Query acceptors are consulted first, before anything happens for the given query. They can either return null or return a messaging indicating why the given query shouldn't be accepted. These can be used to filter out queries at the earliest.|
+|59|lens.server.query.acceptors| |Query Acceptors configured. Query acceptors are consulted first, before anything happens for the given query. They can either return null or return a messaging indicating why the given query shouldn't be accepted. These can be used to filter out queries at the earliest.|
*--+--+---+--+
-|59|lens.server.query.launching.constraint.factories|org.apache.lens.server.query.constraint.TotalQueryCostCeilingConstraintFactory|Factories used to instantiate constraints enforced on queries by lens. Every Factory should be an implementation of org.apache.lens.server.api.common.ConfigBasedObjectCreationFactory and create an implementation of org.apache.lens.server.api.query.constraint.QueryLaunchingConstraint. A query will be launched only if all constraints pass.|
+|60|lens.server.query.launching.constraint.factories|org.apache.lens.server.query.constraint.TotalQueryCostCeilingConstraintFactory|Factories used to instantiate constraints enforced on queries by lens. Every Factory should be an implementation of org.apache.lens.server.api.common.ConfigBasedObjectCreationFactory and create an implementation of org.apache.lens.server.api.query.constraint.QueryLaunchingConstraint. A query will be launched only if all constraints pass.|
*--+--+---+--+
-|60|lens.server.query.phase1.rewriters| |Query phase 1 rewriters. This is to convert user query to cube query. The resulting cube query will be passed for validation and rewriting to hql query.\ |
+|61|lens.server.query.phase1.rewriters| |Query phase 1 rewriters. This is to convert user query to cube query. The resulting cube query will be passed for validation and rewriting to hql query.\ |
| | | |Use cases will be to use extra intelligence to convert user query to optimized cube query. \ |
| | | |Or define shortcuts for certain frequently used queries :) |
*--+--+---+--+
-|61|lens.server.query.resultset.retention|1 day|Lens query resultset retention period. Default 1 day|
-*--+--+---+--+
-|62|lens.server.query.service.impl|org.apache.lens.server.query.QueryExecutionServiceImpl|Implementation class for query execution service|
-*--+--+---+--+
-|63|lens.server.query.state.logger.enabled|true|Disable or enable the query state logger with this config. The location for the logger can be specified in logback xml for the class org.apache.lens.server.query.QueryExecutionServiceImpl.QueryStatusLogger|
-*--+--+---+--+
-|64|lens.server.query.ws.resource.impl|org.apache.lens.server.query.QueryServiceResource|Implementation class for Query Resource|
-*--+--+---+--+
-|65|lens.server.querypurger.sleep.interval|10000|The interval(milliseconds) with which purger to run periodically. Default 10 sec.|
+|62|lens.server.query.resultset.retention|1 day|Lens query resultset retention period. Default 1 day|
*--+--+---+--+
-|66|lens.server.quota.service.impl|org.apache.lens.server.quota.QuotaServiceImpl|Implementation class for quota service|
+|63|lens.server.query.service.impl|org.apache.lens.server.query.QueryExecutionServiceImpl|Implementation class for query execution service|
*--+--+---+--+
-|67|lens.server.quota.ws.resource.impl|org.apache.lens.server.quota.QuotaResource|Implementation class for Quota Resource|
+|64|lens.server.query.state.logger.enabled|true|Disable or enable the query state logger with this config. The location for the logger can be specified in logback xml for the class org.apache.lens.server.query.QueryExecutionServiceImpl.QueryStatusLogger|
*--+--+---+--+
-|68|lens.server.requestlogger.ws.filter.impl|org.apache.lens.server.LensRequestLoggingFilter|Implementation class for Request logging Filter|
+|65|lens.server.query.ws.resource.impl|org.apache.lens.server.query.QueryServiceResource|Implementation class for Query Resource|
*--+--+---+--+
-|69|lens.server.resultset.purge.enabled|false|Whether to purge the query results|
+|66|lens.server.querypurger.sleep.interval|10000|The interval(milliseconds) with which purger to run periodically. Default 10 sec.|
*--+--+---+--+
-|70|lens.server.resultsetpurger.sleep.interval.secs|3600|Periodicity for Query result purger runs. Default 1 hour.|
+|67|lens.server.quota.service.impl|org.apache.lens.server.quota.QuotaServiceImpl|Implementation class for quota service|
*--+--+---+--+
-|71|lens.server.savedquery.jdbc.dialectclass|org.apache.lens.server.query.save.SavedQueryDao$HSQLDialect|Dialect of the target DB, Default is HSQL. Override with the target DB used.|
+|68|lens.server.quota.ws.resource.impl|org.apache.lens.server.quota.QuotaResource|Implementation class for Quota Resource|
*--+--+---+--+
-|72|lens.server.savedquery.list.default.count|20|Key denoting the default fetch value of saved query list api.|
+|69|lens.server.requestlogger.ws.filter.impl|org.apache.lens.server.LensRequestLoggingFilter|Implementation class for Request logging Filter|
*--+--+---+--+
-|73|lens.server.savedquery.list.default.offset|0|Key denoting the default start value of saved query list api.|
+|70|lens.server.resultset.purge.enabled|false|Whether to purge the query results|
*--+--+---+--+
-|74|lens.server.savedquery.service.impl|org.apache.lens.server.query.save.SavedQueryServiceImpl|Implementation class for saved query service|
+|71|lens.server.resultsetpurger.sleep.interval.secs|3600|Periodicity for Query result purger runs. Default 1 hour.|
*--+--+---+--+
-|75|lens.server.savedquery.ws.resource.impl|org.apache.lens.server.query.save.SavedQueryResource|Implementation class for Saved query Resource|
+|72|lens.server.savedquery.jdbc.dialectclass|org.apache.lens.server.query.save.SavedQueryDao$HSQLDialect|Dialect of the target DB, Default is HSQL. Override with the target DB used.|
*--+--+---+--+
-|76|lens.server.scheduler.service.impl|org.apache.lens.server.scheduler.SchedulerServiceImpl|Implementation class for query scheduler service|
+|73|lens.server.savedquery.list.default.count|20|Key denoting the default fetch value of saved query list api.|
*--+--+---+--+
-|77|lens.server.scheduler.store.class|org.apache.lens.server.scheduler.SchedulerDAO$SchedulerHsqlDBStore|A subclass of SchedulerDBStore class used for storing scheduler related information.|
+|74|lens.server.savedquery.list.default.offset|0|Key denoting the default start value of saved query list api.|
*--+--+---+--+
-|78|lens.server.scheduler.ws.resource.impl|org.apache.lens.server.scheduler.ScheduleResource|Implementation class for query scheduler resource|
+|75|lens.server.savedquery.service.impl|org.apache.lens.server.query.save.SavedQueryServiceImpl|Implementation class for saved query service|
*--+--+---+--+
-|79|lens.server.scheduling.queue.poll.interval.millisec|2000|The interval at which submission thread will poll scheduling queue to fetch the next query for submission. If value is less than equal to 0, then it would mean that thread will continuosly poll without sleeping. The interval has to be given in milliseconds.|
+|76|lens.server.savedquery.ws.resource.impl|org.apache.lens.server.query.save.SavedQueryResource|Implementation class for Saved query Resource|
*--+--+---+--+
-|80|lens.server.serverMode.ws.filter.impl|org.apache.lens.server.ServerModeFilter|Implementation class for ServerMode Filter|
+|77|lens.server.scheduler.service.impl|org.apache.lens.server.scheduler.SchedulerServiceImpl|Implementation class for query scheduler service|
*--+--+---+--+
-|81|lens.server.service.provider.factory|org.apache.lens.server.ServiceProviderFactoryImpl|Service provider factory implementation class. This parameter is used to lookup the factory implementation class name that would provide an instance of ServiceProvider. Users should instantiate the class to obtain its instance. Example -- Class spfClass = conf.getClass("lens.server.service.provider.factory", null, ServiceProviderFactory.class); ServiceProviderFactory spf = spfClass.newInstance(); ServiceProvider serviceProvider = spf.getServiceProvider(); -- This is not supposed to be overridden by users.|
+|78|lens.server.scheduler.store.class|org.apache.lens.server.scheduler.SchedulerDAO$SchedulerHsqlDBStore|A subclass of SchedulerDBStore class used for storing scheduler related information.|
*--+--+---+--+
-|82|lens.server.servicenames|session,query,savedquery,metastore,scheduler,quota|These services would be started in the specified order when lens-server starts up|
+|79|lens.server.scheduler.ws.resource.impl|org.apache.lens.server.scheduler.ScheduleResource|Implementation class for query scheduler resource|
*--+--+---+--+
-|83|lens.server.session.expiry.service.interval.secs|3600|Interval at which lens session expiry service runs|
+|80|lens.server.scheduling.queue.poll.interval.millisec|2000|The interval at which submission thread will poll scheduling queue to fetch the next query for submission. If value is less than equal to 0, then it would mean that thread will continuosly poll without sleeping. The interval has to be given in milliseconds.|
*--+--+---+--+
-|84|lens.server.session.service.impl|org.apache.lens.server.session.HiveSessionService|Implementation class for session service|
+|81|lens.server.serverMode.ws.filter.impl|org.apache.lens.server.ServerModeFilter|Implementation class for ServerMode Filter|
*--+--+---+--+
-|85|lens.server.session.timeout.seconds|86400|Lens session timeout in seconds.If there is no activity on the session for this period then the session will be closed.Default timeout is one day.|
+|82|lens.server.service.provider.factory|org.apache.lens.server.ServiceProviderFactoryImpl|Service provider factory implementation class. This parameter is used to lookup the factory implementation class name that would provide an instance of ServiceProvider. Users should instantiate the class to obtain its instance. Example -- Class spfClass = conf.getClass("lens.server.service.provider.factory", null, ServiceProviderFactory.class); ServiceProviderFactory spf = spfClass.newInstance(); ServiceProvider serviceProvider = spf.getServiceProvider(); -- This is not supposed to be overridden by users.|
*--+--+---+--+
-|86|lens.server.session.ws.resource.impl|org.apache.lens.server.session.SessionResource|Implementation class for Session Resource|
+|83|lens.server.servicenames|session,query,savedquery,metastore,scheduler,quota|These services would be started in the specified order when lens-server starts up|
*--+--+---+--+
-|87|lens.server.state.persist.out.stream.buffer.size|1048576|Output Stream Buffer Size used in writing lens server state to file system. Size is in bytes.|
+|84|lens.server.session.expiry.service.interval.secs|3600|Interval at which lens session expiry service runs|
*--+--+---+--+
-|88|lens.server.state.persistence.enabled|true|If flag is enabled, state of all the services will be persisted periodically to a location specified by lens.server.persist.location and on server restart all the services will be started from last saved state.|
+|85|lens.server.session.service.impl|org.apache.lens.server.session.HiveSessionService|Implementation class for session service|
*--+--+---+--+
-|89|lens.server.state.persistence.interval.millis|300000|Lens server state persistence time interval in milliseconds|
+|86|lens.server.session.timeout.seconds|86400|Lens session timeout in seconds.If there is no activity on the session for this period then the session will be closed.Default timeout is one day.|
*--+--+---+--+
-|90|lens.server.statistics.db|lensstats|Database to which statistics tables are created and partitions are added.|
+|87|lens.server.session.ws.resource.impl|org.apache.lens.server.session.SessionResource|Implementation class for Session Resource|
*--+--+---+--+
-|91|lens.server.statistics.log.rollover.interval|3600000|Default rate which log statistics store scans for rollups in milliseconds.|
+|88|lens.server.state.persist.out.stream.buffer.size|1048576|Output Stream Buffer Size used in writing lens server state to file system. Size is in bytes.|
*--+--+---+--+
-|92|lens.server.statistics.store.class|org.apache.lens.server.stats.store.log.LogStatisticsStore|Default implementation of class used to persist Lens Statistics.|
+|89|lens.server.state.persistence.enabled|true|If flag is enabled, state of all the services will be persisted periodically to a location specified by lens.server.persist.location and on server restart all the services will be started from last saved state.|
*--+--+---+--+
-|93|lens.server.statistics.warehouse.dir|file:///tmp/lens/statistics/warehouse|Default top level location where stats are moved by the log statistics store.|
+|90|lens.server.state.persistence.interval.millis|300000|Lens server state persistence time interval in milliseconds|
*--+--+---+--+
-|94|lens.server.status.update.exponential.wait.millis|30000|Number of millis that would grow exponentially for next update, incase of transient failures.|
+|91|lens.server.statistics.db|lensstats|Database to which statistics tables are created and partitions are added.|
*--+--+---+--+
-|95|lens.server.status.update.maximum.delay.secs|1800|The maximum delay in seconds for next status update to happen after any transient failure. This will be used a maximum delay sothat exponential wait times not to grow to bigger value.|
+|92|lens.server.statistics.log.rollover.interval|3600000|Default rate which log statistics store scans for rollups in milliseconds.|
*--+--+---+--+
-|96|lens.server.status.update.num.retries|10|The number of retries a status update will tried with exponentital back off, in case of transient issues, upon which query will be marked FAILED.|
+|93|lens.server.statistics.store.class|org.apache.lens.server.stats.store.log.LogStatisticsStore|Default implementation of class used to persist Lens Statistics.|
*--+--+---+--+
-|97|lens.server.total.query.cost.ceiling.per.user|-1.0|A query submitted by user will be launched only if total query cost of all current launched queries of user is less than or equal to total query cost ceiling defined by this property. This configuration value is only useful when TotalQueryCostCeilingConstraint is enabled by using org.apache.lens.server.query.constraint.TotalQueryCostCeilingConstraintFactory as one of the factories in lens.server.query.constraint.factories property. Default is -1.0 which means that there is no limit on the total query cost of launched queries submitted by a user.|
+|94|lens.server.statistics.warehouse.dir|file:///tmp/lens/statistics/warehouse|Default top level location where stats are moved by the log statistics store.|
*--+--+---+--+
-|98|lens.server.ui.base.uri|http://0.0.0.0:19999/|The base url for the Lens UI Server|
+|95|lens.server.status.update.exponential.wait.millis|30000|Number of millis that would grow exponentially for next update, incase of transient failures.|
*--+--+---+--+
-|99|lens.server.ui.enable|true|Bringing up the ui server is optional. By default it brings up UI server.|
+|96|lens.server.status.update.maximum.delay.secs|1800|The maximum delay in seconds for next status update to happen after any transient failure. This will be used a maximum delay sothat exponential wait times not to grow to bigger value.|
*--+--+---+--+
-|100|lens.server.ui.enable.caching|true|Set this to false to disable static file caching in the UI server|
+|97|lens.server.status.update.num.retries|10|The number of retries a status update will tried with exponentital back off, in case of transient issues, upon which query will be marked FAILED.|
*--+--+---+--+
-|101|lens.server.ui.static.dir|webapp/lens-server/static|The base directory to server UI static files from|
+|98|lens.server.total.query.cost.ceiling.per.user|-1.0|A query submitted by user will be launched only if total query cost of all current launched queries of user is less than or equal to total query cost ceiling defined by this property. This configuration value is only useful when TotalQueryCostCeilingConstraint is enabled by using org.apache.lens.server.query.constraint.TotalQueryCostCeilingConstraintFactory as one of the factories in lens.server.query.constraint.factories property. Default is -1.0 which means that there is no limit on the total query cost of launched queries submitted by a user.|
*--+--+---+--+
-|102|lens.server.user.resolver.custom.class|full.package.name.Classname|Required for CUSTOM user resolver. In case the provided implementations are not sufficient for user config resolver, a custom classname can be provided. Class should extend org.apache.lens.server.user.UserConfigLoader|
+|99|lens.server.user.resolver.custom.class|full.package.name.Classname|Required for CUSTOM user resolver. In case the provided implementations are not sufficient for user config resolver, a custom classname can be provided. Class should extend org.apache.lens.server.user.UserConfigLoader|
*--+--+---+--+
-|103|lens.server.user.resolver.db.keys|lens.session.cluster.user,mapred.job.queue.name|Required for DATABASE and LDAP_BACKED_DATABASE user resolvers. For database based user config loaders, the conf keys that will be loaded from database.|
+|100|lens.server.user.resolver.db.keys|lens.session.cluster.user,mapred.job.queue.name|Required for DATABASE and LDAP_BACKED_DATABASE user resolvers. For database based user config loaders, the conf keys that will be loaded from database.|
*--+--+---+--+
-|104|lens.server.user.resolver.db.query|select clusteruser,queue from user_config_table where username=?|Required for DATABASE and LDAP_BACKED_DATABASE user resolvers. For database based user config loader, this query will be run with single argument = logged in user and the result columns will be assigned to lens.server.user.resolver.db.keys in order. For ldap backed database resolver, the argument to this query will be the intermediate values obtained from ldap.|
+|101|lens.server.user.resolver.db.query|select clusteruser,queue from user_config_table where username=?|Required for DATABASE and LDAP_BACKED_DATABASE user resolvers. For database based user config loader, this query will be run with single argument = logged in user and the result columns will be assigned to lens.server.user.resolver.db.keys in order. For ldap backed database resolver, the argument to this query will be the intermediate values obtained from ldap.|
*--+--+---+--+
-|105|lens.server.user.resolver.fixed.value| |Required for FIXED user resolver. when lens.server.user.resolver.type=FIXED, This will be the value cluster user will resolve to.|
+|102|lens.server.user.resolver.fixed.value| |Required for FIXED user resolver. when lens.server.user.resolver.type=FIXED, This will be the value cluster user will resolve to.|
*--+--+---+--+
-|106|lens.server.user.resolver.ldap.bind.dn| |Required for LDAP_BACKED_DATABASE user resolvers. ldap dn for admin binding example: CN=company-it-admin,ou=service-account,ou=company-service-account,dc=dc1,dc=com...|
+|103|lens.server.user.resolver.ldap.bind.dn| |Required for LDAP_BACKED_DATABASE user resolvers. ldap dn for admin binding example: CN=company-it-admin,ou=service-account,ou=company-service-account,dc=dc1,dc=com...|
*--+--+---+--+
-|107|lens.server.user.resolver.ldap.bind.password| |Required for LDAP_BACKED_DATABASE user resolvers. ldap password for admin binding above|
+|104|lens.server.user.resolver.ldap.bind.password| |Required for LDAP_BACKED_DATABASE user resolvers. ldap password for admin binding above|
*--+--+---+--+
-|108|lens.server.user.resolver.ldap.fields|department|Required for LDAP_BACKED_DATABASE user resolvers. list of fields to be obtained from ldap. These will be cached by the intermediate db.|
+|105|lens.server.user.resolver.ldap.fields|department|Required for LDAP_BACKED_DATABASE user resolvers. list of fields to be obtained from ldap. These will be cached by the intermediate db.|
*--+--+---+--+
-|109|lens.server.user.resolver.ldap.intermediate.db.delete.sql|delete from user_department where username=?|Required for LDAP_BACKED_DATABASE user resolvers. query to delete intermediate values from database backing ldap as cache. one argument: logged in user.|
+|106|lens.server.user.resolver.ldap.intermediate.db.delete.sql|delete from user_department where username=?|Required for LDAP_BACKED_DATABASE user resolvers. query to delete intermediate values from database backing ldap as cache. one argument: logged in user.|
*--+--+---+--+
-|110|lens.server.user.resolver.ldap.intermediate.db.insert.sql|insert into user_department (username, department, expiry) values (?, ?, ?)|Required for LDAP_BACKED_DATABASE user resolvers. query to insert intermediate values from database backing ldap as cache. arguments: first logged in user, then all intermediate values, then current time + expiration time|
+|107|lens.server.user.resolver.ldap.intermediate.db.insert.sql|insert into user_department (username, department, expiry) values (?, ?, ?)|Required for LDAP_BACKED_DATABASE user resolvers. query to insert intermediate values from database backing ldap as cache. arguments: first logged in user, then all intermediate values, then current time + expiration time|
*--+--+---+--+
-|111|lens.server.user.resolver.ldap.intermediate.db.query|select department from user_department where username=? and expiry>?|Required for LDAP_BACKED_DATABASE user resolvers. query to obtain intermediate values from database backing ldap as cache. two arguments: logged in user and current time.|
+|108|lens.server.user.resolver.ldap.intermediate.db.query|select department from user_department where username=? and expiry>?|Required for LDAP_BACKED_DATABASE user resolvers. query to obtain intermediate values from database backing ldap as cache. two arguments: logged in user and current time.|
*--+--+---+--+
-|112|lens.server.user.resolver.ldap.search.base| |Required for LDAP_BACKED_DATABASE user resolvers. for searching intermediate values for a user, the search keys. example: cn=users,dc=dc1,dc=dc2...|
+|109|lens.server.user.resolver.ldap.search.base| |Required for LDAP_BACKED_DATABASE user resolvers. for searching intermediate values for a user, the search keys. example: cn=users,dc=dc1,dc=dc2...|
*--+--+---+--+
-|113|lens.server.user.resolver.ldap.search.filter|(&(objectClass=user)(sAMAccountName=%s))|Required for LDAP_BACKED_DATABASE user resolvers. filter pattern for ldap search|
+|110|lens.server.user.resolver.ldap.search.filter|(&(objectClass=user)(sAMAccountName=%s))|Required for LDAP_BACKED_DATABASE user resolvers. filter pattern for ldap search|
*--+--+---+--+
-|114|lens.server.user.resolver.ldap.url| |Required for LDAP_BACKED_DATABASE user resolvers. ldap url to connect to.|
+|111|lens.server.user.resolver.ldap.url| |Required for LDAP_BACKED_DATABASE user resolvers. ldap url to connect to.|
*--+--+---+--+
-|115|lens.server.user.resolver.propertybased.filename|/path/to/propertyfile|Required for PROPERTYBASED user resolver. when lens.server.user.resolver.type is PROPERTYBASED, then this file will be read and parsed to determine cluster user. Each line should contain username followed by DOT followed by property full name followed by equal-to sign and followed by value. example schema of the file is: user1.lens.server.cluster.user=clusteruser1 user1.mapred.job.queue.name=queue1 *.lens.server.cluster.user=defaultclusteruser *.mapred.job.queue.name=default|
+|112|lens.server.user.resolver.propertybased.filename|/path/to/propertyfile|Required for PROPERTYBASED user resolver. when lens.server.user.resolver.type is PROPERTYBASED, then this file will be read and parsed to determine cluster user. Each line should contain username followed by DOT followed by property full name followed by equal-to sign and followed by value. example schema of the file is: user1.lens.server.cluster.user=clusteruser1 user1.mapred.job.queue.name=queue1 *.lens.server.cluster.user=defaultclusteruser *.mapred.job.queue.name=default|
*--+--+---+--+
-|116|lens.server.user.resolver.type|FIXED|Type of user config resolver. allowed values are FIXED, PROPERTYBASED, DATABASE, LDAP_BACKED_DATABASE, CUSTOM.|
+|113|lens.server.user.resolver.type|FIXED|Type of user config resolver. allowed values are FIXED, PROPERTYBASED, DATABASE, LDAP_BACKED_DATABASE, CUSTOM.|
*--+--+---+--+
-|117|lens.server.waiting.queries.selection.policy.factories|org.apache.lens.server.query.collect.UserSpecificWaitingQueriesSelectionPolicyFactory|Factories used to instantiate waiting queries selection policies. Every factory should be an implementation of org.apache.lens.server.api.common.ConfigBasedObjectCreationFactory and create an implementation of org.apache.lens.server.api.query.collect.WaitingQueriesSelectionPolicy.|
+|114|lens.server.waiting.queries.selection.policy.factories|org.apache.lens.server.query.collect.UserSpecificWaitingQueriesSelectionPolicyFactory|Factories used to instantiate waiting queries selection policies. Every factory should be an implementation of org.apache.lens.server.api.common.ConfigBasedObjectCreationFactory and create an implementation of org.apache.lens.server.api.query.collect.WaitingQueriesSelectionPolicy.|
*--+--+---+--+
-|118|lens.server.ws.featurenames|multipart,moxyjson,moxyjsonconfigresovler|These JAX-RS Feature(s) would be started in the specified order when lens-server starts up|
+|115|lens.server.ws.featurenames|multipart,moxyjson,moxyjsonconfigresovler|These JAX-RS Feature(s) would be started in the specified order when lens-server starts up|
*--+--+---+--+
-|119|lens.server.ws.filternames|requestlogger,consistentState,serverMode|These JAX-RS filters would be started in the specified order when lens-server starts up|
+|116|lens.server.ws.filternames|requestlogger,consistentState,serverMode|These JAX-RS filters would be started in the specified order when lens-server starts up|
*--+--+---+--+
-|120|lens.server.ws.listenernames|appevent|These listeners would be called in the specified order when lens-server starts up|
+|117|lens.server.ws.listenernames|appevent|These listeners would be called in the specified order when lens-server starts up|
*--+--+---+--+
-|121|lens.server.ws.resourcenames|session,metastore,query,savedquery,quota,scheduler,index,log|These JAX-RS resources would be started in the specified order when lens-server starts up|
+|118|lens.server.ws.resourcenames|session,metastore,query,savedquery,quota,scheduler,index,log|These JAX-RS resources would be started in the specified order when lens-server starts up|
*--+--+---+--+
The configuration parameters and their default values
[3/3] lens git commit: LENS-128: SchedulerServiceImpl for scheduler.
Posted by pr...@apache.org.
LENS-128: SchedulerServiceImpl for scheduler.
Project: http://git-wip-us.apache.org/repos/asf/lens/repo
Commit: http://git-wip-us.apache.org/repos/asf/lens/commit/1a96948e
Tree: http://git-wip-us.apache.org/repos/asf/lens/tree/1a96948e
Diff: http://git-wip-us.apache.org/repos/asf/lens/diff/1a96948e
Branch: refs/heads/master
Commit: 1a96948eeeb6433b7b2345cea086900004b631db
Parents: acb32d5
Author: Lavkesh Lahngir <la...@linux.com>
Authored: Mon Jul 25 17:35:55 2016 +0530
Committer: Rajat Khandelwal <ra...@gmail.com>
Committed: Mon Jul 25 17:35:55 2016 +0530
----------------------------------------------------------------------
.../error/InvalidStateTransitionException.java | 46 +++
.../lens/api/scheduler/SchedulerJobEvent.java | 31 ++
.../lens/api/scheduler/SchedulerJobInfo.java | 8 +-
.../scheduler/SchedulerJobInstanceEvent.java | 33 ++
.../api/scheduler/SchedulerJobInstanceInfo.java | 44 +--
.../api/scheduler/SchedulerJobInstanceRun.java | 80 +++++
.../scheduler/SchedulerJobInstanceState.java | 157 ++++++++++
.../scheduler/SchedulerJobInstanceStatus.java | 23 --
.../lens/api/scheduler/SchedulerJobState.java | 109 +++++++
.../lens/api/scheduler/SchedulerJobStatus.java | 23 --
.../lens/api/scheduler/StateTransitioner.java | 34 ++
.../lens/cube/parse/TimerangeResolver.java | 14 +-
.../FieldsCannotBeQueriedTogetherTest.java | 1 -
.../lens/cube/parse/TestTimeRangeResolver.java | 19 ++
.../lens/server/api/LensConfConstants.java | 4 +
.../error/InvalidStateTransitionException.java | 46 ---
.../server/api/events/SchedulerAlarmEvent.java | 15 +-
.../server/api/scheduler/SchedulerService.java | 65 ++--
.../lens/server/api/scheduler/StateMachine.java | 35 ---
.../org/apache/lens/server/BaseLensService.java | 10 +-
.../org/apache/lens/server/LensServices.java | 4 +-
.../lens/server/metrics/MetricsServiceImpl.java | 10 +-
.../lens/server/scheduler/AlarmService.java | 249 +++++++++++++++
.../lens/server/scheduler/ScheduleResource.java | 30 +-
.../lens/server/scheduler/SchedulerDAO.java | 309 ++++++++++++++-----
.../scheduler/SchedulerEventListener.java | 174 +++++++++++
.../scheduler/SchedulerQueryEventListener.java | 87 ++++++
.../server/scheduler/SchedulerServiceImpl.java | 260 +++++++++++-----
.../notification/services/AlarmService.java | 220 -------------
.../state/SchedulerJobInstanceState.java | 193 ------------
.../scheduler/state/SchedulerJobState.java | 150 ---------
.../server/scheduler/util/UtilityMethods.java | 52 ----
.../lens/server/session/LensSessionImpl.java | 2 +-
.../apache/lens/server/util/UtilityMethods.java | 53 +++-
.../src/main/resources/lensserver-default.xml | 16 +-
.../lens/server/scheduler/AlarmServiceTest.java | 177 +++++++++++
.../lens/server/scheduler/SchedulerDAOTest.java | 80 ++---
.../scheduler/TestSchedulerServiceImpl.java | 236 ++++++++++++++
.../notification/services/AlarmServiceTest.java | 180 -----------
src/site/apt/admin/config.apt | 228 +++++++-------
40 files changed, 2147 insertions(+), 1360 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/error/InvalidStateTransitionException.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/error/InvalidStateTransitionException.java b/lens-api/src/main/java/org/apache/lens/api/error/InvalidStateTransitionException.java
new file mode 100644
index 0000000..a98bc30
--- /dev/null
+++ b/lens-api/src/main/java/org/apache/lens/api/error/InvalidStateTransitionException.java
@@ -0,0 +1,46 @@
+/**
+ * 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.lens.api.error;
+
+/**
+ * Exception thrown during state transition of jobs and job instances.
+ */
+public class InvalidStateTransitionException extends Exception {
+ /**
+ * @param e Exception
+ */
+ public InvalidStateTransitionException(Throwable e) {
+ super(e);
+ }
+
+ /**
+ * @param message - custom exception message
+ * @param e
+ */
+ public InvalidStateTransitionException(String message, Throwable e) {
+ super(message, e);
+ }
+
+ /**
+ * @param message - custom exception message
+ */
+ public InvalidStateTransitionException(String message) {
+ super(message);
+ }
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobEvent.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobEvent.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobEvent.java
new file mode 100644
index 0000000..c849151
--- /dev/null
+++ b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobEvent.java
@@ -0,0 +1,31 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.api.scheduler;
+
+/**
+ * All events(actions) which can happen on a Scheduler Job.
+ */
+public enum SchedulerJobEvent {
+ ON_SUBMIT,
+ ON_SCHEDULE,
+ ON_SUSPEND,
+ ON_RESUME,
+ ON_EXPIRE,
+ ON_DELETE
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInfo.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInfo.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInfo.java
index 7d06689..b19248f 100644
--- a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInfo.java
+++ b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInfo.java
@@ -49,13 +49,13 @@ public class SchedulerJobInfo {
private String userName;
/**
- * @param state state of this job.
- * @return current state of this job
+ * @param jobstate of this job.
+ * @return current status of this job
*/
- private SchedulerJobStatus state;
+ private SchedulerJobState jobState;
/**
- * @param createdOn time to be set as createdOn.
+ * @param createdOn time to be set when it was created.
* @return time when this job was submitted.
*/
private long createdOn;
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceEvent.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceEvent.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceEvent.java
new file mode 100644
index 0000000..4f3409c
--- /dev/null
+++ b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceEvent.java
@@ -0,0 +1,33 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.api.scheduler;
+
+/**
+ * All events(actions) which can happen on an instance of <code>SchedulerJob</code>.
+ */
+public enum SchedulerJobInstanceEvent {
+ ON_CREATION, // an instance is first considered by the scheduler.
+ ON_TIME_OUT,
+ ON_CONDITIONS_MET,
+ ON_RUN,
+ ON_SUCCESS,
+ ON_FAILURE,
+ ON_RERUN,
+ ON_KILL
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceInfo.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceInfo.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceInfo.java
index 8158576..52b56ca 100644
--- a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceInfo.java
+++ b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceInfo.java
@@ -18,7 +18,7 @@
*/
package org.apache.lens.api.scheduler;
-import org.apache.lens.api.LensSessionHandle;
+import java.util.List;
import lombok.AllArgsConstructor;
import lombok.Data;
@@ -43,45 +43,15 @@ public class SchedulerJobInstanceInfo {
private SchedulerJobHandle jobId;
/**
- * @param sessionHandle new session handle.
- * @return session handle for this instance.
+ * @param scheduleTime time to be set as the nomial time for the instance.
+ * @return scheduled time of this instance.
*/
- private LensSessionHandle sessionHandle;
+ private long scheduleTime;
/**
- * @param startTime start time to be set for the instance.
- * @return actual start time of this instance.
+ * @param instanceRunList: List of runs.
+ * @return A list of instance-run for this instance.
*/
- private long startTime;
-
- /**
- * @param endTime end time to be set for the instance.
- * @return actual finish time of this instance.
- */
- private long endTime;
-
- /**
- * @param resultPath result path to be set.
- * @return result path of this instance.
- */
- private String resultPath;
-
- /**
- * @param query query to be set
- * @return query of this instance.
- */
- private String query;
-
- /**
- * @param state state to be set.
- * @return state of this instance.
- */
- private SchedulerJobInstanceStatus state;
-
- /**
- * @param createdOn time to be set as created_on time for the instance.
- * @return created_on time of this instance.
- */
- private long createdOn;
+ private List<SchedulerJobInstanceRun> instanceRunList;
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceRun.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceRun.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceRun.java
new file mode 100644
index 0000000..e6c1571
--- /dev/null
+++ b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceRun.java
@@ -0,0 +1,80 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.api.scheduler;
+
+import org.apache.lens.api.LensSessionHandle;
+import org.apache.lens.api.query.QueryHandle;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@AllArgsConstructor
+@EqualsAndHashCode
+public class SchedulerJobInstanceRun {
+
+ /**
+ * @param handle : Instance handle
+ * @return the handle
+ */
+ private SchedulerJobInstanceHandle handle;
+
+ /**
+ * @param runId : run number of the instance run. Highest run number will represent the latest run.
+ * @return the runId
+ */
+ private int runId;
+
+ /**
+ * @param sessionHandle new session handle.
+ * @return session handle for this instance run.
+ */
+ private LensSessionHandle sessionHandle;
+ /**
+ * @param startTime start time to be set for the instance run.
+ * @return actual start time of this instance run .
+ */
+ private long startTime;
+
+ /**
+ * @param endTime end time to be set for the instance run.
+ * @return actual finish time of this instance run.
+ */
+ private long endTime;
+
+ /**
+ * @param resultPath result path to be set.
+ * @return result path of this instance run.
+ */
+ private String resultPath;
+
+ /**
+ * @param queryHandle query to be set
+ * @return queryHandle of this instance run.
+ */
+ private QueryHandle queryHandle;
+
+ /**
+ * @param instanceState state to be set.
+ * @return status of this instance.
+ */
+ private SchedulerJobInstanceState instanceState;
+
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceState.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceState.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceState.java
new file mode 100644
index 0000000..93d3d7e
--- /dev/null
+++ b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceState.java
@@ -0,0 +1,157 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.api.scheduler;
+
+import org.apache.lens.api.error.InvalidStateTransitionException;
+
+public enum SchedulerJobInstanceState
+ implements StateTransitioner<SchedulerJobInstanceState, SchedulerJobInstanceEvent> {
+ // repeating same operation will return the same state to ensure idempotent behavior.
+ WAITING {
+ @Override
+ public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
+ throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_CREATION:
+ return this;
+ case ON_CONDITIONS_MET:
+ return SchedulerJobInstanceState.LAUNCHED;
+ case ON_TIME_OUT:
+ return SchedulerJobInstanceState.TIMED_OUT;
+ case ON_RUN:
+ return SchedulerJobInstanceState.RUNNING;
+ case ON_SUCCESS:
+ return SchedulerJobInstanceState.SUCCEEDED;
+ case ON_FAILURE:
+ return SchedulerJobInstanceState.FAILED;
+ case ON_KILL:
+ return SchedulerJobInstanceState.KILLED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ LAUNCHED {
+ @Override
+ public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
+ throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_CONDITIONS_MET:
+ return this;
+ case ON_RUN:
+ return SchedulerJobInstanceState.RUNNING;
+ case ON_SUCCESS:
+ return SchedulerJobInstanceState.SUCCEEDED;
+ case ON_FAILURE:
+ return SchedulerJobInstanceState.FAILED;
+ case ON_KILL:
+ return SchedulerJobInstanceState.KILLED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ RUNNING {
+ @Override
+ public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
+ throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_RUN:
+ return this;
+ case ON_SUCCESS:
+ return SchedulerJobInstanceState.SUCCEEDED;
+ case ON_FAILURE:
+ return SchedulerJobInstanceState.FAILED;
+ case ON_KILL:
+ return SchedulerJobInstanceState.KILLED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ FAILED {
+ @Override
+ public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
+ throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_FAILURE:
+ return this;
+ case ON_RERUN:
+ return SchedulerJobInstanceState.LAUNCHED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ SUCCEEDED {
+ @Override
+ public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
+ throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_SUCCESS:
+ return this;
+ case ON_RERUN:
+ return SchedulerJobInstanceState.LAUNCHED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ TIMED_OUT {
+ @Override
+ public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
+ throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_TIME_OUT:
+ return this;
+ case ON_RERUN:
+ return SchedulerJobInstanceState.WAITING;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ KILLED {
+ @Override
+ public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
+ throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_KILL:
+ return this;
+ case ON_RERUN:
+ return SchedulerJobInstanceState.LAUNCHED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceStatus.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceStatus.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceStatus.java
deleted file mode 100644
index 85e7e85..0000000
--- a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobInstanceStatus.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.lens.api.scheduler;
-
-public enum SchedulerJobInstanceStatus {
- WAITING, TIMED_OUT, LAUNCHED, RUNNING, FAILED, SUCCEEDED, KILLED,
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobState.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobState.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobState.java
new file mode 100644
index 0000000..ffaae6c
--- /dev/null
+++ b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobState.java
@@ -0,0 +1,109 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.api.scheduler;
+
+import org.apache.lens.api.error.InvalidStateTransitionException;
+
+public enum SchedulerJobState implements StateTransitioner<SchedulerJobState, SchedulerJobEvent> {
+ // repeating same operation will return the same state to ensure idempotent behavior.
+ NEW {
+ @Override
+ public SchedulerJobState nextTransition(SchedulerJobEvent event) throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_SUBMIT:
+ return this;
+ case ON_SCHEDULE:
+ return SchedulerJobState.SCHEDULED;
+ case ON_EXPIRE:
+ return SchedulerJobState.EXPIRED;
+ case ON_DELETE:
+ return SchedulerJobState.DELETED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ SCHEDULED {
+ @Override
+ public SchedulerJobState nextTransition(SchedulerJobEvent event) throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_SCHEDULE:
+ return this;
+ case ON_SUSPEND:
+ return SchedulerJobState.SUSPENDED;
+ case ON_EXPIRE:
+ return SchedulerJobState.EXPIRED;
+ case ON_DELETE:
+ return SchedulerJobState.DELETED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ SUSPENDED {
+ @Override
+ public SchedulerJobState nextTransition(SchedulerJobEvent event) throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_SUSPEND:
+ return this;
+ case ON_RESUME:
+ return SchedulerJobState.SCHEDULED;
+ case ON_EXPIRE:
+ return SchedulerJobState.EXPIRED;
+ case ON_DELETE:
+ return SchedulerJobState.DELETED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ EXPIRED {
+ @Override
+ public SchedulerJobState nextTransition(SchedulerJobEvent event) throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_EXPIRE:
+ return this;
+ case ON_DELETE:
+ return SchedulerJobState.DELETED;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ },
+
+ DELETED {
+ @Override
+ public SchedulerJobState nextTransition(SchedulerJobEvent event) throws InvalidStateTransitionException {
+ switch (event) {
+ case ON_DELETE:
+ return this;
+ default:
+ throw new InvalidStateTransitionException(
+ "SchedulerJobEvent: " + event.name() + " is not a valid event for state: " + this.name());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobStatus.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobStatus.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobStatus.java
deleted file mode 100644
index aaf403c..0000000
--- a/lens-api/src/main/java/org/apache/lens/api/scheduler/SchedulerJobStatus.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.lens.api.scheduler;
-
-public enum SchedulerJobStatus {
- NEW, SCHEDULED, SUSPENDED, EXPIRED, DELETED
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-api/src/main/java/org/apache/lens/api/scheduler/StateTransitioner.java
----------------------------------------------------------------------
diff --git a/lens-api/src/main/java/org/apache/lens/api/scheduler/StateTransitioner.java b/lens-api/src/main/java/org/apache/lens/api/scheduler/StateTransitioner.java
new file mode 100644
index 0000000..d2742ef
--- /dev/null
+++ b/lens-api/src/main/java/org/apache/lens/api/scheduler/StateTransitioner.java
@@ -0,0 +1,34 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.api.scheduler;
+
+import org.apache.lens.api.error.InvalidStateTransitionException;
+
+/**
+ * Interface to be implemented by a class that handles state transitions.
+ */
+public interface StateTransitioner<STATE extends Enum<STATE>
+ & StateTransitioner<STATE, EVENT>, EVENT extends Enum<EVENT>> {
+
+ /**
+ * @param event
+ * @return The state that the machine enters into as a result of the event.
+ * @throws InvalidStateTransitionException
+ */
+ STATE nextTransition(EVENT event) throws InvalidStateTransitionException;
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-cube/src/main/java/org/apache/lens/cube/parse/TimerangeResolver.java
----------------------------------------------------------------------
diff --git a/lens-cube/src/main/java/org/apache/lens/cube/parse/TimerangeResolver.java b/lens-cube/src/main/java/org/apache/lens/cube/parse/TimerangeResolver.java
index 33ec9d9..06ba148 100644
--- a/lens-cube/src/main/java/org/apache/lens/cube/parse/TimerangeResolver.java
+++ b/lens-cube/src/main/java/org/apache/lens/cube/parse/TimerangeResolver.java
@@ -18,8 +18,6 @@
*/
package org.apache.lens.cube.parse;
-import static org.apache.lens.cube.parse.CandidateTablePruneCause.CandidateTablePruneCode.FACT_NOT_AVAILABLE_IN_RANGE;
-
import static org.apache.hadoop.hive.ql.parse.HiveParser.*;
import java.util.*;
@@ -31,6 +29,7 @@ import org.apache.lens.cube.metadata.*;
import org.apache.lens.cube.metadata.join.JoinPath;
import org.apache.lens.cube.parse.DenormalizationResolver.ReferencedQueriedColumn;
import org.apache.lens.cube.parse.join.AutoJoinContext;
+import org.apache.lens.server.api.LensConfConstants;
import org.apache.lens.server.api.error.LensException;
import org.apache.commons.lang.StringUtils;
@@ -127,8 +126,13 @@ class TimerangeResolver implements ContextRewriter {
toDateRaw = PlanUtils.stripQuotes(timenode.getChild(3).getText());
}
}
-
- Date now = new Date();
+ long currentTime = cubeql.getConf().getLong(LensConfConstants.QUERY_CURRENT_TIME_IN_MILLIS, 0);
+ Date now;
+ if (currentTime != 0) {
+ now = new Date(currentTime);
+ } else {
+ now = new Date();
+ }
builder.fromDate(DateUtil.resolveDate(fromDateRaw, now));
if (StringUtils.isNotBlank(toDateRaw)) {
builder.toDate(DateUtil.resolveDate(toDateRaw, now));
@@ -256,6 +260,6 @@ class TimerangeResolver implements ContextRewriter {
iter.remove();
}
}
- cubeql.pruneCandidateFactSet(FACT_NOT_AVAILABLE_IN_RANGE);
+ cubeql.pruneCandidateFactSet(CandidateTablePruneCause.CandidateTablePruneCode.FACT_NOT_AVAILABLE_IN_RANGE);
}
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-cube/src/test/java/org/apache/lens/cube/parse/FieldsCannotBeQueriedTogetherTest.java
----------------------------------------------------------------------
diff --git a/lens-cube/src/test/java/org/apache/lens/cube/parse/FieldsCannotBeQueriedTogetherTest.java b/lens-cube/src/test/java/org/apache/lens/cube/parse/FieldsCannotBeQueriedTogetherTest.java
index fe2dfb3..7afa32e 100644
--- a/lens-cube/src/test/java/org/apache/lens/cube/parse/FieldsCannotBeQueriedTogetherTest.java
+++ b/lens-cube/src/test/java/org/apache/lens/cube/parse/FieldsCannotBeQueriedTogetherTest.java
@@ -47,7 +47,6 @@ public class FieldsCannotBeQueriedTogetherTest extends TestQueryRewrite {
conf.setBoolean(CubeQueryConfUtil.ENABLE_SELECT_TO_GROUPBY, true);
conf.setBoolean(CubeQueryConfUtil.DISABLE_AGGREGATE_RESOLVER, false);
conf.setBoolean(CubeQueryConfUtil.DISABLE_AUTO_JOINS, false);
-
}
@Test
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-cube/src/test/java/org/apache/lens/cube/parse/TestTimeRangeResolver.java
----------------------------------------------------------------------
diff --git a/lens-cube/src/test/java/org/apache/lens/cube/parse/TestTimeRangeResolver.java b/lens-cube/src/test/java/org/apache/lens/cube/parse/TestTimeRangeResolver.java
index c8a9ac6..280a8c4 100644
--- a/lens-cube/src/test/java/org/apache/lens/cube/parse/TestTimeRangeResolver.java
+++ b/lens-cube/src/test/java/org/apache/lens/cube/parse/TestTimeRangeResolver.java
@@ -26,16 +26,21 @@ import static org.apache.lens.cube.parse.CandidateTablePruneCause.CandidateTable
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
+import java.util.Calendar;
+import java.util.GregorianCalendar;
import java.util.List;
import java.util.Set;
import org.apache.lens.cube.error.NoCandidateFactAvailableException;
+import org.apache.lens.cube.metadata.TimeRange;
+import org.apache.lens.server.api.LensConfConstants;
import org.apache.lens.server.api.error.LensException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.ParseException;
+import org.joda.time.DateTime;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;
@@ -96,4 +101,18 @@ public class TestTimeRangeResolver extends TestQueryRewrite {
assertEquals(pruningMsg.getCause(), FACT_NOT_AVAILABLE_IN_RANGE);
assertTrue(pruningMsg.getInvalidRanges().containsAll(ctx.getTimeRanges()));
}
+
+ @Test
+ public void testCustomNow() throws Exception {
+ Configuration conf = getConf();
+ DateTime dt = new DateTime(1990, 3, 23, 12, 0, 0, 0);
+ conf.setLong(LensConfConstants.QUERY_CURRENT_TIME_IN_MILLIS, dt.getMillis());
+ CubeQueryContext ctx = rewriteCtx("select msr12 from basecube where time_range_in(d_time, 'now.day-275days','now')",
+ conf);
+ TimeRange timeRange = ctx.getTimeRanges().get(0);
+ // Month starts from zero.
+ Calendar from = new GregorianCalendar(1989, 5, 21, 0, 0, 0);
+ assertEquals(timeRange.getFromDate(), from.getTime());
+ assertEquals(timeRange.getToDate(), dt.toDate());
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java b/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java
index 43bd7e4..72e2b61 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/LensConfConstants.java
@@ -1094,4 +1094,8 @@ public final class LensConfConstants {
*/
public static final java.lang.String SCHEDULER_STORE_CLASS = SERVER_PFX + "scheduler.store.class";
+ /**
+ * Query current time for the scheduled query.
+ */
+ public static final String QUERY_CURRENT_TIME_IN_MILLIS = QUERY_PFX + "current.time.millis";
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server-api/src/main/java/org/apache/lens/server/api/error/InvalidStateTransitionException.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/error/InvalidStateTransitionException.java b/lens-server-api/src/main/java/org/apache/lens/server/api/error/InvalidStateTransitionException.java
deleted file mode 100644
index 5882151..0000000
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/error/InvalidStateTransitionException.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/**
- * 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.lens.server.api.error;
-
-/**
- * Exception thrown during state transition of jobs and job instances.
- */
-public class InvalidStateTransitionException extends LensException {
- /**
- * @param e Exception
- */
- public InvalidStateTransitionException(Throwable e) {
- super(e);
- }
-
- /**
- * @param message - custom exception message
- * @param e
- */
- public InvalidStateTransitionException(String message, Throwable e) {
- super(message, e);
- }
-
- /**
- * @param message - custom exception message
- */
- public InvalidStateTransitionException(String message) {
- super(message);
- }
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server-api/src/main/java/org/apache/lens/server/api/events/SchedulerAlarmEvent.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/events/SchedulerAlarmEvent.java b/lens-server-api/src/main/java/org/apache/lens/server/api/events/SchedulerAlarmEvent.java
index 3ca7eb9..0f2dabe 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/events/SchedulerAlarmEvent.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/events/SchedulerAlarmEvent.java
@@ -19,6 +19,7 @@
package org.apache.lens.server.api.events;
import org.apache.lens.api.scheduler.SchedulerJobHandle;
+import org.apache.lens.api.scheduler.SchedulerJobInstanceHandle;
import org.joda.time.DateTime;
@@ -34,13 +35,17 @@ public class SchedulerAlarmEvent extends LensEvent {
* jobHandle for which the alarm needs to be triggered.
*/
private SchedulerJobHandle jobHandle;
-
private DateTime nominalTime;
+ private EventType type;
+ private SchedulerJobInstanceHandle previousInstance;
- public SchedulerAlarmEvent(SchedulerJobHandle jobHandle, DateTime nominalTime) {
+ public SchedulerAlarmEvent(SchedulerJobHandle jobHandle, DateTime nominalTime, EventType type,
+ SchedulerJobInstanceHandle previousInstance) {
super(nominalTime.getMillis());
this.jobHandle = jobHandle;
this.nominalTime = nominalTime;
+ this.type = type;
+ this.previousInstance = previousInstance;
}
@Override
@@ -48,4 +53,10 @@ public class SchedulerAlarmEvent extends LensEvent {
return jobHandle.getHandleIdString();
}
+ /**
+ * Event type to know what kind of operations we want.
+ */
+ public static enum EventType {
+ SCHEDULE, EXPIRE
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/SchedulerService.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/SchedulerService.java b/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/SchedulerService.java
index d0af876..c7f73eb 100644
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/SchedulerService.java
+++ b/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/SchedulerService.java
@@ -6,9 +6,9 @@
* 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
* 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
@@ -27,13 +27,14 @@ import org.apache.lens.server.api.LensService;
import org.apache.lens.server.api.SessionValidator;
import org.apache.lens.server.api.error.LensException;
-
/**
* Scheduler interface.
*/
public interface SchedulerService extends LensService, SessionValidator {
- /** The constant NAME */
+ /**
+ * The constant name for scheduler service.
+ */
String NAME = "scheduler";
/**
@@ -51,10 +52,10 @@ public interface SchedulerService extends LensService, SessionValidator {
*
* @param sessionHandle handle for the current session.
* @param jobHandle handle for the job to be scheduled.
+ * @return true if there is a successful schedule
* @throws LensException the lens exception
*/
- void scheduleJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException;
-
+ boolean scheduleJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException;
/**
* Submit a job and also schedule it.
@@ -76,7 +77,6 @@ public interface SchedulerService extends LensService, SessionValidator {
*/
XJob getJobDefinition(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException;
-
/**
* Returns the details of a job. Details may contain extra system information like id for the job.
*
@@ -85,12 +85,11 @@ public interface SchedulerService extends LensService, SessionValidator {
* @return job details for the job
* @throws LensException the lens exception
*/
- SchedulerJobInfo getJobDetails(LensSessionHandle sessionHandle,
- SchedulerJobHandle jobHandle) throws LensException;
+ SchedulerJobInfo getJobDetails(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException;
/**
* Update a job with new definition.
- *
+ * <p>
* Updates will be applied only for newer instances. Running instances will be running with old definition
*
* @param sessionHandle
@@ -99,23 +98,22 @@ public interface SchedulerService extends LensService, SessionValidator {
* @return true or false based on whether the update was successful or failed.
* @throws LensException the lens exception
*/
- boolean updateJob(LensSessionHandle sessionHandle,
- SchedulerJobHandle jobHandle, XJob newJobDefinition) throws LensException;
-
+ boolean updateJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle, XJob newJobDefinition)
+ throws LensException;
/**
* End a job by specifying an expiry time.
*
* @param sessionHandle handle for the current session.
* @param jobHandle handle for the job
+ * @return true if the operation is successful.
* @throws LensException the lens exception
*/
- void expireJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException;
-
+ boolean expireJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException;
/**
* Suspend a job.
- *
+ * <p>
* If the job is not in scheduled state, it will return true.
* Once a job is suspended, no further instances of that job will run.
* Any running instances of that job will continue normally.
@@ -127,7 +125,6 @@ public interface SchedulerService extends LensService, SessionValidator {
*/
boolean suspendJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException;
-
/**
* Resume a job from a given time.
*
@@ -148,20 +145,21 @@ public interface SchedulerService extends LensService, SessionValidator {
*/
boolean deleteJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException;
-
/**
* @param sessionHandle handle for the current session.
* @param state filter for status, if specified only jobs in that state will be returned,
* if null no entries will be removed from result
* @param user filter for user who submitted the job, if specified only jobs submitted by the given user
* will be returned, if not specified no entries will be removed from result on basis of userName
+ * @param jobName filter for jobName, if specified only the jobs with name same as given name will be considered
+ * , else no jobs will be filtered out on the basis of name.
* @param startTime if specified only instances with scheduleTime after this time will be considered.
* @param endTime if specified only instances with scheduleTime before this time will be considered.
* @return A collection of stats per job
* @throws LensException
*/
- Collection<SchedulerJobStats> getAllJobStats(LensSessionHandle sessionHandle,
- String state, String user, long startTime, long endTime) throws LensException;
+ Collection<SchedulerJobStats> getAllJobStats(LensSessionHandle sessionHandle, String state, String user,
+ String jobName, long startTime, long endTime) throws LensException;
/**
* Returns stats for a job.
@@ -174,9 +172,8 @@ public interface SchedulerService extends LensService, SessionValidator {
* @param endTime if specified only instances with scheduleTime before this time will be considered.
* @throws LensException the lens exception
*/
- SchedulerJobStats getJobStats(LensSessionHandle sessionHandle, SchedulerJobHandle handle,
- String state, long startTime, long endTime) throws LensException;
-
+ SchedulerJobStats getJobStats(LensSessionHandle sessionHandle, SchedulerJobHandle handle, String state,
+ long startTime, long endTime) throws LensException;
/**
* Returns handles for last <code>numResults</code> instances for the job.
@@ -184,15 +181,15 @@ public interface SchedulerService extends LensService, SessionValidator {
* @param sessionHandle handle for the session.
* @param jobHandle handle for the job
* @param numResults - number of results to be returned, default 100.
- * @return list of instances for the job
+ * @return list of instance ids for the job
* @throws LensException the lens exception
*/
- List<SchedulerJobInstanceInfo> getJobInstances(LensSessionHandle sessionHandle,
- SchedulerJobHandle jobHandle, Long numResults) throws LensException;
+ List<SchedulerJobInstanceInfo> getJobInstances(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle,
+ Long numResults) throws LensException;
/**
* Kills a running job instance.
- *
+ * <p>
* If the job instance is already completed or not in running state, this will be a no-op and will return false.
*
* @param sessionHandle handle for the session.
@@ -200,12 +197,11 @@ public interface SchedulerService extends LensService, SessionValidator {
* @return true if the instance was killed successfully, false otherwise.
* @throws LensException the lens exception
*/
- boolean killInstance(LensSessionHandle sessionHandle,
- SchedulerJobInstanceHandle instanceHandle) throws LensException;
+ boolean killInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle) throws LensException;
/**
* Reruns a failed/killed/completed job instance.
- *
+ * <p>
* If the instance is not in a terminal state, then this operation will be a no-op and will return false.
*
* @param sessionHandle handle for the session.
@@ -213,8 +209,8 @@ public interface SchedulerService extends LensService, SessionValidator {
* @return true if the instance was re run successfully, false otherwise.
* @throws LensException the lens exception
*/
- boolean rerunInstance(LensSessionHandle sessionHandle,
- SchedulerJobInstanceHandle instanceHandle) throws LensException;
+ boolean rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle)
+ throws LensException;
/**
* Instance details for an instance.
@@ -225,7 +221,6 @@ public interface SchedulerService extends LensService, SessionValidator {
* @throws LensException the lens exception
*/
SchedulerJobInstanceInfo getInstanceDetails(LensSessionHandle sessionHandle,
- SchedulerJobInstanceHandle instanceHandle) throws LensException;
-
+ SchedulerJobInstanceHandle instanceHandle) throws LensException;
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/StateMachine.java
----------------------------------------------------------------------
diff --git a/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/StateMachine.java b/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/StateMachine.java
deleted file mode 100644
index ba6a0a2..0000000
--- a/lens-server-api/src/main/java/org/apache/lens/server/api/scheduler/StateMachine.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/**
- * 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.lens.server.api.scheduler;
-
-
-import org.apache.lens.server.api.error.InvalidStateTransitionException;
-
-/**
- * Interface to be implemented by a class that handles state transitions.
- */
-public interface StateMachine<STATE extends Enum<STATE>, EVENT extends Enum<EVENT>> {
-
- /**
- * @param event
- * @return The state that the machine enters into as a result of the event.
- * @throws InvalidStateTransitionException
- */
- STATE nextTransition(EVENT event) throws InvalidStateTransitionException;
-
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/BaseLensService.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/BaseLensService.java b/lens-server/src/main/java/org/apache/lens/server/BaseLensService.java
index 74bc0be..d706306 100644
--- a/lens-server/src/main/java/org/apache/lens/server/BaseLensService.java
+++ b/lens-server/src/main/java/org/apache/lens/server/BaseLensService.java
@@ -146,6 +146,7 @@ public abstract class BaseLensService extends CompositeService implements Extern
return numSessions != null && numSessions >= maxNumSessionsPerUser;
}
+
/**
* Open session.
*
@@ -157,12 +158,19 @@ public abstract class BaseLensService extends CompositeService implements Extern
*/
public LensSessionHandle openSession(String username, String password, Map<String, String> configuration)
throws LensException {
+ return openSession(username, password, configuration, true);
+ }
+
+ public LensSessionHandle openSession(String username, String password, Map<String, String> configuration,
+ boolean auth) throws LensException {
if (StringUtils.isBlank(username)) {
throw new BadRequestException("User name cannot be null or empty");
}
SessionHandle sessionHandle;
username = UtilityMethods.removeDomain(username);
- doPasswdAuth(username, password);
+ if (auth) {
+ doPasswdAuth(username, password);
+ }
SessionUser sessionUser = SESSION_USER_INSTANCE_MAP.get(username);
if (sessionUser == null) {
sessionUser = new SessionUser(username);
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/LensServices.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/LensServices.java b/lens-server/src/main/java/org/apache/lens/server/LensServices.java
index 56094e0..a709f16 100644
--- a/lens-server/src/main/java/org/apache/lens/server/LensServices.java
+++ b/lens-server/src/main/java/org/apache/lens/server/LensServices.java
@@ -37,7 +37,7 @@ import org.apache.lens.server.api.metrics.MetricsService;
import org.apache.lens.server.metrics.MetricsServiceImpl;
import org.apache.lens.server.model.LogSegregationContext;
import org.apache.lens.server.model.MappedDiagnosticLogSegregationContext;
-import org.apache.lens.server.scheduler.notification.services.AlarmService;
+import org.apache.lens.server.scheduler.AlarmService;
import org.apache.lens.server.session.LensSessionImpl;
import org.apache.lens.server.stats.StatisticsService;
import org.apache.lens.server.user.UserConfigLoaderFactory;
@@ -179,7 +179,7 @@ public class LensServices extends CompositeService implements ServiceProvider {
}
// This is only for test, to simulate a restart of the server
- static void setInstance(LensServices newInstance) {
+ public static void setInstance(LensServices newInstance) {
instance = newInstance;
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/metrics/MetricsServiceImpl.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/metrics/MetricsServiceImpl.java b/lens-server/src/main/java/org/apache/lens/server/metrics/MetricsServiceImpl.java
index f1ee169..e09180f 100644
--- a/lens-server/src/main/java/org/apache/lens/server/metrics/MetricsServiceImpl.java
+++ b/lens-server/src/main/java/org/apache/lens/server/metrics/MetricsServiceImpl.java
@@ -40,16 +40,12 @@ import org.apache.lens.server.api.metastore.CubeMetastoreService;
import org.apache.lens.server.api.metrics.*;
import org.apache.lens.server.api.query.QueryExecutionService;
import org.apache.lens.server.api.query.StatusChange;
-import org.apache.lens.server.api.session.SessionClosed;
-import org.apache.lens.server.api.session.SessionEvent;
-import org.apache.lens.server.api.session.SessionExpired;
-import org.apache.lens.server.api.session.SessionOpened;
-import org.apache.lens.server.api.session.SessionService;
+import org.apache.lens.server.api.session.*;
import org.apache.lens.server.healthcheck.LensServiceHealthCheck;
import org.apache.lens.server.query.QueryExecutionServiceImpl;
import org.apache.lens.server.quota.QuotaServiceImpl;
+import org.apache.lens.server.scheduler.AlarmService;
import org.apache.lens.server.scheduler.SchedulerServiceImpl;
-import org.apache.lens.server.scheduler.notification.services.AlarmService;
import org.apache.lens.server.session.DatabaseResourceService;
import org.apache.lens.server.session.HiveSessionService;
@@ -67,11 +63,9 @@ import com.codahale.metrics.health.HealthCheckRegistry;
import com.codahale.metrics.jvm.GarbageCollectorMetricSet;
import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
import com.codahale.metrics.jvm.ThreadStatesGaugeSet;
-
import info.ganglia.gmetric4j.gmetric.GMetric;
import info.ganglia.gmetric4j.gmetric.GMetric.UDPAddressingMode;
import lombok.Getter;
-
import lombok.extern.slf4j.Slf4j;
/**
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/AlarmService.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/AlarmService.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/AlarmService.java
new file mode 100644
index 0000000..217879f
--- /dev/null
+++ b/lens-server/src/main/java/org/apache/lens/server/scheduler/AlarmService.java
@@ -0,0 +1,249 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.lens.server.scheduler;
+
+import org.apache.lens.api.scheduler.SchedulerJobHandle;
+import org.apache.lens.api.scheduler.XFrequency;
+import org.apache.lens.api.scheduler.XFrequencyEnum;
+import org.apache.lens.server.LensServices;
+import org.apache.lens.server.api.LensService;
+import org.apache.lens.server.api.error.LensException;
+import org.apache.lens.server.api.events.LensEventService;
+import org.apache.lens.server.api.events.SchedulerAlarmEvent;
+import org.apache.lens.server.api.health.HealthStatus;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hive.service.AbstractService;
+
+import org.joda.time.DateTime;
+import org.quartz.*;
+import org.quartz.impl.StdSchedulerFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * This service is used primarily by Scheduler to get alarm notifications for scheduled queries.
+ * <p>
+ * As a schedule this service accepts start time, frequency, end time and timeZone. It also requires the
+ * {@link SchedulerJobHandle} which it sends as part of the
+ * {@link org.apache.lens.server.api.events.SchedulerAlarmEvent} to inform the scheduler about the job for which
+ * job the notification has been generated.
+ */
+@Slf4j
+public class AlarmService extends AbstractService implements LensService {
+
+ public static final String NAME = "alarm-service";
+
+ public static final String LENS_JOBS = "LensJobs";
+ public static final String ALARM_SERVICE = "AlarmService";
+
+ private Scheduler scheduler;
+
+ /**
+ * True if the service started properly and is running fine, false otherwise.
+ */
+ private boolean isHealthy = true;
+
+ /**
+ * Contains the reason if service is not healthy.
+ */
+ private String healthCause;
+
+ /**
+ * Creates a new instance of AlarmService.
+ *
+ * @param name the name
+ */
+ public AlarmService(String name) {
+ super(name);
+ }
+
+ @Override
+ public HealthStatus getHealthStatus() {
+ return isHealthy
+ ? new HealthStatus(isHealthy, "Alarm service is healthy.")
+ : new HealthStatus(isHealthy, healthCause);
+ }
+
+ public synchronized void init(HiveConf hiveConf) {
+ super.init(hiveConf);
+ try {
+ this.scheduler = StdSchedulerFactory.getDefaultScheduler();
+ } catch (SchedulerException e) {
+ isHealthy = false;
+ healthCause = "Failed to initialize the Quartz Scheduler for AlarmService.";
+ log.error(healthCause, e);
+ throw new IllegalStateException("Could not initialize the Alarm Service", e);
+ }
+ }
+
+ @Override
+ public synchronized void start() {
+ try {
+ scheduler.start();
+ log.info("Alarm service started successfully!");
+ } catch (SchedulerException e) {
+ isHealthy = false;
+ healthCause = "Failed to start the Quartz Scheduler for AlarmService.";
+ log.error(healthCause, e);
+ throw new IllegalStateException("Could not start the Alarm service", e);
+ }
+ }
+
+ @Override
+ public synchronized void stop() {
+ try {
+ scheduler.standby();
+ log.info("Alarm Service stopped successfully.");
+ } catch (SchedulerException e) {
+ log.error("Failed to shut down the Quartz Scheduler for AlarmService.", e);
+ }
+ }
+
+ /**
+ * This method can be used by any consumer who wants to receive notifications during a time range at a given
+ * frequency.
+ * <p>
+ * This method is intended to be used by LensScheduler to subscribe for time based notifications to schedule queries.
+ * On receiving a job to be scheduled LensScheduler will subscribe to all triggers required for the job, including
+ * AlarmService for time based triggers.
+ *
+ * @param start start time for notifications
+ * @param end end time for notifications
+ * @param frequency Frequency to determine the frequency at which notification should be sent.
+ * @param jobHandle Must be a unique jobHanlde across all consumers
+ */
+ public void schedule(DateTime start, DateTime end, XFrequency frequency, String jobHandle) throws LensException {
+ // accept the schedule and then keep on sending the notifications for that schedule
+ JobDataMap map = new JobDataMap();
+ map.put("jobHandle", jobHandle);
+
+ JobDetail job = JobBuilder.newJob(LensJob.class).withIdentity(jobHandle, LENS_JOBS).usingJobData(map).build();
+
+ Trigger trigger;
+ if (frequency.getEnum() != null) { //for enum expression: create a trigger using calendar interval
+ CalendarIntervalScheduleBuilder scheduleBuilder = CalendarIntervalScheduleBuilder.calendarIntervalSchedule()
+ .withInterval(getTimeInterval(frequency.getEnum()), getTimeUnit(frequency.getEnum()))
+ .withMisfireHandlingInstructionIgnoreMisfires();
+ trigger = TriggerBuilder.newTrigger().withIdentity(jobHandle, ALARM_SERVICE).startAt(start.toDate())
+ .endAt(end.toDate()).withSchedule(scheduleBuilder).build();
+ } else { // for cron expression create a cron trigger
+ trigger = TriggerBuilder.newTrigger().withIdentity(jobHandle, ALARM_SERVICE).startAt(start.toDate())
+ .endAt(end.toDate()).withSchedule(CronScheduleBuilder.cronSchedule(frequency.getCronExpression())).build();
+ }
+
+ // Tell quartz to run the job using our trigger
+ try {
+ scheduler.scheduleJob(job, trigger);
+ } catch (SchedulerException e) {
+ log.error("Error scheduling job with jobHandle: {}", jobHandle);
+ throw new LensException("Failed to schedule job with jobHandle: " + jobHandle, e);
+ }
+ }
+
+ private int getTimeInterval(XFrequencyEnum frequencyEnum) {
+ // since quarterly is not supported natively, we express it as 3 months
+ return frequencyEnum == XFrequencyEnum.QUARTERLY ? 3 : 1;
+ }
+
+ // Maps the timeunit in entity specification to the one in Quartz DateBuilder
+ private DateBuilder.IntervalUnit getTimeUnit(XFrequencyEnum frequencyEnum) {
+ switch (frequencyEnum) {
+
+ case DAILY:
+ return DateBuilder.IntervalUnit.DAY;
+
+ case WEEKLY:
+ return DateBuilder.IntervalUnit.WEEK;
+
+ case MONTHLY:
+ return DateBuilder.IntervalUnit.MONTH;
+
+ case QUARTERLY:
+ return DateBuilder.IntervalUnit.MONTH;
+
+ case YEARLY:
+ return DateBuilder.IntervalUnit.YEAR;
+
+ default:
+ throw new IllegalArgumentException("Invalid frequency enum expression: " + frequencyEnum.name());
+ }
+ }
+
+ public boolean unSchedule(SchedulerJobHandle jobHandle) throws LensException {
+ // stop sending notifications for this job handle
+ try {
+ return scheduler.deleteJob(JobKey.jobKey(jobHandle.getHandleIdString(), LENS_JOBS));
+ } catch (SchedulerException e) {
+ log.error("Failed to remove alarm triggers for job with jobHandle: " + jobHandle, e);
+ throw new LensException("Failed to remove alarm triggers for job with jobHandle: " + jobHandle, e);
+ }
+ }
+
+ public boolean checkExists(SchedulerJobHandle handle) throws LensException {
+ try {
+ return scheduler.checkExists(JobKey.jobKey(handle.getHandleIdString(), LENS_JOBS));
+ } catch (SchedulerException e) {
+ log.error("Failed to check the job with jobHandle: " + handle, e);
+ throw new LensException("Failed to check the job with jobHandle: " + handle, e);
+ }
+ }
+
+ public void pauseJob(SchedulerJobHandle jobHandle) throws LensException {
+ try {
+ scheduler.pauseJob(JobKey.jobKey(jobHandle.getHandleIdString(), LENS_JOBS));
+ } catch (SchedulerException e) {
+ log.error("Failed to pause alarm triggers for job with jobHandle: " + jobHandle, e);
+ throw new LensException("Failed to pause alarm triggers for job with jobHandle: " + jobHandle, e);
+ }
+ }
+
+ public void resumeJob(SchedulerJobHandle jobHandle) throws LensException {
+ try {
+ scheduler.resumeJob(JobKey.jobKey(jobHandle.getHandleIdString(), LENS_JOBS));
+ } catch (SchedulerException e) {
+ log.error("Failed to resume alarm triggers for job with jobHandle: " + jobHandle, e);
+ throw new LensException("Failed to resume alarm triggers for job with jobHandle: " + jobHandle, e);
+ }
+ }
+
+ public static class LensJob implements Job {
+
+ @Override
+ public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
+ JobDataMap data = jobExecutionContext.getMergedJobDataMap();
+ DateTime nominalTime = new DateTime(jobExecutionContext.getScheduledFireTime());
+ SchedulerJobHandle jobHandle = SchedulerJobHandle.fromString(data.getString("jobHandle"));
+ SchedulerAlarmEvent alarmEvent = new SchedulerAlarmEvent(jobHandle, nominalTime,
+ SchedulerAlarmEvent.EventType.SCHEDULE, null);
+ try {
+ LensEventService eventService = LensServices.get().getService(LensEventService.NAME);
+ eventService.notifyEvent(alarmEvent);
+ if (jobExecutionContext.getNextFireTime() == null) {
+ eventService
+ .notifyEvent(new SchedulerAlarmEvent(jobHandle, nominalTime, SchedulerAlarmEvent.EventType.EXPIRE, null));
+ }
+ } catch (LensException e) {
+ log.error("Failed to notify SchedulerAlarmEvent for jobHandle: {} and scheduleTime: {}",
+ jobHandle.getHandleIdString(), nominalTime.toString(), e);
+ throw new JobExecutionException("Failed to notify alarmEvent", e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/ScheduleResource.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/ScheduleResource.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/ScheduleResource.java
index 39c4d98..8603edf 100644
--- a/lens-server/src/main/java/org/apache/lens/server/scheduler/ScheduleResource.java
+++ b/lens-server/src/main/java/org/apache/lens/server/scheduler/ScheduleResource.java
@@ -40,7 +40,7 @@ import org.apache.commons.lang3.StringUtils;
@Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML})
public class ScheduleResource {
- public static enum INSTANCE_ACTIONS {
+ public enum INSTANCE_ACTIONS {
KILL, RERUN;
public static INSTANCE_ACTIONS fromString(String name) {
@@ -48,7 +48,7 @@ public class ScheduleResource {
}
}
- public static enum JOB_ACTIONS {
+ public enum JOB_ACTIONS {
SCHEDULE, EXPIRE, SUSPEND, RESUME;
public static JOB_ACTIONS fromString(String name) {
@@ -89,12 +89,12 @@ public class ScheduleResource {
@GET
@Path("jobs/stats")
public Collection<SchedulerJobStats> getAllJobStats(@QueryParam("sessionid") LensSessionHandle sessionId,
- @DefaultValue("running") @QueryParam("state") String state,
+ @DefaultValue("running") @QueryParam("status") String status,
@QueryParam("name") String jobName,
@DefaultValue("user") @QueryParam("user") String user,
@DefaultValue("-1") @QueryParam("start") long start,
@DefaultValue("-1") @QueryParam("end") long end) throws LensException {
- return getSchedulerService().getAllJobStats(sessionId, state, user, start, end);
+ return getSchedulerService().getAllJobStats(sessionId, status, user, jobName, start, end);
}
@GET
@@ -187,20 +187,30 @@ public class ScheduleResource {
@QueryParam("action") INSTANCE_ACTIONS action) throws LensException {
validateSession(sessionId);
+ APIResult res;
switch (action) {
case KILL:
- getSchedulerService().killInstance(sessionId, instanceHandle);
+ if (getSchedulerService().killInstance(sessionId, instanceHandle)) {
+ res = new APIResult(APIResult.Status.SUCCEEDED,
+ "Killing the instance with id " + instanceHandle + " was successful");
+ } else {
+ res = new APIResult(APIResult.Status.FAILED,
+ "Killing the instance with id " + instanceHandle + " was not successful");
+ }
break;
-
case RERUN:
- getSchedulerService().rerunInstance(sessionId, instanceHandle);
+ if (getSchedulerService().rerunInstance(sessionId, instanceHandle)) {
+ res = new APIResult(APIResult.Status.SUCCEEDED,
+ "Rerunning the instance with id " + instanceHandle + " was successful");
+ } else {
+ res = new APIResult(APIResult.Status.FAILED,
+ "Rerunning the instance with id " + instanceHandle + " was not successful");
+ }
break;
-
default:
throw new BadRequestException("Unsupported action " + action.toString());
}
- return APIResult.success();
+ return res;
}
-
}
[2/3] lens git commit: LENS-128: SchedulerServiceImpl for scheduler.
Posted by pr...@apache.org.
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerDAO.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerDAO.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerDAO.java
index bf99fde..7a2b06a 100644
--- a/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerDAO.java
+++ b/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerDAO.java
@@ -28,10 +28,11 @@ import javax.xml.bind.JAXBElement;
import org.apache.lens.api.LensSessionHandle;
import org.apache.lens.api.ToXMLString;
+import org.apache.lens.api.query.QueryHandle;
import org.apache.lens.api.scheduler.*;
import org.apache.lens.server.api.LensConfConstants;
import org.apache.lens.server.api.error.LensException;
-import org.apache.lens.server.scheduler.util.UtilityMethods;
+import org.apache.lens.server.util.UtilityMethods;
import org.apache.commons.dbcp.BasicDataSource;
import org.apache.commons.dbutils.QueryRunner;
@@ -54,9 +55,10 @@ public class SchedulerDAO {
Class dbStoreClass = Class
.forName(conf.get(LensConfConstants.SCHEDULER_STORE_CLASS, SchedulerHsqlDBStore.class.getName()));
this.store = (SchedulerDBStore) dbStoreClass.newInstance();
- this.store.init(UtilityMethods.getDataSourceFromConf(conf));
+ this.store.init(UtilityMethods.getDataSourceFromConfForScheduler(conf));
this.store.createJobTable();
- this.store.createJobInstaceTable();
+ this.store.createJobInstanceTable();
+ this.store.createJobInstanceRunTable();
} catch (SQLException e) {
log.error("Error creating job tables", e);
throw new LensException("Error creating job tables ", e);
@@ -115,16 +117,31 @@ public class SchedulerDAO {
}
/**
- * Gets the Job state
+ * Get the user of the job who submitted the job.
+ *
+ * @param id
+ * @return
+ */
+ public String getUser(SchedulerJobHandle id) {
+ try {
+ return store.getUser(id.getHandleIdString());
+ } catch (SQLException e) {
+ log.error("Error while getting the user for the job with handle " + id.getHandleIdString(), e);
+ return null;
+ }
+ }
+
+ /**
+ * Gets the Job status
*
* @param id : Job handle id.
- * @return SchedulerJobState of the job.
+ * @return SchedulerJobStatus of the job.
*/
- public SchedulerJobStatus getJobState(SchedulerJobHandle id) {
+ public SchedulerJobState getJobState(SchedulerJobHandle id) {
try {
return store.getJobState(id.getHandleIdString());
} catch (SQLException e) {
- log.error("Error while getting the job state for " + id.getHandleIdString(), e);
+ log.error("Error while getting the job status for " + id.getHandleIdString(), e);
return null;
}
}
@@ -145,16 +162,16 @@ public class SchedulerDAO {
}
/**
- * Updates the job state form the new SchedulerJobInfo
+ * Updates the job status form the new SchedulerJobInfo
*
* @param info: Updated info objects
* @return number of rows updated.
*/
- public int updateJobState(SchedulerJobInfo info) {
+ public int updateJobStatus(SchedulerJobInfo info) {
try {
- return store.updateJobState(info.getId().getHandleIdString(), info.getState().name(), info.getModifiedOn());
+ return store.updateJobStatus(info.getId().getHandleIdString(), info.getJobState().name(), info.getModifiedOn());
} catch (SQLException e) {
- log.error("Error while updating job state for " + info.getId().getHandleIdString(), e);
+ log.error("Error while updating job status for " + info.getId().getHandleIdString(), e);
return 0;
}
}
@@ -168,6 +185,17 @@ public class SchedulerDAO {
}
}
+ public int storeJobInstanceRun(SchedulerJobInstanceRun instanceRun) {
+ try {
+ return store.insertIntoJobInstanceRunTable(instanceRun);
+ } catch (SQLException e) {
+ log.error(
+ "Error while storing job instance run for " + instanceRun.getRunId() + " and instance handle " + instanceRun
+ .getHandle().getHandleIdString(), e);
+ return 0;
+ }
+ }
+
/**
* Gets the SchedulerJobInstanceInfo corresponding instance handle id.
*
@@ -184,16 +212,17 @@ public class SchedulerDAO {
}
/**
- * Updates the instance state
+ * Updates the instance status
*
- * @param info: Updated instance info
+ * @param instanceRun : instance Run object
* @return number of rows updated.
*/
- public int updateJobInstanceState(SchedulerJobInstanceInfo info) {
+ public int updateJobInstanceRun(SchedulerJobInstanceRun instanceRun) {
try {
- return store.updateJobInstanceState(info.getId().getHandleIdString(), info.getState().name());
+ return store.updateJobInstanceRun(instanceRun);
} catch (SQLException e) {
- log.error("Error while updating the job instance state for " + info.getId().getHandleIdString(), e);
+ log.error("Error while updating the job instance status for " + instanceRun.getHandle().getHandleIdString()
+ + " and run: " + instanceRun.getRunId(), e);
return 0;
}
}
@@ -204,7 +233,7 @@ public class SchedulerDAO {
* @param id: Job handle id.
* @return List of instance handles.
*/
- public List<SchedulerJobInstanceHandle> getJobInstances(SchedulerJobHandle id) {
+ public List<SchedulerJobInstanceInfo> getJobInstances(SchedulerJobHandle id) {
// TODO: Add number of results to be fetched
try {
return store.getAllJobInstances(id.getHandleIdString());
@@ -218,15 +247,29 @@ public class SchedulerDAO {
* Gets all jobs which match the filter requirements.
*
* @param username : User name of the job
- * @param state : State of the job
+ * @param jobState : state of the job
* @param startTime : Created on should be greater than this start time.
* @param endTime : Created on should be less than the end time.
* @return List of Job handles
*/
- public List<SchedulerJobHandle> getJobs(String username, SchedulerJobStatus state, Long startTime,
- Long endTime) {
+ public List<SchedulerJobHandle> getJobs(String username, SchedulerJobState jobState, Long startTime, Long endTime) {
+ try {
+ return store.getJobs(username, jobState == null ? null : jobState.name(), startTime, endTime);
+ } catch (SQLException e) {
+ log.error("Error while getting jobs ", e);
+ return null;
+ }
+ }
+
+ /**
+ * Get the handle given the job name.
+ *
+ * @param jobName
+ * @return
+ */
+ public List<SchedulerJobHandle> getJob(String jobName) {
try {
- return store.getJobs(username, state == null ? null : state.name(), startTime, endTime);
+ return store.getJobsByName(jobName);
} catch (SQLException e) {
log.error("Error while getting jobs ", e);
return null;
@@ -236,18 +279,22 @@ public class SchedulerDAO {
public abstract static class SchedulerDBStore {
protected static final String JOB_TABLE = "job_table";
protected static final String JOB_INSTANCE_TABLE = "job_instance_table";
+ protected static final String JOB_INSTANCE_RUN_TABLE = "job_instance_run_table";
protected static final String COLUMN_ID = "id";
+ protected static final String COLUMN_RUN_ID = "runid";
protected static final String COLUMN_JOB = "job";
protected static final String COLUMN_USER = "username";
- protected static final String COLUMN_STATE = "state";
+ protected static final String COLUMN_STATUS = "status";
protected static final String COLUMN_CREATED_ON = "createdon";
+ protected static final String COLUMN_SCHEDULE_TIME = "schedultime";
protected static final String COLUMN_MODIFIED_ON = "modifiedon";
protected static final String COLUMN_JOB_ID = "jobid";
protected static final String COLUMN_SESSION_HANDLE = "sessionhandle";
protected static final String COLUMN_START_TIME = "starttime";
protected static final String COLUMN_END_TIME = "endtime";
protected static final String COLUMN_RESULT_PATH = "resultpath";
- protected static final String COLUMN_QUERY = "query";
+ protected static final String COLUMN_QUERY_HANDLE = "queryhandle";
+ protected static final String COLUMN_JOB_NAME = "jobname";
protected QueryRunner runner;
protected ObjectFactory jobFactory = new ObjectFactory();
// Generic multiple row handler for the fetch query.
@@ -289,7 +336,14 @@ public class SchedulerDAO {
*
* @throws SQLException
*/
- public abstract void createJobInstaceTable() throws SQLException;
+ public abstract void createJobInstanceTable() throws SQLException;
+
+ /**
+ * Creates the job instance run table
+ *
+ * @throws SQLException
+ */
+ public abstract void createJobInstanceRunTable() throws SQLException;
/**
* Inserts the Job info object into job table
@@ -299,10 +353,10 @@ public class SchedulerDAO {
* @throws SQLException
*/
public int insertIntoJobTable(SchedulerJobInfo jobInfo) throws SQLException {
- String insertSQL = "INSERT INTO " + JOB_TABLE + " VALUES(?,?,?,?,?,?)";
+ String insertSQL = "INSERT INTO " + JOB_TABLE + " VALUES(?,?,?,?,?,?,?)";
JAXBElement<XJob> xmlJob = jobFactory.createJob(jobInfo.getJob());
return runner.update(insertSQL, jobInfo.getId().toString(), ToXMLString.toString(xmlJob), jobInfo.getUserName(),
- jobInfo.getState().name(), jobInfo.getCreatedOn(), jobInfo.getModifiedOn());
+ jobInfo.getJobState().name(), jobInfo.getCreatedOn(), jobInfo.getModifiedOn(), jobInfo.getJob().getName());
}
/**
@@ -313,12 +367,19 @@ public class SchedulerDAO {
* @throws SQLException
*/
public int insertIntoJobInstanceTable(SchedulerJobInstanceInfo instanceInfo) throws SQLException {
- String insertSQL = "INSERT INTO " + JOB_INSTANCE_TABLE + " VALUES(?,?,?,?,?,?,?,?,?)";
+ String insertSQL = "INSERT INTO " + JOB_INSTANCE_TABLE + " VALUES(?,?,?)";
return runner
.update(insertSQL, instanceInfo.getId().getHandleIdString(), instanceInfo.getJobId().getHandleIdString(),
- instanceInfo.getSessionHandle().toString(), instanceInfo.getStartTime(), instanceInfo.getEndTime(),
- instanceInfo.getResultPath(), instanceInfo.getQuery(), instanceInfo.getState().name(),
- instanceInfo.getCreatedOn());
+ instanceInfo.getScheduleTime());
+ }
+
+ public int insertIntoJobInstanceRunTable(SchedulerJobInstanceRun instanceRun) throws SQLException {
+ String insetSQL = "INSERT INTO " + JOB_INSTANCE_RUN_TABLE + " VALUES(?,?,?,?,?,?,?,?)";
+ return runner.update(insetSQL, instanceRun.getHandle().getHandleIdString(), instanceRun.getRunId(),
+ instanceRun.getSessionHandle().toString(), instanceRun.getStartTime(), instanceRun.getEndTime(),
+ instanceRun.getResultPath(),
+ instanceRun.getQueryHandle() == null ? "" : instanceRun.getQueryHandle().getHandleIdString(),
+ instanceRun.getInstanceState().name());
}
/**
@@ -338,10 +399,11 @@ public class SchedulerDAO {
SchedulerJobHandle id = SchedulerJobHandle.fromString((String) jobInfo[0]);
XJob xJob = ToXMLString.valueOf((String) jobInfo[1], ObjectFactory.class);
String userName = (String) jobInfo[2];
- String state = (String) jobInfo[3];
+ String status = (String) jobInfo[3];
long createdOn = (Long) jobInfo[4];
long modifiedOn = (Long) jobInfo[5];
- return new SchedulerJobInfo(id, xJob, userName, SchedulerJobStatus.valueOf(state), createdOn, modifiedOn);
+ SchedulerJobState jobState = SchedulerJobState.valueOf(status);
+ return new SchedulerJobInfo(id, xJob, userName, jobState, createdOn, modifiedOn);
}
}
@@ -363,19 +425,37 @@ public class SchedulerDAO {
}
/**
- * Gets the job state
+ * Get the job handles given the job name
+ *
+ * @param jobname
+ * @return A list of handles
+ * @throws SQLException
+ */
+ public List<SchedulerJobHandle> getJobsByName(String jobname) throws SQLException {
+ String fetchSQL = "SELCET " + COLUMN_ID + " FROM " + JOB_TABLE + " WHERE " + COLUMN_JOB_NAME + "=?";
+ List<Object[]> result = runner.query(fetchSQL, multipleRowsHandler, jobname);
+ List<SchedulerJobHandle> resOut = new ArrayList<>();
+ for (int i = 0; i < result.size(); i++) {
+ Object[] row = result.get(i);
+ resOut.add(SchedulerJobHandle.fromString((String) row[0]));
+ }
+ return resOut;
+ }
+
+ /**
+ * Gets the job status
*
* @param id
- * @return SchedulerJobState
+ * @return SchedulerJobStatus
* @throws SQLException
*/
- public SchedulerJobStatus getJobState(String id) throws SQLException {
- String fetchSQL = "SELECT " + COLUMN_STATE + " FROM " + JOB_TABLE + " WHERE " + COLUMN_ID + "=?";
+ public SchedulerJobState getJobState(String id) throws SQLException {
+ String fetchSQL = "SELECT " + COLUMN_STATUS + " FROM " + JOB_TABLE + " WHERE " + COLUMN_ID + "=?";
List<Object[]> result = runner.query(fetchSQL, multipleRowsHandler, id);
if (result.size() == 0) {
return null;
} else {
- return SchedulerJobStatus.valueOf((String) result.get(0)[0]);
+ return SchedulerJobState.valueOf((String) result.get(0)[0]);
}
}
@@ -383,20 +463,20 @@ public class SchedulerDAO {
* Gets all the jobs which match the filter requirements.
*
* @param username
- * @param state
+ * @param status
* @param starttime
* @param endtime
* @return the list of job handles.
* @throws SQLException
*/
- public List<SchedulerJobHandle> getJobs(String username, String state, Long starttime, Long endtime)
+ public List<SchedulerJobHandle> getJobs(String username, String status, Long starttime, Long endtime)
throws SQLException {
String whereClause = "";
if (username != null && !username.isEmpty()) {
- whereClause += ((whereClause.isEmpty()) ? " WHERE " : " AND ") + COLUMN_USER + " = '" + username+"'";
+ whereClause += ((whereClause.isEmpty()) ? " WHERE " : " AND ") + COLUMN_USER + " = '" + username + "'";
}
- if (state != null && !state.isEmpty()) {
- whereClause += ((whereClause.isEmpty()) ? " WHERE " : " AND ") + COLUMN_STATE + " = '" + state + "'";
+ if (status != null && !status.isEmpty()) {
+ whereClause += ((whereClause.isEmpty()) ? " WHERE " : " AND ") + COLUMN_STATUS + " = '" + status + "'";
}
if (starttime != null && starttime > 0) {
whereClause += ((whereClause.isEmpty()) ? " WHERE " : " AND ") + COLUMN_CREATED_ON + " >= " + starttime;
@@ -432,19 +512,19 @@ public class SchedulerDAO {
}
/**
- * Updates the job state into the job table
+ * Updates the job status into the job table
*
* @param id
- * @param state
+ * @param status
* @param modifiedOn
* @return number of rows updated.
* @throws SQLException
*/
- public int updateJobState(String id, String state, long modifiedOn) throws SQLException {
+ public int updateJobStatus(String id, String status, long modifiedOn) throws SQLException {
String updateSQL =
- "UPDATE " + JOB_TABLE + " SET " + COLUMN_STATE + "=?, " + COLUMN_MODIFIED_ON + "=? " + " WHERE " + COLUMN_ID
+ "UPDATE " + JOB_TABLE + " SET " + COLUMN_STATUS + "=?, " + COLUMN_MODIFIED_ON + "=? " + " WHERE " + COLUMN_ID
+ "=?";
- return runner.update(updateSQL, state, modifiedOn, id);
+ return runner.update(updateSQL, status, modifiedOn, id);
}
/**
@@ -461,31 +541,54 @@ public class SchedulerDAO {
return null;
} else {
Object[] instanceInfo = result.get(0);
- SchedulerJobInstanceHandle id = SchedulerJobInstanceHandle.fromString((String) instanceInfo[0]);
- SchedulerJobHandle jobId = SchedulerJobHandle.fromString((String) instanceInfo[1]);
- LensSessionHandle sessionHandle = LensSessionHandle.valueOf((String) instanceInfo[2]);
- long starttime = (Long) instanceInfo[3];
- long endtime = (Long) instanceInfo[4];
- String resultPath = (String) instanceInfo[5];
- String query = (String) instanceInfo[6];
- SchedulerJobInstanceStatus state = SchedulerJobInstanceStatus.valueOf((String) instanceInfo[7]);
- long createdOn = (Long) instanceInfo[8];
- return new SchedulerJobInstanceInfo(id, jobId, sessionHandle, starttime, endtime, resultPath, query, state,
- createdOn);
+ return parseSchedulerInstance(instanceInfo);
+ }
+ }
+
+ private SchedulerJobInstanceInfo parseSchedulerInstance(Object[] instanceInfo) throws SQLException {
+ SchedulerJobInstanceHandle id = SchedulerJobInstanceHandle.fromString((String) instanceInfo[0]);
+ SchedulerJobHandle jobId = SchedulerJobHandle.fromString((String) instanceInfo[1]);
+ long createdOn = (Long) instanceInfo[2];
+ // Get the Runs
+ String fetchSQL = "SELECT * FROM " + JOB_INSTANCE_RUN_TABLE + " WHERE " + COLUMN_ID + "=?";
+ List<Object[]> instanceRuns = runner.query(fetchSQL, multipleRowsHandler, (String) instanceInfo[0]);
+ List<SchedulerJobInstanceRun> runList = new ArrayList<>();
+ for (Object[] run : instanceRuns) {
+ // run[0] will contain the instanceID
+ int runId = (Integer) run[1];
+ LensSessionHandle sessionHandle = LensSessionHandle.valueOf((String) run[2]);
+ long starttime = (Long) run[3];
+ long endtime = (Long) run[4];
+ String resultPath = (String) run[5];
+ String queryHandleString = (String) run[6];
+ QueryHandle queryHandle = null;
+ if (!queryHandleString.isEmpty()) {
+ queryHandle = QueryHandle.fromString((String) run[6]);
+ }
+ SchedulerJobInstanceState instanceStatus = SchedulerJobInstanceState.valueOf((String) run[7]);
+ SchedulerJobInstanceRun instanceRun = new SchedulerJobInstanceRun(id, runId, sessionHandle, starttime, endtime,
+ resultPath, queryHandle, instanceStatus);
+ runList.add(instanceRun);
}
+ return new SchedulerJobInstanceInfo(id, jobId, createdOn, runList);
}
/**
- * Updates the state of a job instance.
+ * Updates the status of a job instance.
*
- * @param id
- * @param state
+ * @param instanceRun
* @return number of rows updated.
* @throws SQLException
*/
- public int updateJobInstanceState(String id, String state) throws SQLException {
- String updateSQL = "UPDATE " + JOB_INSTANCE_TABLE + " SET " + COLUMN_STATE + "=?" + " WHERE " + COLUMN_ID + "=?";
- return runner.update(updateSQL, state, id);
+ public int updateJobInstanceRun(SchedulerJobInstanceRun instanceRun) throws SQLException {
+ String updateSQL =
+ "UPDATE " + JOB_INSTANCE_RUN_TABLE + " SET " + COLUMN_END_TIME + "=?, " + COLUMN_RESULT_PATH + "=?, "
+ + COLUMN_QUERY_HANDLE + "=?, " + COLUMN_STATUS + "=?" + " WHERE " + COLUMN_ID + "=? AND " + COLUMN_RUN_ID
+ + "=?";
+
+ return runner.update(updateSQL, instanceRun.getEndTime(), instanceRun.getResultPath(),
+ instanceRun.getQueryHandle() == null ? "" : instanceRun.getQueryHandle().getHandleIdString(),
+ instanceRun.getInstanceState().name(), instanceRun.getHandle().getHandleIdString(), instanceRun.getRunId());
}
/**
@@ -495,17 +598,33 @@ public class SchedulerDAO {
* @return List of SchedulerJobInstanceHandle
* @throws SQLException
*/
- public List<SchedulerJobInstanceHandle> getAllJobInstances(String jobId) throws SQLException {
- String fetchSQL = "SELECT " + COLUMN_ID + " FROM " + JOB_INSTANCE_TABLE + " WHERE " + COLUMN_JOB_ID + "=?";
+ public List<SchedulerJobInstanceInfo> getAllJobInstances(String jobId) throws SQLException {
+ String fetchSQL = "SELECT * FROM " + JOB_INSTANCE_TABLE + " WHERE " + COLUMN_JOB_ID + "=?";
List<Object[]> result = runner.query(fetchSQL, multipleRowsHandler, jobId);
- List<SchedulerJobInstanceHandle> resOut = new ArrayList<>();
+ List<SchedulerJobInstanceInfo> resOut = new ArrayList<>();
for (int i = 0; i < result.size(); i++) {
Object[] row = result.get(i);
- resOut.add(SchedulerJobInstanceHandle.fromString((String) row[0]));
+ resOut.add(parseSchedulerInstance(row));
}
return resOut;
}
+ /**
+ * Get the user who submitted the job.
+ *
+ * @param id
+ * @return
+ * @throws SQLException
+ */
+ public String getUser(String id) throws SQLException {
+ String fetchSQL = "SELECT " + COLUMN_USER + " FROM " + JOB_TABLE + " WHERE " + COLUMN_ID + "=?";
+ List<Object[]> result = runner.query(fetchSQL, multipleRowsHandler, id);
+ if (result.size() == 0) {
+ return null;
+ } else {
+ return (String) result.get(0)[0];
+ }
+ }
}
/**
@@ -519,8 +638,9 @@ public class SchedulerDAO {
public void createJobTable() throws SQLException {
String createSQL =
"CREATE TABLE IF NOT EXISTS " + JOB_TABLE + "( " + COLUMN_ID + " VARCHAR(255) NOT NULL," + COLUMN_JOB
- + " TEXT," + COLUMN_USER + " VARCHAR(255)," + COLUMN_STATE + " VARCHAR(20)," + COLUMN_CREATED_ON
- + " BIGINT, " + COLUMN_MODIFIED_ON + " BIGINT, " + " PRIMARY KEY ( " + COLUMN_ID + ")" + ")";
+ + " TEXT," + COLUMN_USER + " VARCHAR(255)," + COLUMN_STATUS + " VARCHAR(20)," + COLUMN_CREATED_ON
+ + " BIGINT, " + COLUMN_MODIFIED_ON + " BIGINT, " + COLUMN_JOB_NAME + " VARCHAR(255), " + " PRIMARY KEY ( "
+ + COLUMN_ID + ")" + ")";
runner.update(createSQL);
}
@@ -528,13 +648,25 @@ public class SchedulerDAO {
* {@inheritDoc}
*/
@Override
- public void createJobInstaceTable() throws SQLException {
+ public void createJobInstanceTable() throws SQLException {
String createSQL =
"CREATE TABLE IF NOT EXISTS " + JOB_INSTANCE_TABLE + "( " + COLUMN_ID + " VARCHAR(255) NOT NULL, "
- + COLUMN_JOB_ID + " VARCHAR(255) NOT NULL, " + COLUMN_SESSION_HANDLE + " VARCHAR(255), "
- + COLUMN_START_TIME + " BIGINT, " + COLUMN_END_TIME + " BIGINT, " + COLUMN_RESULT_PATH + " TEXT, "
- + COLUMN_QUERY + " TEXT, " + COLUMN_STATE + " VARCHAR(20), " + COLUMN_CREATED_ON + " BIGINT, "
- + " PRIMARY KEY ( " + COLUMN_ID + ")" + ")";
+ + COLUMN_JOB_ID + " VARCHAR(255) NOT NULL, " + COLUMN_SCHEDULE_TIME + " BIGINT, " + " PRIMARY KEY ( "
+ + COLUMN_ID + ")" + ")";
+ runner.update(createSQL);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void createJobInstanceRunTable() throws SQLException {
+ String createSQL =
+ "CREATE TABLE IF NOT EXISTS " + JOB_INSTANCE_RUN_TABLE + "( " + COLUMN_ID + " VARCHAR(255) NOT NULL, "
+ + COLUMN_RUN_ID + " INT NOT NULL, " + COLUMN_SESSION_HANDLE + " VARCHAR(255), " + COLUMN_START_TIME
+ + " BIGINT, " + COLUMN_END_TIME + " BIGINT, " + COLUMN_RESULT_PATH + " TEXT, " + COLUMN_QUERY_HANDLE
+ + " TEXT, " + COLUMN_STATUS + " VARCHAR(20), " + " PRIMARY KEY ( " + COLUMN_ID + ", " + COLUMN_RUN_ID
+ + ")" + ")";
runner.update(createSQL);
}
}
@@ -550,8 +682,9 @@ public class SchedulerDAO {
public void createJobTable() throws SQLException {
String createSQL =
"CREATE TABLE IF NOT EXISTS " + JOB_TABLE + "( " + COLUMN_ID + " VARCHAR(255) NOT NULL," + COLUMN_JOB
- + " VARCHAR(1024)," + COLUMN_USER + " VARCHAR(255)," + COLUMN_STATE + " VARCHAR(20)," + COLUMN_CREATED_ON
- + " BIGINT, " + COLUMN_MODIFIED_ON + " BIGINT, " + " PRIMARY KEY ( " + COLUMN_ID + ")" + ")";
+ + " VARCHAR(1024)," + COLUMN_USER + " VARCHAR(255)," + COLUMN_STATUS + " VARCHAR(20)," + COLUMN_CREATED_ON
+ + " BIGINT, " + COLUMN_MODIFIED_ON + " BIGINT, " + COLUMN_JOB_NAME + " VARCHAR(255), " + " PRIMARY KEY ( "
+ + COLUMN_ID + ")" + ")";
runner.update(createSQL);
}
@@ -559,13 +692,25 @@ public class SchedulerDAO {
* {@inheritDoc}
*/
@Override
- public void createJobInstaceTable() throws SQLException {
+ public void createJobInstanceTable() throws SQLException {
String createSQL =
"CREATE TABLE IF NOT EXISTS " + JOB_INSTANCE_TABLE + "( " + COLUMN_ID + " VARCHAR(255) NOT NULL, "
- + COLUMN_JOB_ID + " VARCHAR(255) NOT NULL, " + COLUMN_SESSION_HANDLE + " VARCHAR(255), "
- + COLUMN_START_TIME + " BIGINT, " + COLUMN_END_TIME + " BIGINT, " + COLUMN_RESULT_PATH + " VARCHAR(1024),"
- + COLUMN_QUERY + " VARCHAR(1024), " + COLUMN_STATE + " VARCHAR(20), " + COLUMN_CREATED_ON + " BIGINT, "
- + " PRIMARY KEY ( " + COLUMN_ID + ")" + ")";
+ + COLUMN_JOB_ID + " VARCHAR(255) NOT NULL, " + COLUMN_SCHEDULE_TIME + " BIGINT, " + " PRIMARY KEY ( "
+ + COLUMN_ID + ")" + ")";
+ runner.update(createSQL);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void createJobInstanceRunTable() throws SQLException {
+ String createSQL =
+ "CREATE TABLE IF NOT EXISTS " + JOB_INSTANCE_RUN_TABLE + "( " + COLUMN_ID + " VARCHAR(255) NOT NULL, "
+ + COLUMN_RUN_ID + " INT NOT NULL, " + COLUMN_SESSION_HANDLE + " VARCHAR(255), " + COLUMN_START_TIME
+ + " BIGINT, " + COLUMN_END_TIME + " BIGINT, " + COLUMN_RESULT_PATH + " VARCHAR(1024),"
+ + COLUMN_QUERY_HANDLE + " VARCHAR(1024), " + COLUMN_STATUS + " VARCHAR(20), " + " PRIMARY KEY ( "
+ + COLUMN_ID + ", " + COLUMN_RUN_ID + " )" + ")";
runner.update(createSQL);
}
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerEventListener.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerEventListener.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerEventListener.java
new file mode 100644
index 0000000..7323add
--- /dev/null
+++ b/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerEventListener.java
@@ -0,0 +1,174 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.server.scheduler;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.lens.api.LensConf;
+import org.apache.lens.api.LensSessionHandle;
+import org.apache.lens.api.error.InvalidStateTransitionException;
+import org.apache.lens.api.query.QueryHandle;
+import org.apache.lens.api.scheduler.*;
+import org.apache.lens.server.LensServices;
+import org.apache.lens.server.api.LensConfConstants;
+import org.apache.lens.server.api.error.LensException;
+import org.apache.lens.server.api.events.AsyncEventListener;
+import org.apache.lens.server.api.events.SchedulerAlarmEvent;
+import org.apache.lens.server.api.query.QueryExecutionService;
+import org.apache.lens.server.api.scheduler.SchedulerService;
+import org.apache.lens.server.query.QueryExecutionServiceImpl;
+import org.apache.lens.server.util.UtilityMethods;
+
+import org.joda.time.DateTime;
+
+import com.google.common.annotations.VisibleForTesting;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SchedulerEventListener extends AsyncEventListener<SchedulerAlarmEvent> {
+ private static final int CORE_POOL_SIZE = 10;
+ private static final String JOB_INSTANCE_ID_KEY = "job_instance_key";
+ @Getter
+ @Setter
+ @VisibleForTesting
+ protected QueryExecutionService queryService;
+ private SchedulerDAO schedulerDAO;
+ private SchedulerService schedulerService;
+
+ public SchedulerEventListener(SchedulerDAO schedulerDAO) {
+ super(CORE_POOL_SIZE);
+ this.queryService = LensServices.get().getService(QueryExecutionService.NAME);
+ this.schedulerService = LensServices.get().getService(SchedulerService.NAME);
+ this.schedulerDAO = schedulerDAO;
+ }
+
+ /**
+ * @param event the event
+ */
+ @Override
+ public void process(SchedulerAlarmEvent event) {
+ DateTime scheduledTime = event.getNominalTime();
+ SchedulerJobHandle jobHandle = event.getJobHandle();
+ if (event.getType() == SchedulerAlarmEvent.EventType.EXPIRE) {
+ try {
+ schedulerService.expireJob(null, jobHandle);
+ } catch (LensException e) {
+ log.error("Error while expiring the job", e);
+ }
+ return;
+ }
+ /*
+ * Get the job from the store.
+ * Create an instance.
+ * Store the instance.
+ * Try to run the instance.
+ * If successfully submitted change the status to running.
+ * Otherwise update the status to killed.
+ */
+ //TODO: Get the job status and if it is not Scheduled, don't do anything.
+ XJob job = schedulerDAO.getJob(jobHandle);
+ String user = schedulerDAO.getUser(jobHandle);
+ SchedulerJobInstanceHandle instanceHandle = event.getPreviousInstance() == null
+ ? UtilityMethods.generateSchedulerJobInstanceHandle()
+ : event.getPreviousInstance();
+ Map<String, String> conf = new HashMap<>();
+ LensSessionHandle sessionHandle = null;
+ try {
+ // Open the session with the user who submitted the job.
+ sessionHandle = ((QueryExecutionServiceImpl) LensServices.get().getService(QueryExecutionServiceImpl.NAME))
+ .openSession(user, "dummy", conf, false);
+ } catch (LensException e) {
+ log.error("Error occurred while opening a session ", e);
+ return;
+ }
+ SchedulerJobInstanceInfo instance = null;
+ SchedulerJobInstanceRun run = null;
+ // Session needs to be closed after the launch.
+ try {
+ long scheduledTimeMillis = scheduledTime.getMillis();
+ String query = job.getExecution().getQuery().getQuery();
+ List<MapType> jobConf = job.getExecution().getQuery().getConf();
+ LensConf queryConf = new LensConf();
+ for (MapType element : jobConf) {
+ queryConf.addProperty(element.getKey(), element.getValue());
+ }
+ queryConf.addProperty(JOB_INSTANCE_ID_KEY, instanceHandle.getHandleId());
+ // Current time is used for resolving date.
+ queryConf.addProperty(LensConfConstants.QUERY_CURRENT_TIME_IN_MILLIS, scheduledTime.getMillis());
+ String queryName = job.getName();
+ queryName += "-" + scheduledTime.getMillis();
+ // If the instance is new then create otherwise get from the store
+ if (event.getPreviousInstance() == null) {
+ instance = new SchedulerJobInstanceInfo(instanceHandle, jobHandle, scheduledTimeMillis,
+ new ArrayList<SchedulerJobInstanceRun>());
+ } else {
+ instance = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle);
+ }
+ // Next run of the instance
+ long currentTime = System.currentTimeMillis();
+ run = new SchedulerJobInstanceRun(instanceHandle, instance.getInstanceRunList().size() + 1, sessionHandle,
+ currentTime, 0, "N/A", null, SchedulerJobInstanceState.WAITING);
+ instance.getInstanceRunList().add(run);
+ boolean success;
+ if (event.getPreviousInstance() == null) {
+ success = schedulerDAO.storeJobInstance(instance) == 1;
+ if (!success) {
+ log.error(
+ "Exception occurred while storing the instance for instance handle " + instance + " of job " + jobHandle);
+ return;
+ }
+ }
+ success = schedulerDAO.storeJobInstanceRun(run) == 1;
+ if (!success) {
+ log.error(
+ "Exception occurred while storing the instance for instance handle " + instance + " of job " + jobHandle);
+ return;
+ }
+
+ QueryHandle handle = queryService.executeAsync(sessionHandle, query, queryConf, queryName);
+ run.setQueryHandle(handle);
+ run.setInstanceState(run.getInstanceState().nextTransition(SchedulerJobInstanceEvent.ON_RUN));
+ run.setEndTime(System.currentTimeMillis());
+ schedulerDAO.updateJobInstanceRun(run);
+ } catch (LensException | InvalidStateTransitionException e) {
+ log.error(
+ "Exception occurred while launching the job instance for " + jobHandle + " for nominal time " + scheduledTime
+ .getMillis(), e);
+ try {
+ run.setInstanceState(run.getInstanceState().nextTransition(SchedulerJobInstanceEvent.ON_FAILURE));
+ run.setEndTime(System.currentTimeMillis());
+ schedulerDAO.updateJobInstanceRun(run);
+ } catch (InvalidStateTransitionException e1) {
+ log.error("Can't make transition for instance " + instance.getId() + " of job " + instance.getJobId(), e);
+ }
+ } finally {
+ try {
+ ((QueryExecutionServiceImpl) LensServices.get().getService(QueryExecutionServiceImpl.NAME))
+ .closeSession(sessionHandle);
+ } catch (LensException e) {
+ log.error("Error closing session ", e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerQueryEventListener.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerQueryEventListener.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerQueryEventListener.java
new file mode 100644
index 0000000..5b12720
--- /dev/null
+++ b/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerQueryEventListener.java
@@ -0,0 +1,87 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.lens.server.scheduler;
+
+import java.util.List;
+
+import org.apache.lens.api.error.InvalidStateTransitionException;
+import org.apache.lens.api.query.QueryStatus;
+import org.apache.lens.api.scheduler.*;
+import org.apache.lens.server.api.events.AsyncEventListener;
+import org.apache.lens.server.api.query.QueryContext;
+import org.apache.lens.server.api.query.QueryEnded;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SchedulerQueryEventListener extends AsyncEventListener<QueryEnded> {
+ private static final String JOB_INSTANCE_ID_KEY = "job_instance_key";
+ private static final int CORE_POOL_SIZE = 10;
+ private SchedulerDAO schedulerDAO;
+
+ public SchedulerQueryEventListener(SchedulerDAO schedulerDAO) {
+ super(CORE_POOL_SIZE);
+ this.schedulerDAO = schedulerDAO;
+ }
+
+ @Override
+ public void process(QueryEnded event) {
+ if (event.getCurrentValue() == QueryStatus.Status.CLOSED) {
+ return;
+ }
+ QueryContext queryContext = event.getQueryContext();
+ if (queryContext == null) {
+ log.warn("Could not find the context for {} for event:{}.", event.getQueryHandle(), event.getCurrentValue());
+ return;
+ }
+ String instanceHandle = queryContext.getConf().get(JOB_INSTANCE_ID_KEY);
+ if (instanceHandle == null) {
+ // Nothing to do
+ return;
+ }
+ SchedulerJobInstanceInfo info = schedulerDAO
+ .getSchedulerJobInstanceInfo(SchedulerJobInstanceHandle.fromString(instanceHandle));
+ List<SchedulerJobInstanceRun> runList = info.getInstanceRunList();
+ if (runList.size() == 0) {
+ log.error("No instance run for " + instanceHandle + " with query " + queryContext.getQueryHandle());
+ return;
+ }
+ SchedulerJobInstanceRun latestRun = runList.get(runList.size() - 1);
+ SchedulerJobInstanceState state = latestRun.getInstanceState();
+ try {
+ switch (event.getCurrentValue()) {
+ case CANCELED:
+ state = state.nextTransition(SchedulerJobInstanceEvent.ON_KILL);
+ break;
+ case SUCCESSFUL:
+ state = state.nextTransition(SchedulerJobInstanceEvent.ON_SUCCESS);
+ break;
+ case FAILED:
+ state = state.nextTransition(SchedulerJobInstanceEvent.ON_FAILURE);
+ break;
+ }
+ latestRun.setEndTime(System.currentTimeMillis());
+ latestRun.setInstanceState(state);
+ latestRun.setResultPath(queryContext.getDriverResultPath());
+ schedulerDAO.updateJobInstanceRun(latestRun);
+ } catch (InvalidStateTransitionException e) {
+ log.error("Instance Transition Failed ", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerServiceImpl.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerServiceImpl.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerServiceImpl.java
index 3952671..14ca32d 100644
--- a/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerServiceImpl.java
+++ b/lens-server/src/main/java/org/apache/lens/server/scheduler/SchedulerServiceImpl.java
@@ -6,9 +6,9 @@
* 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
* 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
@@ -20,43 +20,83 @@ package org.apache.lens.server.scheduler;
import java.util.Collection;
import java.util.List;
-import java.util.UUID;
+import org.apache.lens.api.LensConf;
import org.apache.lens.api.LensSessionHandle;
+import org.apache.lens.api.error.InvalidStateTransitionException;
+import org.apache.lens.api.query.QueryHandle;
import org.apache.lens.api.scheduler.*;
+import org.apache.lens.cube.parse.CubeQueryConfUtil;
import org.apache.lens.server.BaseLensService;
-import org.apache.lens.server.LensServerConf;
+import org.apache.lens.server.LensServices;
import org.apache.lens.server.api.error.LensException;
+import org.apache.lens.server.api.events.SchedulerAlarmEvent;
import org.apache.lens.server.api.health.HealthStatus;
+import org.apache.lens.server.api.query.QueryEnded;
+import org.apache.lens.server.api.query.QueryExecutionService;
import org.apache.lens.server.api.scheduler.SchedulerService;
import org.apache.lens.server.session.LensSessionImpl;
+import org.apache.lens.server.util.UtilityMethods;
+import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hive.service.cli.CLIService;
+import org.joda.time.DateTime;
+
+import com.google.common.annotations.VisibleForTesting;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+
/**
* This class handles all the scheduler operations.
*/
+@Slf4j
public class SchedulerServiceImpl extends BaseLensService implements SchedulerService {
- // get the state store
- private SchedulerDAO schedulerDAO;
+ @Getter
+ @Setter
+ @VisibleForTesting
+ protected QueryExecutionService queryService;
+ @Getter
+ @VisibleForTesting
+ protected SchedulerEventListener schedulerEventListener;
+ @Getter
+ @VisibleForTesting
+ protected SchedulerQueryEventListener schedulerQueryEventListener;
+ @Getter
+ @VisibleForTesting
+ protected SchedulerDAO schedulerDAO;
+ private AlarmService alarmService;
- private LensScheduler scheduler;
/**
- * The constant name for scheduler service.
+ * Instantiates a new scheduler service.
+ *
+ * @param cliService the cli service
*/
- public static final String NAME = "scheduler";
-
public SchedulerServiceImpl(CLIService cliService) throws LensException {
super(NAME, cliService);
- this.schedulerDAO = new SchedulerDAO(LensServerConf.getHiveConf());
- this.scheduler = LensScheduler.get();
}
- public SchedulerServiceImpl(CLIService cliService, SchedulerDAO schedulerDAO) {
- super(NAME, cliService);
- this.schedulerDAO = schedulerDAO;
- this.scheduler = LensScheduler.get();
+ public synchronized void init(HiveConf hiveConf) {
+ super.init(hiveConf);
+ try {
+ schedulerDAO = new SchedulerDAO(hiveConf);
+ alarmService = LensServices.get().getService(AlarmService.NAME);
+ queryService = LensServices.get().getService(QueryExecutionService.NAME);
+ // Get the listeners' classes from the configuration.
+ this.schedulerEventListener = new SchedulerEventListener(schedulerDAO);
+ this.schedulerQueryEventListener = new SchedulerQueryEventListener(schedulerDAO);
+ getEventService().addListenerForType(schedulerEventListener, SchedulerAlarmEvent.class);
+ getEventService().addListenerForType(schedulerQueryEventListener, QueryEnded.class);
+ } catch (LensException e) {
+ log.error("Error Initialising Scheduler-service", e);
+ }
+ }
+
+ @Override
+ public synchronized void start() {
+ super.start();
}
/**
@@ -65,8 +105,8 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
@Override
public HealthStatus getHealthStatus() {
return this.getServiceState().equals(STATE.STARTED)
- ? new HealthStatus(true, "Scheduler service is healthy.")
- : new HealthStatus(false, "Scheduler service is down.");
+ ? new HealthStatus(true, "Scheduler service is healthy.")
+ : new HealthStatus(false, "Scheduler service is down.");
}
/**
@@ -74,27 +114,57 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
*/
@Override
public SchedulerJobHandle submitJob(LensSessionHandle sessionHandle, XJob job) throws LensException {
- //TBD place holder code
LensSessionImpl session = getSession(sessionHandle);
- return null;
+ // Validate XJob
+ validateJob(job);
+ SchedulerJobHandle handle = UtilityMethods.generateSchedulerJobHandle();
+ long createdOn = System.currentTimeMillis();
+ long modifiedOn = createdOn;
+ SchedulerJobInfo info = new SchedulerJobInfo(handle, job, session.getLoggedInUser(), SchedulerJobState.NEW,
+ createdOn, modifiedOn);
+ if (schedulerDAO.storeJob(info) == 1) {
+ return handle;
+ } else {
+ throw new LensException("Could not Submit the job");
+ }
+ }
+
+ private void validateJob(XJob job) throws LensException {
}
/**
* {@inheritDoc}
*/
@Override
- public void scheduleJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException {
- //TBD place holder code
- // send the schedule request to the scheduler.
- UUID externalID = jobHandle.getHandleId();
- // get the job from the database and schedule
+ public boolean scheduleJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException {
+ SchedulerJobInfo jobInfo = schedulerDAO.getSchedulerJobInfo(jobHandle);
+ XJob job = jobInfo.getJob();
+ DateTime start = new DateTime(job.getStartTime().toGregorianCalendar().getTime());
+ DateTime end = new DateTime(job.getEndTime().toGregorianCalendar().getTime());
+ XFrequency frequency = job.getTrigger().getFrequency();
+ // check query
+ checkQuery(sessionHandle, job);
+ alarmService.schedule(start, end, frequency, jobHandle.getHandleIdString());
+ return setStateOfJob(jobHandle, SchedulerJobEvent.ON_SCHEDULE) == 1;
+ }
+
+ private void checkQuery(LensSessionHandle sessionHandle, XJob job) throws LensException {
+ List<MapType> jobConf = job.getExecution().getQuery().getConf();
+ LensConf queryConf = new LensConf();
+ for (MapType element : jobConf) {
+ queryConf.addProperty(element.getKey(), element.getValue());
+ }
+ queryConf.addProperty(CubeQueryConfUtil.FAIL_QUERY_ON_PARTIAL_DATA, false);
+ queryService.estimate(LensServices.get().getLogSegregationContext().getLogSegragationId(), sessionHandle,
+ job.getExecution().getQuery().getQuery(), queryConf);
+ return;
}
@Override
public SchedulerJobHandle submitAndScheduleJob(LensSessionHandle sessionHandle, XJob job) throws LensException {
- //TBD place holder code
- // take job, validate it, submit it(check duplicate, persist it), schedule it.
- return null;
+ SchedulerJobHandle handle = submitJob(sessionHandle, job);
+ scheduleJob(sessionHandle, handle);
+ return handle;
}
/**
@@ -102,39 +172,42 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
*/
@Override
public XJob getJobDefinition(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException {
- //TBD place holder code
- // get the job definition from the persisted store, return it.
- return null;
+ return schedulerDAO.getJob(jobHandle);
}
/**
* {@inheritDoc}
*/
@Override
- public SchedulerJobInfo getJobDetails(LensSessionHandle sessionHandle,
- SchedulerJobHandle jobHandle) throws LensException {
- //TBD place holder code
- return null;
+ public SchedulerJobInfo getJobDetails(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle)
+ throws LensException {
+ return schedulerDAO.getSchedulerJobInfo(jobHandle);
}
/**
* {@inheritDoc}
*/
@Override
- public boolean updateJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle,
- XJob newJobDefinition) throws LensException {
- //TBD place holder code
- XJob job = schedulerDAO.getJob(jobHandle);
- return false;
+ public boolean updateJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle, XJob newJobDefinition)
+ throws LensException {
+ SchedulerJobInfo jobInfo = schedulerDAO.getSchedulerJobInfo(jobHandle);
+ // This will allow only the job definition and configuration change.
+ // TODO: fix start and end time changes
+ jobInfo.setJob(newJobDefinition);
+ jobInfo.setModifiedOn(System.currentTimeMillis());
+ int updated = schedulerDAO.updateJob(jobInfo);
+ return updated > 0;
}
/**
- *
* {@inheritDoc}
*/
@Override
- public void expireJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException {
- //TBD place holder code
+ public boolean expireJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException {
+ if (alarmService.checkExists(jobHandle)) {
+ alarmService.unSchedule(jobHandle);
+ }
+ return setStateOfJob(jobHandle, SchedulerJobEvent.ON_EXPIRE) == 1;
}
/**
@@ -142,8 +215,8 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
*/
@Override
public boolean suspendJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException {
- //TBD place holder code
- return false;
+ alarmService.pauseJob(jobHandle);
+ return setStateOfJob(jobHandle, SchedulerJobEvent.ON_SUSPEND) == 1;
}
/**
@@ -151,8 +224,8 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
*/
@Override
public boolean resumeJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException {
- // TBD place holder code
- return false;
+ alarmService.resumeJob(jobHandle);
+ return setStateOfJob(jobHandle, SchedulerJobEvent.ON_RESUME) == 1;
}
/**
@@ -160,20 +233,18 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
*/
@Override
public boolean deleteJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle) throws LensException {
- // TBD place holder code
- // it should only be a soft delete. Later on we will make a purge service and that service will delete
- // all the soft delete things.
- return false;
+ if (alarmService.checkExists(jobHandle)) {
+ alarmService.unSchedule(jobHandle);
+ }
+ return setStateOfJob(jobHandle, SchedulerJobEvent.ON_DELETE) == 1;
}
/**
* {@inheritDoc}
*/
@Override
- public Collection<SchedulerJobStats> getAllJobStats(LensSessionHandle sessionHandle, String state, String userName,
- long startTime, long endTime) throws LensException {
- // TBD place holder code
- // validate that the state is a valid state (enum)
+ public Collection<SchedulerJobStats> getAllJobStats(LensSessionHandle sessionHandle, String state, String user,
+ String jobName, long startTime, long endTime) throws LensException {
return null;
}
@@ -182,9 +253,7 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
*/
@Override
public SchedulerJobStats getJobStats(LensSessionHandle sessionHandle, SchedulerJobHandle handle, String state,
- long startTime, long endTime) throws LensException {
- // TBD place holder code
- // validate that the state is a valid state (enum)
+ long startTime, long endTime) throws LensException {
return null;
}
@@ -192,39 +261,78 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
* {@inheritDoc}
*/
@Override
- public boolean rerunInstance(LensSessionHandle sessionHandle,
- SchedulerJobInstanceHandle instanceHandle) throws LensException {
- //TBD place holder code
- return false;
+ public boolean rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle)
+ throws LensException {
+ SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle);
+ if (schedulerDAO.getJobState(instanceInfo.getJobId()) != SchedulerJobState.SCHEDULED) {
+ throw new LensException("Job with handle " + instanceInfo.getJobId() + " is not scheduled");
+ }
+ // Get the latest run.
+ List<SchedulerJobInstanceRun> runList = instanceInfo.getInstanceRunList();
+ if (runList.size() == 0) {
+ throw new LensException("Job instance " + instanceHandle + " is not yet run");
+ }
+ SchedulerJobInstanceRun latestRun = runList.get(runList.size() - 1);
+ // This call is for the test that it can be re run.
+ try {
+ latestRun.getInstanceState().nextTransition(SchedulerJobInstanceEvent.ON_RERUN);
+ getEventService().notifyEvent(
+ new SchedulerAlarmEvent(instanceInfo.getJobId(), new DateTime(instanceInfo.getScheduleTime()),
+ SchedulerAlarmEvent.EventType.SCHEDULE, instanceHandle));
+ } catch (InvalidStateTransitionException e) {
+ throw new LensException("Invalid State Transition ", e);
+ }
+ return true;
}
/**
* {@inheritDoc}
*/
@Override
- public List<SchedulerJobInstanceInfo> getJobInstances(LensSessionHandle sessionHandle,
- SchedulerJobHandle jobHandle, Long numResults) throws LensException {
- // TBD place holder code
- // By default return 100 results - make it configurable
- return null;
+ public List<SchedulerJobInstanceInfo> getJobInstances(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle,
+ Long numResults) throws LensException {
+ return schedulerDAO.getJobInstances(jobHandle);
}
@Override
- public boolean killInstance(LensSessionHandle sessionHandle,
- SchedulerJobInstanceHandle instanceHandle) throws LensException {
- // TBD place holder code
- return true;
+ public boolean killInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle)
+ throws LensException {
+ /**
+ * Get the query handle from the latest run.
+ */
+ SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle);
+ List<SchedulerJobInstanceRun> runList = instanceInfo.getInstanceRunList();
+ if (runList.size() == 0) {
+ throw new LensException("Job instance " + instanceHandle + " is not yet run");
+ }
+ SchedulerJobInstanceRun latestRun = runList.get(runList.size() - 1);
+ QueryHandle handle = latestRun.getQueryHandle();
+ if (handle.getHandleIdString().isEmpty()) {
+ return false;
+ }
+ // This will cause the QueryEnd event which will set the status of the instance to KILLED.
+ return queryService.cancelQuery(sessionHandle, handle);
}
-
/**
* {@inheritDoc}
*/
@Override
public SchedulerJobInstanceInfo getInstanceDetails(LensSessionHandle sessionHandle,
- SchedulerJobInstanceHandle instanceHandle) throws LensException {
- // TBD place holder code
- return null;
+ SchedulerJobInstanceHandle instanceHandle) throws LensException {
+ return schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle);
}
+ private int setStateOfJob(SchedulerJobHandle handle, SchedulerJobEvent event) throws LensException {
+ try {
+ SchedulerJobInfo info = schedulerDAO.getSchedulerJobInfo(handle);
+ SchedulerJobState currentState = info.getJobState();
+ SchedulerJobState nextState = currentState.nextTransition(event);
+ info.setJobState(nextState);
+ info.setModifiedOn(System.currentTimeMillis());
+ return schedulerDAO.updateJobStatus(info);
+ } catch (InvalidStateTransitionException e) {
+ throw new LensException("Invalid state ", e);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/notification/services/AlarmService.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/notification/services/AlarmService.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/notification/services/AlarmService.java
deleted file mode 100644
index a4cdd83..0000000
--- a/lens-server/src/main/java/org/apache/lens/server/scheduler/notification/services/AlarmService.java
+++ /dev/null
@@ -1,220 +0,0 @@
-/**
- * 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
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * 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.lens.server.scheduler.notification.services;
-
-import org.apache.lens.api.scheduler.SchedulerJobHandle;
-import org.apache.lens.api.scheduler.XFrequency;
-import org.apache.lens.api.scheduler.XFrequencyEnum;
-import org.apache.lens.server.LensServices;
-import org.apache.lens.server.api.LensService;
-import org.apache.lens.server.api.error.LensException;
-import org.apache.lens.server.api.events.LensEventService;
-import org.apache.lens.server.api.events.SchedulerAlarmEvent;
-import org.apache.lens.server.api.health.HealthStatus;
-
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hive.service.AbstractService;
-
-import org.joda.time.DateTime;
-import org.quartz.*;
-import org.quartz.impl.StdSchedulerFactory;
-
-import lombok.extern.slf4j.Slf4j;
-
-/**
- * This service is used primarily by Scheduler to get alarm notifications for scheduled queries.
- *
- * As a schedule this service accepts start time, frequency, end time and timeZone. It also requires the
- * {@link SchedulerJobHandle} which it sends as part of the
- * {@link org.apache.lens.server.api.events.SchedulerAlarmEvent} to inform the scheduler about the job for which
- * job the notification has been generated.
- */
-@Slf4j
-public class AlarmService extends AbstractService implements LensService {
-
- public static final String NAME = "alarm-service";
-
- private Scheduler scheduler;
-
- /**
- * True if the service started properly and is running fine, false otherwise.
- */
- private boolean isHealthy = true;
-
- /**
- * Contains the reason if service is not healthy.
- */
- private String healthCause;
-
- /**
- * Creates a new instance of AlarmService.
- *
- * @param name the name
- */
- public AlarmService(String name) {
- super(name);
- }
-
- @Override
- public HealthStatus getHealthStatus() {
- return isHealthy
- ? new HealthStatus(isHealthy, "Alarm service is healthy.")
- : new HealthStatus(isHealthy, healthCause);
- }
-
- public synchronized void init(HiveConf hiveConf) {
- super.init(hiveConf);
- try {
- this.scheduler = StdSchedulerFactory.getDefaultScheduler();
- } catch (SchedulerException e) {
- isHealthy = false;
- healthCause = "Failed to initialize the Quartz Scheduler for AlarmService.";
- log.error(healthCause, e);
- }
- }
-
- public synchronized void start() {
- try {
- scheduler.start();
- log.info("Alarm service started successfully!");
- } catch (SchedulerException e) {
- isHealthy = false;
- healthCause = "Failed to start the Quartz Scheduler for AlarmService.";
- log.error(healthCause, e);
- }
- }
-
- @Override
- public synchronized void stop() {
- try {
- scheduler.shutdown();
- log.info("Alarm Service stopped successfully.");
- } catch (SchedulerException e) {
- log.error("Failed to shut down the Quartz Scheduler for AlarmService.", e);
- }
- }
-
- /**
- * This method can be used by any consumer who wants to receive notifications during a time range at a given
- * frequency.
- *
- * This method is intended to be used by LensScheduler to subscribe for time based notifications to schedule queries.
- * On receiving a job to be scheduled LensScheduler will subscribe to all triggers required for the job, including
- * AlarmService for time based triggers.
- *
- * @param start start time for notifications
- * @param end end time for notifications
- * @param frequency Frequency to determine the frequency at which notification should be sent.
- * @param jobHandle Must be a unique jobHanlde across all consumers
- */
- public void schedule(DateTime start, DateTime end, XFrequency frequency, String jobHandle) throws LensException {
- // accept the schedule and then keep on sending the notifications for that schedule
- JobDataMap map = new JobDataMap();
- map.put("jobHandle", jobHandle);
-
- JobDetail job = JobBuilder.newJob(LensJob.class).withIdentity(jobHandle, "LensJobs")
- .usingJobData(map).build();
-
- Trigger trigger;
- if (frequency.getEnum() != null) { //for enum expression: create a trigger using calendar interval
- CalendarIntervalScheduleBuilder scheduleBuilder = CalendarIntervalScheduleBuilder.calendarIntervalSchedule()
- .withInterval(getTimeInterval(frequency.getEnum()), getTimeUnit(frequency.getEnum()))
- .withMisfireHandlingInstructionIgnoreMisfires();
- trigger = TriggerBuilder.newTrigger()
- .withIdentity(jobHandle, "AlarmService")
- .startAt(start.toDate())
- .endAt(end.toDate())
- .withSchedule(scheduleBuilder)
- .build();
- } else { // for cron expression create a cron trigger
- trigger = TriggerBuilder.newTrigger()
- .withIdentity(jobHandle, "AlarmService")
- .withSchedule(CronScheduleBuilder.cronSchedule(frequency.getCronExpression()))
- .build();
- }
-
- // Tell quartz to run the job using our trigger
- try {
- scheduler.scheduleJob(job, trigger);
- } catch (SchedulerException e) {
- log.error("Error scheduling job with jobHandle: {}", jobHandle);
- throw new LensException("Failed to schedule job with jobHandle: " + jobHandle, e);
- }
- }
-
-
- private int getTimeInterval(XFrequencyEnum frequencyEnum) {
- // since quarterly is not supported natively, we express it as 3 months
- return frequencyEnum == XFrequencyEnum.QUARTERLY ? 3 : 1;
- }
-
-
- // Maps the timeunit in entity specification to the one in Quartz DateBuilder
- private DateBuilder.IntervalUnit getTimeUnit(XFrequencyEnum frequencyEnum) {
- switch (frequencyEnum) {
-
- case DAILY:
- return DateBuilder.IntervalUnit.DAY;
-
- case WEEKLY:
- return DateBuilder.IntervalUnit.WEEK;
-
- case MONTHLY:
- return DateBuilder.IntervalUnit.MONTH;
-
- case QUARTERLY:
- return DateBuilder.IntervalUnit.MONTH;
-
- case YEARLY:
- return DateBuilder.IntervalUnit.YEAR;
-
- default:
- throw new IllegalArgumentException("Invalid frequency enum expression: " + frequencyEnum.name());
- }
- }
-
- public boolean unSchedule(SchedulerJobHandle jobHandle) throws LensException {
- // stop sending notifications for this job handle
- try {
- return scheduler.deleteJob(JobKey.jobKey(jobHandle.getHandleIdString(), "LensScheduler"));
- } catch (SchedulerException e) {
- log.error("Failed to remove alarm triggers for job with jobHandle: " + jobHandle, e);
- throw new LensException("Failed to remove alarm triggers for job with jobHandle: " + jobHandle, e);
- }
- }
-
- public static class LensJob implements Job {
-
- @Override
- public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
- JobDataMap data = jobExecutionContext.getMergedJobDataMap();
- DateTime nominalTime = new DateTime(jobExecutionContext.getScheduledFireTime());
- SchedulerJobHandle jobHandle = SchedulerJobHandle.fromString(data.getString("jobHandle"));
- SchedulerAlarmEvent alarmEvent = new SchedulerAlarmEvent(jobHandle, nominalTime);
- try {
- LensEventService eventService = LensServices.get().getService(LensEventService.NAME);
- eventService.notifyEvent(alarmEvent);
- } catch (LensException e) {
- log.error("Failed to notify SchedulerAlarmEvent for jobHandle: {} and nominalTime: {}",
- jobHandle.getHandleIdString(), nominalTime.toString(), e);
- throw new JobExecutionException("Failed to notify alarmEvent", e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/state/SchedulerJobInstanceState.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/state/SchedulerJobInstanceState.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/state/SchedulerJobInstanceState.java
deleted file mode 100644
index 95057e4..0000000
--- a/lens-server/src/main/java/org/apache/lens/server/scheduler/state/SchedulerJobInstanceState.java
+++ /dev/null
@@ -1,193 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.lens.server.scheduler.state;
-
-import org.apache.lens.api.scheduler.SchedulerJobInstanceStatus;
-import org.apache.lens.server.api.error.InvalidStateTransitionException;
-import org.apache.lens.server.api.scheduler.StateMachine;
-
-import lombok.Getter;
-import lombok.Setter;
-
-/**
- * State machine for transitions on Scheduler Jobs.
- */
-public class SchedulerJobInstanceState {
-
- public SchedulerJobInstanceState(SchedulerJobInstanceStatus status) {
- this.currentStatus = status;
- }
-
- public SchedulerJobInstanceState() {
- this.currentStatus = INITIAL_STATUS;
- }
-
- @Getter @Setter
- private SchedulerJobInstanceStatus currentStatus;
-
- private static final SchedulerJobInstanceStatus INITIAL_STATUS = SchedulerJobInstanceStatus.WAITING;
-
- public SchedulerJobInstanceStatus nextTransition(EVENT event) throws InvalidStateTransitionException {
- STATE currentState = STATE.valueOf(currentStatus.name());
- return SchedulerJobInstanceStatus.valueOf(currentState.nextTransition(event).name());
- }
-
- public enum STATE implements StateMachine<STATE, EVENT> {
- // repeating same operation will return the same state to ensure idempotent behavior.
- WAITING {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_CREATION:
- return this;
- case ON_CONDITIONS_MET:
- return STATE.LAUNCHED;
- case ON_TIME_OUT:
- return STATE.TIMED_OUT;
- case ON_RUN:
- return STATE.RUNNING;
- case ON_SUCCESS:
- return STATE.SUCCEEDED;
- case ON_FAILURE:
- return STATE.FAILED;
- case ON_KILL:
- return STATE.KILLED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- LAUNCHED {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_CONDITIONS_MET:
- return this;
- case ON_RUN:
- return STATE.RUNNING;
- case ON_SUCCESS:
- return STATE.SUCCEEDED;
- case ON_FAILURE:
- return STATE.FAILED;
- case ON_KILL:
- return STATE.KILLED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- RUNNING {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_RUN:
- return this;
- case ON_SUCCESS:
- return STATE.SUCCEEDED;
- case ON_FAILURE:
- return STATE.FAILED;
- case ON_KILL:
- return STATE.KILLED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- FAILED {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_FAILURE:
- return this;
- case ON_RERUN:
- return STATE.LAUNCHED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- SUCCEEDED {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_SUCCESS:
- return this;
- case ON_RERUN:
- return STATE.LAUNCHED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
-
- TIMED_OUT {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_TIME_OUT:
- return this;
- case ON_RERUN:
- return STATE.WAITING;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- KILLED {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_KILL:
- return this;
- case ON_RERUN:
- return STATE.LAUNCHED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- }
- }
-
- /**
- * All events(actions) which can happen on an instance of <code>SchedulerJob</code>.
- */
- public enum EVENT {
- ON_CREATION, // an instance is first considered by the scheduler.
- ON_TIME_OUT,
- ON_CONDITIONS_MET,
- ON_RUN,
- ON_SUCCESS,
- ON_FAILURE,
- ON_RERUN,
- ON_KILL
- }
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/state/SchedulerJobState.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/state/SchedulerJobState.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/state/SchedulerJobState.java
deleted file mode 100644
index d21cd05..0000000
--- a/lens-server/src/main/java/org/apache/lens/server/scheduler/state/SchedulerJobState.java
+++ /dev/null
@@ -1,150 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.lens.server.scheduler.state;
-
-import org.apache.lens.api.scheduler.SchedulerJobStatus;
-import org.apache.lens.server.api.error.InvalidStateTransitionException;
-import org.apache.lens.server.api.scheduler.StateMachine;
-
-import lombok.Getter;
-import lombok.Setter;
-
-/**
- * This class represents current state of a SchedulerJob and provides helper methods
- * for handling different events and lifecycle transition for a SchedulerJob.
- */
-public class SchedulerJobState {
-
- public SchedulerJobState(SchedulerJobStatus status) {
- this.currentStatus = status;
- }
- public SchedulerJobState() {
- this.currentStatus = INITIAL_STATUS;
- }
-
- @Getter @Setter
- private SchedulerJobStatus currentStatus;
-
- private static final SchedulerJobStatus INITIAL_STATUS = SchedulerJobStatus.NEW;
-
- public SchedulerJobStatus nextTransition(EVENT event) throws InvalidStateTransitionException {
- STATE currentState = STATE.valueOf(currentStatus.name());
- return SchedulerJobStatus.valueOf(currentState.nextTransition(event).name());
- }
-
- private enum STATE implements StateMachine<STATE, EVENT> {
- // repeating same operation will return the same state to ensure idempotent behavior.
- NEW {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_SUBMIT:
- return this;
- case ON_SCHEDULE:
- return STATE.SCHEDULED;
- case ON_EXPIRE:
- return STATE.EXPIRED;
- case ON_DELETE:
- return STATE.DELETED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- SCHEDULED {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_SCHEDULE:
- return this;
- case ON_SUSPEND:
- return STATE.SUSPENDED;
- case ON_EXPIRE:
- return STATE.EXPIRED;
- case ON_DELETE:
- return STATE.DELETED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- SUSPENDED {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_SUSPEND:
- return this;
- case ON_RESUME:
- return STATE.SCHEDULED;
- case ON_EXPIRE:
- return STATE.EXPIRED;
- case ON_DELETE:
- return STATE.DELETED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- EXPIRED {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_EXPIRE:
- return this;
- case ON_DELETE:
- return STATE.DELETED;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- },
-
- DELETED {
- @Override
- public STATE nextTransition(EVENT event) throws InvalidStateTransitionException {
- switch (event) {
- case ON_DELETE:
- return this;
- default:
- throw new InvalidStateTransitionException("Event: " + event.name() + " is not a valid event for state: "
- + this.name());
- }
- }
- }
- }
-
- /**
- * All events(actions) which can happen on a Scheduler Job.
- */
- public enum EVENT {
- ON_SUBMIT,
- ON_SCHEDULE,
- ON_SUSPEND,
- ON_RESUME,
- ON_EXPIRE,
- ON_DELETE
- }
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/scheduler/util/UtilityMethods.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/scheduler/util/UtilityMethods.java b/lens-server/src/main/java/org/apache/lens/server/scheduler/util/UtilityMethods.java
deleted file mode 100644
index 31783ad..0000000
--- a/lens-server/src/main/java/org/apache/lens/server/scheduler/util/UtilityMethods.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.lens.server.scheduler.util;
-
-import org.apache.lens.server.api.LensConfConstants;
-
-import org.apache.commons.dbcp.BasicDataSource;
-import org.apache.hadoop.conf.Configuration;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class UtilityMethods {
- private UtilityMethods() {
-
- }
-
- /**
- *
- * @param conf
- * @return
- */
- public static BasicDataSource getDataSourceFromConf(Configuration conf) {
- BasicDataSource basicDataSource = new BasicDataSource();
- basicDataSource.setDriverClassName(
- conf.get(LensConfConstants.SERVER_DB_DRIVER_NAME, LensConfConstants.DEFAULT_SERVER_DB_DRIVER_NAME));
- basicDataSource
- .setUrl(conf.get(LensConfConstants.SERVER_DB_JDBC_URL, LensConfConstants.DEFAULT_SERVER_DB_JDBC_URL));
- basicDataSource
- .setUsername(conf.get(LensConfConstants.SERVER_DB_JDBC_USER, LensConfConstants.DEFAULT_SERVER_DB_USER));
- basicDataSource
- .setPassword(conf.get(LensConfConstants.SERVER_DB_JDBC_PASS, LensConfConstants.DEFAULT_SERVER_DB_PASS));
- basicDataSource.setDefaultAutoCommit(true);
- return basicDataSource;
- }
-}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/session/LensSessionImpl.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/session/LensSessionImpl.java b/lens-server/src/main/java/org/apache/lens/server/session/LensSessionImpl.java
index db53924..3003ab7 100644
--- a/lens-server/src/main/java/org/apache/lens/server/session/LensSessionImpl.java
+++ b/lens-server/src/main/java/org/apache/lens/server/session/LensSessionImpl.java
@@ -57,7 +57,7 @@ import lombok.extern.slf4j.Slf4j;
* The Class LensSessionImpl.
*/
@Slf4j
-public class LensSessionImpl extends HiveSessionImpl {
+public class LensSessionImpl extends HiveSessionImpl implements AutoCloseable {
/** The persist info. */
private LensSessionPersistInfo persistInfo = new LensSessionPersistInfo();
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/java/org/apache/lens/server/util/UtilityMethods.java
----------------------------------------------------------------------
diff --git a/lens-server/src/main/java/org/apache/lens/server/util/UtilityMethods.java b/lens-server/src/main/java/org/apache/lens/server/util/UtilityMethods.java
index 63a7874..42d9fe0 100644
--- a/lens-server/src/main/java/org/apache/lens/server/util/UtilityMethods.java
+++ b/lens-server/src/main/java/org/apache/lens/server/util/UtilityMethods.java
@@ -6,9 +6,9 @@
* 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
* 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
@@ -26,9 +26,12 @@ import java.security.NoSuchAlgorithmException;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Map;
+import java.util.UUID;
import javax.sql.DataSource;
+import org.apache.lens.api.scheduler.SchedulerJobHandle;
+import org.apache.lens.api.scheduler.SchedulerJobInstanceHandle;
import org.apache.lens.server.api.LensConfConstants;
import org.apache.commons.dbcp.*;
@@ -41,7 +44,6 @@ import org.apache.hadoop.io.WritableUtils;
import lombok.extern.slf4j.Slf4j;
-
/**
* The Class UtilityMethods.
*/
@@ -138,30 +140,29 @@ public final class UtilityMethods {
*/
public static BasicDataSource getDataSourceFromConf(Configuration conf) {
BasicDataSource tmp = new BasicDataSource();
- tmp.setDriverClassName(conf.get(LensConfConstants.SERVER_DB_DRIVER_NAME,
- LensConfConstants.DEFAULT_SERVER_DB_DRIVER_NAME));
+ tmp.setDriverClassName(
+ conf.get(LensConfConstants.SERVER_DB_DRIVER_NAME, LensConfConstants.DEFAULT_SERVER_DB_DRIVER_NAME));
tmp.setUrl(conf.get(LensConfConstants.SERVER_DB_JDBC_URL, LensConfConstants.DEFAULT_SERVER_DB_JDBC_URL));
tmp.setUsername(conf.get(LensConfConstants.SERVER_DB_JDBC_USER, LensConfConstants.DEFAULT_SERVER_DB_USER));
tmp.setPassword(conf.get(LensConfConstants.SERVER_DB_JDBC_PASS, LensConfConstants.DEFAULT_SERVER_DB_PASS));
- tmp.setValidationQuery(conf.get(LensConfConstants.SERVER_DB_VALIDATION_QUERY,
- LensConfConstants.DEFAULT_SERVER_DB_VALIDATION_QUERY));
+ tmp.setValidationQuery(
+ conf.get(LensConfConstants.SERVER_DB_VALIDATION_QUERY, LensConfConstants.DEFAULT_SERVER_DB_VALIDATION_QUERY));
tmp.setDefaultAutoCommit(false);
return tmp;
}
public static DataSource getPoolingDataSourceFromConf(Configuration conf) {
final ConnectionFactory cf = new DriverManagerConnectionFactory(
- conf.get(LensConfConstants.SERVER_DB_JDBC_URL, LensConfConstants.DEFAULT_SERVER_DB_JDBC_URL),
- conf.get(LensConfConstants.SERVER_DB_JDBC_USER, LensConfConstants.DEFAULT_SERVER_DB_USER),
- conf.get(LensConfConstants.SERVER_DB_JDBC_PASS, LensConfConstants.DEFAULT_SERVER_DB_PASS));
+ conf.get(LensConfConstants.SERVER_DB_JDBC_URL, LensConfConstants.DEFAULT_SERVER_DB_JDBC_URL),
+ conf.get(LensConfConstants.SERVER_DB_JDBC_USER, LensConfConstants.DEFAULT_SERVER_DB_USER),
+ conf.get(LensConfConstants.SERVER_DB_JDBC_PASS, LensConfConstants.DEFAULT_SERVER_DB_PASS));
final GenericObjectPool connectionPool = new GenericObjectPool();
connectionPool.setTestOnBorrow(false);
connectionPool.setTestOnReturn(false);
connectionPool.setTestWhileIdle(true);
- new PoolableConnectionFactory(cf, connectionPool, null
- , conf.get(LensConfConstants.SERVER_DB_VALIDATION_QUERY,
- LensConfConstants.DEFAULT_SERVER_DB_VALIDATION_QUERY), false, false)
- .setDefaultAutoCommit(true);
+ new PoolableConnectionFactory(cf, connectionPool, null,
+ conf.get(LensConfConstants.SERVER_DB_VALIDATION_QUERY, LensConfConstants.DEFAULT_SERVER_DB_VALIDATION_QUERY),
+ false, false).setDefaultAutoCommit(true);
return new PoolingDataSource(connectionPool);
}
@@ -204,13 +205,31 @@ public final class UtilityMethods {
public static byte[] generateHashOfWritable(Writable writable) {
try {
MessageDigest md = MessageDigest.getInstance("MD5");
- byte [] lensConfBytes = WritableUtils.toByteArray(writable);
+ byte[] lensConfBytes = WritableUtils.toByteArray(writable);
md.update(lensConfBytes);
- byte [] digest = md.digest();
+ byte[] digest = md.digest();
return digest;
} catch (NoSuchAlgorithmException e) {
log.warn("MD5: No such method error " + writable);
return null;
}
}
+
+ /**
+ * @param conf
+ * @return
+ */
+ public static BasicDataSource getDataSourceFromConfForScheduler(Configuration conf) {
+ BasicDataSource basicDataSource = getDataSourceFromConf(conf);
+ basicDataSource.setDefaultAutoCommit(true);
+ return basicDataSource;
+ }
+
+ public static SchedulerJobHandle generateSchedulerJobHandle() {
+ return new SchedulerJobHandle(UUID.randomUUID());
+ }
+
+ public static SchedulerJobInstanceHandle generateSchedulerJobInstanceHandle() {
+ return new SchedulerJobInstanceHandle(UUID.randomUUID());
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/1a96948e/lens-server/src/main/resources/lensserver-default.xml
----------------------------------------------------------------------
diff --git a/lens-server/src/main/resources/lensserver-default.xml b/lens-server/src/main/resources/lensserver-default.xml
index a57cad6..e9525fd 100644
--- a/lens-server/src/main/resources/lensserver-default.xml
+++ b/lens-server/src/main/resources/lensserver-default.xml
@@ -864,9 +864,15 @@
</description>
</property>
<property>
- <name>lens.server.status.update.exponential.wait.millis</name>
- <value>30000</value>
- <description>Number of millis that would grow exponentially for next update, incase of transient failures.
- </description>
-</property>
+ <name>lens.server.status.update.exponential.wait.millis</name>
+ <value>30000</value>
+ <description>Number of millis that would grow exponentially for next update, incase of transient failures.
+ </description>
+ </property>
+ <property>
+ <name>lens.query.current.time.millis</name>
+ <value>0</value>
+ <description>Query current time in millis. This is used to resolve 'now'. If value is set to zero, 'now' is
+ resolved to current value</description>
+ </property>
</configuration>