You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lens.apache.org by am...@apache.org on 2016/11/02 08:28:49 UTC
lens git commit: LENS-1327 : Handle data delay for scheduled jobs
Repository: lens
Updated Branches:
refs/heads/master d09e5e2c8 -> 95a4f596c
LENS-1327 : Handle data delay for scheduled jobs
Project: http://git-wip-us.apache.org/repos/asf/lens/repo
Commit: http://git-wip-us.apache.org/repos/asf/lens/commit/95a4f596
Tree: http://git-wip-us.apache.org/repos/asf/lens/tree/95a4f596
Diff: http://git-wip-us.apache.org/repos/asf/lens/diff/95a4f596
Branch: refs/heads/master
Commit: 95a4f596c42f57647b2eb12c0419175c9c87de8e
Parents: d09e5e2
Author: Lavkesh Lahngir <la...@linux.com>
Authored: Wed Nov 2 13:58:36 2016 +0530
Committer: Amareshwari Sriramadasu <am...@apache.org>
Committed: Wed Nov 2 13:58:36 2016 +0530
----------------------------------------------------------------------
.../scheduler/SchedulerJobInstanceEvent.java | 3 +-
.../scheduler/SchedulerJobInstanceState.java | 39 ++++++---
.../lens/server/api/LensConfConstants.java | 15 +++-
.../lens/server/scheduler/SchedulerDAO.java | 5 +-
.../scheduler/SchedulerEventListener.java | 46 +++++++---
.../server/scheduler/SchedulerServiceImpl.java | 77 +++++++++++++++--
.../src/main/resources/lensserver-default.xml | 10 ++-
.../lens/server/scheduler/SchedulerDAOTest.java | 2 +-
.../server/scheduler/SchedulerRestartTest.java | 12 +++
src/site/apt/admin/config.apt | 88 ++++++++++----------
10 files changed, 217 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/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
index 4f3409c..1bcbeca 100644
--- 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
@@ -22,9 +22,10 @@ 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_PREPARE, // an instance is first considered by the scheduler.
ON_TIME_OUT,
ON_CONDITIONS_MET,
+ ON_CONDITIONS_NOT_MET,
ON_RUN,
ON_SUCCESS,
ON_FAILURE,
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/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
index 3d2605e..52b3aec 100644
--- 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
@@ -18,7 +18,7 @@
*/
package org.apache.lens.api.scheduler;
-import javax.xml.bind.annotation.*;
+import javax.xml.bind.annotation.XmlRootElement;
import org.apache.lens.api.error.InvalidStateTransitionException;
@@ -31,10 +31,8 @@ public enum SchedulerJobInstanceState
public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
throws InvalidStateTransitionException {
switch (event) {
- case ON_CREATION:
- return this;
- case ON_CONDITIONS_MET:
- return SchedulerJobInstanceState.LAUNCHED;
+ case ON_PREPARE:
+ return SchedulerJobInstanceState.LAUNCHING;
case ON_TIME_OUT:
return SchedulerJobInstanceState.TIMED_OUT;
case ON_RUN:
@@ -52,13 +50,15 @@ public enum SchedulerJobInstanceState
}
},
- LAUNCHED {
+ LAUNCHING {
@Override
public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
throws InvalidStateTransitionException {
switch (event) {
case ON_CONDITIONS_MET:
- return this;
+ return SchedulerJobInstanceState.LAUNCHED;
+ case ON_CONDITIONS_NOT_MET:
+ return SchedulerJobInstanceState.WAITING;
case ON_RUN:
return SchedulerJobInstanceState.RUNNING;
case ON_SUCCESS:
@@ -74,6 +74,25 @@ public enum SchedulerJobInstanceState
}
},
+ LAUNCHED {
+ @Override
+ public SchedulerJobInstanceState nextTransition(SchedulerJobInstanceEvent event)
+ throws InvalidStateTransitionException {
+ switch (event) {
+ 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)
@@ -102,7 +121,7 @@ public enum SchedulerJobInstanceState
case ON_FAILURE:
return this;
case ON_RERUN:
- return SchedulerJobInstanceState.LAUNCHED;
+ return SchedulerJobInstanceState.LAUNCHING;
default:
throw new InvalidStateTransitionException(
"SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
@@ -118,7 +137,7 @@ public enum SchedulerJobInstanceState
case ON_SUCCESS:
return this;
case ON_RERUN:
- return SchedulerJobInstanceState.LAUNCHED;
+ return SchedulerJobInstanceState.LAUNCHING;
default:
throw new InvalidStateTransitionException(
"SchedulerJobInstanceEvent: " + event.name() + " is not a valid event for state: " + this.name());
@@ -150,7 +169,7 @@ public enum SchedulerJobInstanceState
case ON_KILL:
return this;
case ON_RERUN:
- return SchedulerJobInstanceState.LAUNCHED;
+ return SchedulerJobInstanceState.LAUNCHING;
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/95a4f596/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 3ae59c6..8f9db2a 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
@@ -1185,14 +1185,23 @@ public final class LensConfConstants {
public static final int DEFAULT_GRIZZLY_MAX_POOL_SIZE = 40;
/**
- * Maximum Scheduled job per user.
+ * Thread interval for checking the waiting instances
*/
- public static final String MAX_SCHEDULED_JOB_PER_USER = SERVER_PFX + "scheduler.max.job.per.user";
+ public static final String SCHEDULED_INSTANCE_WAITING_THREAD_INTERVAL_MILLIS =
+ SERVER_PFX + "scheduler.instance.waiting.thread.interval";
/**
- * -1 represents that the default is unlimited
+ * Default waiting thread interval in milliseconds
*/
+ public static final long DEFAULT_SCHEDULED_INSTANCE_WAITING_THREAD_INTERVAL_MILLIS = 60 * 5 * 1000;
+ /**
+ * Default value is less than zero, that means an user can scheduler unlimited number of jobs.
+ */
public static final int DEFAULT_MAX_SCHEDULED_JOB_PER_USER = -1;
+ /**
+ * Maximum number of scheduled job per user.
+ */
+ public static final String MAX_SCHEDULED_JOB_PER_USER = SERVER_PFX + "scheduler.max.job.per.user";
}
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/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 5f472fc..28b1436 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
@@ -573,8 +573,9 @@ public class SchedulerDAO {
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 + "=?";
+ // Get the Runs sorted by run id to make sure the last on the list is the latest run.
+ String fetchSQL =
+ "SELECT * FROM " + JOB_INSTANCE_RUN_TABLE + " WHERE " + COLUMN_ID + "=? ORDER BY " + COLUMN_RUN_ID;
List<Object[]> instanceRuns = runner.query(fetchSQL, multipleRowsHandler, (String) instanceInfo[0]);
List<SchedulerJobInstanceRun> runList = processInstanceRun(instanceRuns);
return new SchedulerJobInstanceInfo(id, jobId, createdOn, runList);
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/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
index 1db8136..90f90f4 100644
--- 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
@@ -26,6 +26,7 @@ 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.error.LensCubeErrorCode;
import org.apache.lens.server.LensServices;
import org.apache.lens.server.api.LensConfConstants;
import org.apache.lens.server.api.error.LensException;
@@ -141,22 +142,45 @@ public class SchedulerEventListener extends AsyncEventListener<SchedulerAlarmEve
} else {
instance = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle);
}
- // Next run of the instance
- long currentTime = System.currentTimeMillis();
- run = new SchedulerJobInstanceRun(instanceHandle, instance.getInstanceRunList().size() + 1, null, currentTime, 0,
- "N/A", null, SchedulerJobInstanceState.WAITING);
- instance.getInstanceRunList().add(run);
- if (schedulerDAO.storeJobInstanceRun(run) != 1) {
- log.error("Exception occurred while storing the instance run for instance handle {} of job {}", instance,
- jobHandle);
- return;
- }
- run.setSessionHandle(sessionHandle);
LensConf queryConf = getLensConf(job, instanceHandle, scheduledTime);
// Query Launch
String query = job.getExecution().getQuery().getQuery();
String queryName = job.getName();
queryName += "-" + scheduledTimeMillis;
+ // Fetch the latest run and if it is in waiting state then don't create a new run.
+ run = instance.getInstanceRunList().size() == 0
+ ? null
+ : instance.getInstanceRunList().get(instance.getInstanceRunList().size() - 1);
+ if (run == null || run.getInstanceState() != SchedulerJobInstanceState.LAUNCHING) {
+ // Next run of the instance
+ // If not true means run is in waiting state, so we don't need to create a new run.
+ long currentTime = System.currentTimeMillis();
+ run = new SchedulerJobInstanceRun(instanceHandle, instance.getInstanceRunList().size() + 1, null, currentTime,
+ currentTime, "N/A", null, SchedulerJobInstanceState.LAUNCHING);
+ instance.getInstanceRunList().add(run);
+ if (schedulerDAO.storeJobInstanceRun(run) != 1) {
+ log.error("Exception occurred while storing the instance run for instance handle {} of job {}", instance,
+ jobHandle);
+ return;
+ }
+ }
+ run.setSessionHandle(sessionHandle);
+ // Check for the data availability.
+ try {
+ queryService.estimate(LensServices.get().getLogSegregationContext().getLogSegragationId(), sessionHandle, query,
+ queryConf);
+ } catch (LensException e) {
+ if (e.getErrorCode() == LensCubeErrorCode.NO_CANDIDATE_FACT_AVAILABLE.getLensErrorInfo().getErrorCode()) {
+ // This error code suggests that the data is not available.
+ run.setInstanceState(run.getInstanceState().nextTransition(SchedulerJobInstanceEvent.ON_CONDITIONS_NOT_MET));
+ run.setEndTime(System.currentTimeMillis());
+ schedulerDAO.updateJobInstanceRun(run);
+ return;
+ } else {
+ throw e;
+ }
+ }
+
QueryHandle handle = queryService.executeAsync(sessionHandle, query, queryConf, queryName);
log.info("Running instance {} of job {} with run {} with query handle {}", instanceHandle, jobHandle,
run.getRunId(), handle);
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/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 73277de..4d3b8e5 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
@@ -18,8 +18,7 @@
*/
package org.apache.lens.server.scheduler;
-import static org.apache.lens.api.scheduler.SchedulerJobInstanceEvent.ON_KILL;
-import static org.apache.lens.api.scheduler.SchedulerJobInstanceEvent.ON_RERUN;
+import static org.apache.lens.api.scheduler.SchedulerJobInstanceEvent.*;
import java.util.Collection;
import java.util.HashMap;
@@ -39,6 +38,7 @@ import org.apache.lens.server.LensServices;
import org.apache.lens.server.api.LensConfConstants;
import org.apache.lens.server.api.LensErrorInfo;
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.lens.server.api.query.QueryExecutionService;
@@ -65,6 +65,9 @@ import lombok.extern.slf4j.Slf4j;
public class SchedulerServiceImpl extends BaseLensService implements SchedulerService {
@Getter
+ @VisibleForTesting
+ protected SchedulerDAO schedulerDAO;
+ @Getter
@Setter
@VisibleForTesting
protected QueryExecutionService queryService;
@@ -75,10 +78,9 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
@VisibleForTesting
protected SchedulerQueryEventListener schedulerQueryEventListener;
@Getter
- @VisibleForTesting
- protected SchedulerDAO schedulerDAO;
- @Getter
private AlarmService alarmService;
+ private Thread waitingInstanceThread;
+ private long waitingThreadInterval = LensConfConstants.DEFAULT_SCHEDULED_INSTANCE_WAITING_THREAD_INTERVAL_MILLIS;
private int maxJobsPerUser = LensConfConstants.DEFAULT_MAX_SCHEDULED_JOB_PER_USER;
private boolean healthy = true;
@@ -98,12 +100,16 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
maxJobsPerUser = hiveConf.getInt(LensConfConstants.MAX_SCHEDULED_JOB_PER_USER, maxJobsPerUser);
try {
schedulerDAO = new SchedulerDAO(hiveConf);
+ waitingThreadInterval = hiveConf.getLong(LensConfConstants.SCHEDULED_INSTANCE_WAITING_THREAD_INTERVAL_MILLIS,
+ LensConfConstants.DEFAULT_SCHEDULED_INSTANCE_WAITING_THREAD_INTERVAL_MILLIS);
alarmService = LensServices.get().getService(AlarmService.NAME);
queryService = LensServices.get().getService(QueryExecutionService.NAME);
this.schedulerEventListener = new SchedulerEventListener(schedulerDAO);
this.schedulerQueryEventListener = new SchedulerQueryEventListener(schedulerDAO);
getEventService().addListenerForType(schedulerEventListener, SchedulerAlarmEvent.class);
getEventService().addListenerForType(schedulerQueryEventListener, QueryEnded.class);
+ this.waitingInstanceThread = new Thread(
+ new WaitingInstanceHandler(getEventService(), schedulerDAO, waitingThreadInterval));
} catch (LensException e) {
log.error("Error Initialising Scheduler-service", e);
healthy = false;
@@ -129,15 +135,20 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
@Override
public synchronized void start() {
List<SchedulerJobInstanceRun> instanceRuns = schedulerDAO
- .getInstanceRuns(SchedulerJobInstanceState.WAITING, SchedulerJobInstanceState.LAUNCHED,
- SchedulerJobInstanceState.RUNNING);
+ .getInstanceRuns(SchedulerJobInstanceState.WAITING, SchedulerJobInstanceState.LAUNCHING,
+ SchedulerJobInstanceState.LAUNCHED, SchedulerJobInstanceState.RUNNING);
for (SchedulerJobInstanceRun run : instanceRuns) {
LensSessionHandle sessionHandle = null;
try {
SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(run.getHandle());
+ if (instanceInfo == null) {
+ log.error("Instance info was not found for {}", run.getHandle());
+ continue;
+ }
log.info("Recovering instance {} of job {} ", instanceInfo.getId(), instanceInfo.getJobId());
switch (run.getInstanceState()) {
case WAITING:
+ case LAUNCHING:
case LAUNCHED:
// Kill and rerun
if (updateInstanceRun(run, SchedulerJobInstanceState.KILLED)) {
@@ -168,11 +179,22 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
}
}
if (healthy) {
+ // Start thread
+ this.waitingInstanceThread.start();
super.start();
}
}
/**
+ *
+ */
+ @Override
+ public void stop() {
+ super.stop();
+ waitingInstanceThread.interrupt();
+ }
+
+ /**
* If query is not found of is invalid then rerun again else get the status and update correspondingly.
*
* @param sessionHandle
@@ -546,4 +568,45 @@ public class SchedulerServiceImpl extends BaseLensService implements SchedulerSe
new SchedulerAlarmEvent(instanceInfo.getJobId(), new DateTime(instanceInfo.getScheduleTime()),
SchedulerAlarmEvent.EventType.SCHEDULE, instanceInfo.getId()));
}
+
+ public static class WaitingInstanceHandler implements Runnable {
+ private LensEventService eventService;
+ private SchedulerDAO schedulerDAO;
+ private long delay;
+
+ WaitingInstanceHandler(LensEventService eventService, SchedulerDAO schedulerDAO, long delay) {
+ this.eventService = eventService;
+ this.schedulerDAO = schedulerDAO;
+ this.delay = delay;
+ }
+
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ // Get all the instance runs which are waiting.
+ List<SchedulerJobInstanceRun> instanceRuns = schedulerDAO.getInstanceRuns(SchedulerJobInstanceState.WAITING);
+ for (SchedulerJobInstanceRun run : instanceRuns) {
+ SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(run.getHandle());
+ try {
+ run.setInstanceState(run.getInstanceState().nextTransition(ON_PREPARE));
+ run.setEndTime(System.currentTimeMillis());
+ schedulerDAO.updateJobInstanceRun(run);
+ eventService.notifyEvent(
+ new SchedulerAlarmEvent(instanceInfo.getJobId(), new DateTime(instanceInfo.getScheduleTime()),
+ SchedulerAlarmEvent.EventType.SCHEDULE, instanceInfo.getId()));
+ } catch (LensException e) {
+ log.error("Not able to notify schedule event for job {} and instanceId {}", instanceInfo.getJobId(),
+ instanceInfo.getId());
+ } catch (InvalidStateTransitionException e) {
+ log.error("Wrong state transition", e);
+ }
+ }
+ }
+ try {
+ Thread.sleep(delay);
+ } catch (InterruptedException e) {
+ log.warn("Thread WaitingInstanceHandler was inerrupted", e);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/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 8e5fe8a..a00048b 100644
--- a/lens-server/src/main/resources/lensserver-default.xml
+++ b/lens-server/src/main/resources/lensserver-default.xml
@@ -917,10 +917,16 @@
<value>40</value>
<description>Max pool size for lens grizzly server</description>
</property>
-
+ <property>
+ <name>lens.server.scheduler.instance.waiting.thread.interval.millis</name>
+ <value>300000</value>
+ <description>Thread interval for checking the waiting instances in milliseconds</description>
+ </property>
<property>
<name>lens.server.scheduler.max.job.per.user</name>
<value>-1</value>
- <description>The maximum number of jobs per user which can be scheduled. -1 means unlimited jobs.</description>
+ <description>Maximum number of jobs that can be scheduled by a single user. If the number is less than zero,
+ then there is no restriction on the number of jobs scheduled.
+ </description>
</property>
</configuration>
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/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 e698e2b..4754c1e 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
@@ -172,7 +172,7 @@ public class SchedulerDAOTest {
SchedulerJobInstanceHandle handle = instances.keySet().iterator().next();
SchedulerJobInstanceInfo info = instances.get(handle);
SchedulerJobInstanceRun run = info.getInstanceRunList().get(0);
- run.setInstanceState(SchedulerJobInstanceState.LAUNCHED);
+ run.setInstanceState(SchedulerJobInstanceState.LAUNCHING);
schedulerDAO.updateJobInstanceRun(run);
// Get the instance
Assert.assertEquals(schedulerDAO.getSchedulerJobInstanceInfo(handle), info);
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerRestartTest.java
----------------------------------------------------------------------
diff --git a/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerRestartTest.java b/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerRestartTest.java
index 03f5d9b..b45c28a 100644
--- a/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerRestartTest.java
+++ b/lens-server/src/test/java/org/apache/lens/server/scheduler/SchedulerRestartTest.java
@@ -22,18 +22,22 @@ import static org.apache.lens.server.scheduler.util.SchedulerTestUtils.getTestJo
import static org.apache.lens.server.scheduler.util.SchedulerTestUtils.setupQueryService;
import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
import java.util.ArrayList;
import java.util.List;
+import org.apache.lens.api.LensConf;
import org.apache.lens.api.LensSessionHandle;
import org.apache.lens.api.query.LensQuery;
import org.apache.lens.api.query.QueryHandle;
import org.apache.lens.api.query.QueryStatus;
import org.apache.lens.api.scheduler.*;
+import org.apache.lens.cube.error.LensCubeErrorCode;
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.metrics.LensMetricsUtil;
import org.apache.lens.server.api.scheduler.SchedulerService;
import org.apache.lens.server.model.LogSegregationContext;
@@ -78,6 +82,10 @@ public class SchedulerRestartTest {
List<SchedulerJobInstanceInfo> instanceInfoList = scheduler.getJobInstances(jobHandle, 10L);
Assert.assertEquals(instanceInfoList.size(), 1);
+ PowerMockito.when(
+ scheduler.getQueryService().estimate(anyString(), any(LensSessionHandle.class), anyString(), any(LensConf.class)))
+ .thenThrow(new LensException(LensCubeErrorCode.NO_CANDIDATE_FACT_AVAILABLE.getLensErrorInfo(),
+ "some name " + " does not have any facts"));
// Store new instance
SchedulerJobInstanceHandle instanceHandle = UtilityMethods.generateSchedulerJobInstanceHandle();
SchedulerJobInstanceInfo instance = new SchedulerJobInstanceInfo(instanceHandle, jobHandle, currentTime,
@@ -98,6 +106,10 @@ public class SchedulerRestartTest {
LensServices.get().init(LensServerConf.getHiveConf());
scheduler = LensServices.get().getService(SchedulerService.NAME);
setupQueryService(scheduler);
+ PowerMockito.when(
+ scheduler.getQueryService().estimate(anyString(), any(LensSessionHandle.class), anyString(), any(LensConf.class)))
+ .thenThrow(new LensException(LensCubeErrorCode.NO_CANDIDATE_FACT_AVAILABLE.getLensErrorInfo(),
+ "some name " + " does not have any facts"));
LensQuery mockedQuery = PowerMockito.mock(LensQuery.class);
QueryStatus mockStatus = PowerMockito.mock(QueryStatus.class);
PowerMockito.when(mockStatus.getStatus()).thenReturn(QueryStatus.Status.SUCCESSFUL);
http://git-wip-us.apache.org/repos/asf/lens/blob/95a4f596/src/site/apt/admin/config.apt
----------------------------------------------------------------------
diff --git a/src/site/apt/admin/config.apt b/src/site/apt/admin/config.apt
index f22fc0c..4d49781 100644
--- a/src/site/apt/admin/config.apt
+++ b/src/site/apt/admin/config.apt
@@ -195,90 +195,92 @@ Lens server configuration
*--+--+---+--+
|83|lens.server.savedquery.ws.resource.impl|org.apache.lens.server.query.save.SavedQueryResource|Implementation class for Saved query Resource|
*--+--+---+--+
-|84|lens.server.scheduler.max.job.per.user|-1|The maximum number of jobs per user which can be scheduled. -1 means unlimited jobs.|
+|84|lens.server.scheduler.instance.waiting.thread.interval.millis|300000|Thread interval for checking the waiting instances in milliseconds|
*--+--+---+--+
-|85|lens.server.scheduler.service.impl|org.apache.lens.server.scheduler.SchedulerServiceImpl|Implementation class for query scheduler service|
+|85|lens.server.scheduler.max.job.per.user|-1|Maximum number of jobs that can be scheduled by a single user. If the number is less than zero, then there is no restriction on the number of jobs scheduled.|
*--+--+---+--+
-|86|lens.server.scheduler.store.class|org.apache.lens.server.scheduler.SchedulerDAO$SchedulerHsqlDBStore|A subclass of SchedulerDBStore class used for storing scheduler related information.|
+|86|lens.server.scheduler.service.impl|org.apache.lens.server.scheduler.SchedulerServiceImpl|Implementation class for query scheduler service|
*--+--+---+--+
-|87|lens.server.scheduler.ws.resource.impl|org.apache.lens.server.scheduler.ScheduleResource|Implementation class for query scheduler resource|
+|87|lens.server.scheduler.store.class|org.apache.lens.server.scheduler.SchedulerDAO$SchedulerHsqlDBStore|A subclass of SchedulerDBStore class used for storing scheduler related information.|
*--+--+---+--+
-|88|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.|
+|88|lens.server.scheduler.ws.resource.impl|org.apache.lens.server.scheduler.ScheduleResource|Implementation class for query scheduler resource|
*--+--+---+--+
-|89|lens.server.serverMode.ws.filter.impl|org.apache.lens.server.ServerModeFilter|Implementation class for ServerMode Filter|
+|89|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.|
*--+--+---+--+
-|90|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.|
+|90|lens.server.serverMode.ws.filter.impl|org.apache.lens.server.ServerModeFilter|Implementation class for ServerMode Filter|
*--+--+---+--+
-|91|lens.server.servicenames|session,alarm,query,savedquery,metastore,scheduler,quota|These services would be started in the specified order when lens-server starts up|
+|91|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.|
*--+--+---+--+
-|92|lens.server.session.expiry.service.interval.secs|3600|Interval at which lens session expiry service runs|
+|92|lens.server.servicenames|session,alarm,query,savedquery,metastore,scheduler,quota|These services would be started in the specified order when lens-server starts up|
*--+--+---+--+
-|93|lens.server.session.service.impl|org.apache.lens.server.session.HiveSessionService|Implementation class for session service|
+|93|lens.server.session.expiry.service.interval.secs|3600|Interval at which lens session expiry service runs|
*--+--+---+--+
-|94|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.|
+|94|lens.server.session.service.impl|org.apache.lens.server.session.HiveSessionService|Implementation class for session service|
*--+--+---+--+
-|95|lens.server.session.ws.resource.impl|org.apache.lens.server.session.SessionResource|Implementation class for Session Resource|
+|95|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.|
*--+--+---+--+
-|96|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.|
+|96|lens.server.session.ws.resource.impl|org.apache.lens.server.session.SessionResource|Implementation class for Session Resource|
*--+--+---+--+
-|97|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.|
+|97|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.|
*--+--+---+--+
-|98|lens.server.state.persistence.interval.millis|300000|Lens server state persistence time interval in milliseconds|
+|98|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.|
*--+--+---+--+
-|99|lens.server.statistics.db|lensstats|Database to which statistics tables are created and partitions are added.|
+|99|lens.server.state.persistence.interval.millis|300000|Lens server state persistence time interval in milliseconds|
*--+--+---+--+
-|100|lens.server.statistics.log.rollover.interval|3600000|Default rate which log statistics store scans for rollups in milliseconds.|
+|100|lens.server.statistics.db|lensstats|Database to which statistics tables are created and partitions are added.|
*--+--+---+--+
-|101|lens.server.statistics.store.class|org.apache.lens.server.stats.store.log.LogStatisticsStore|Default implementation of class used to persist Lens Statistics.|
+|101|lens.server.statistics.log.rollover.interval|3600000|Default rate which log statistics store scans for rollups in milliseconds.|
*--+--+---+--+
-|102|lens.server.statistics.warehouse.dir|file:///tmp/lens/statistics/warehouse|Default top level location where stats are moved by the log statistics store.|
+|102|lens.server.statistics.store.class|org.apache.lens.server.stats.store.log.LogStatisticsStore|Default implementation of class used to persist Lens Statistics.|
*--+--+---+--+
-|103|lens.server.status.update.exponential.wait.millis|30000|Number of millis that would grow exponentially for next update, incase of transient failures.|
+|103|lens.server.statistics.warehouse.dir|file:///tmp/lens/statistics/warehouse|Default top level location where stats are moved by the log statistics store.|
*--+--+---+--+
-|104|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.|
+|104|lens.server.status.update.exponential.wait.millis|30000|Number of millis that would grow exponentially for next update, incase of transient failures.|
*--+--+---+--+
-|105|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.|
+|105|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.|
*--+--+---+--+
-|106|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.|
+|106|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.|
*--+--+---+--+
-|107|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|
+|107|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.|
*--+--+---+--+
-|108|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.|
+|108|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|
*--+--+---+--+
-|109|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.|
+|109|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.|
*--+--+---+--+
-|110|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.|
+|110|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.|
*--+--+---+--+
-|111|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...|
+|111|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.|
*--+--+---+--+
-|112|lens.server.user.resolver.ldap.bind.password| |Required for LDAP_BACKED_DATABASE user resolvers. ldap password for admin binding above|
+|112|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...|
*--+--+---+--+
-|113|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.|
+|113|lens.server.user.resolver.ldap.bind.password| |Required for LDAP_BACKED_DATABASE user resolvers. ldap password for admin binding above|
*--+--+---+--+
-|114|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.|
+|114|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.|
*--+--+---+--+
-|115|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|
+|115|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.|
*--+--+---+--+
-|116|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.|
+|116|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|
*--+--+---+--+
-|117|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...|
+|117|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.|
*--+--+---+--+
-|118|lens.server.user.resolver.ldap.search.filter|(&(objectClass=user)(sAMAccountName=%s))|Required for LDAP_BACKED_DATABASE user resolvers. filter pattern for ldap search|
+|118|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...|
*--+--+---+--+
-|119|lens.server.user.resolver.ldap.url| |Required for LDAP_BACKED_DATABASE user resolvers. ldap url to connect to.|
+|119|lens.server.user.resolver.ldap.search.filter|(&(objectClass=user)(sAMAccountName=%s))|Required for LDAP_BACKED_DATABASE user resolvers. filter pattern for ldap search|
*--+--+---+--+
-|120|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|
+|120|lens.server.user.resolver.ldap.url| |Required for LDAP_BACKED_DATABASE user resolvers. ldap url to connect to.|
*--+--+---+--+
-|121|lens.server.user.resolver.type|FIXED|Type of user config resolver. allowed values are FIXED, PROPERTYBASED, DATABASE, LDAP_BACKED_DATABASE, CUSTOM.|
+|121|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|
*--+--+---+--+
-|122|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.|
+|122|lens.server.user.resolver.type|FIXED|Type of user config resolver. allowed values are FIXED, PROPERTYBASED, DATABASE, LDAP_BACKED_DATABASE, CUSTOM.|
*--+--+---+--+
-|123|lens.server.ws.featurenames|multipart,moxyjson,moxyjsonconfigresovler|These JAX-RS Feature(s) would be started in the specified order when lens-server starts up|
+|123|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.|
*--+--+---+--+
-|124|lens.server.ws.filternames|requestlogger,consistentState,serverMode|These JAX-RS filters would be started in the specified order when lens-server starts up|
+|124|lens.server.ws.featurenames|multipart,moxyjson,moxyjsonconfigresovler|These JAX-RS Feature(s) would be started in the specified order when lens-server starts up|
*--+--+---+--+
-|125|lens.server.ws.listenernames|appevent|These listeners would be called in the specified order when lens-server starts up|
+|125|lens.server.ws.filternames|requestlogger,consistentState,serverMode|These JAX-RS filters would be started in the specified order when lens-server starts up|
*--+--+---+--+
-|126|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|
+|126|lens.server.ws.listenernames|appevent|These listeners would be called in the specified order when lens-server starts up|
+*--+--+---+--+
+|127|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