Author: kamrul
Date: Tue Oct 23 17:46:51 2012
New Revision: 1401361
URL: http://svn.apache.org/viewvc?rev=1401361&view=rev
Log:
OOZIE-1020 BulkJPAExecutor handling date-time value incorrectly.(Mona via
Mohammad)
Modified:
oozie/trunk/core/src/main/java/org/apache/oozie/BundleJobBean.java
oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java
oozie/trunk/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java
oozie/trunk/docs/src/site/twiki/WebServicesAPI.twiki
oozie/trunk/release-log.txt
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/BundleJobBean.java
URL:
http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/BundleJobBean.java?rev=1401361&r1=1401360&r2=1401361&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/BundleJobBean.java
(original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/BundleJobBean.java Tue Oct
23 17:46:51 2012
@@ -66,7 +66,17 @@ import org.apache.openjpa.persistence.jd
@NamedQuery(name = "GET_BUNDLE_JOBS_OLDER_THAN_STATUS", query =
"select OBJECT(w) from BundleJobBean w where w.status = :status AND
w.lastModifiedTimestamp <= :lastModTime order by w.lastModifiedTimestamp"),
- @NamedQuery(name = "GET_COMPLETED_BUNDLE_JOBS_OLDER_THAN", query =
"select OBJECT(w) from BundleJobBean w where ( w.status = 'SUCCEEDED' OR
w.status = 'FAILED' OR w.status = 'KILLED' OR w.status = 'DONEWITHERROR') AND
w.lastModifiedTimestamp <= :lastModTime order by w.lastModifiedTimestamp")})
+ @NamedQuery(name = "GET_COMPLETED_BUNDLE_JOBS_OLDER_THAN", query =
"select OBJECT(w) from BundleJobBean w where ( w.status = 'SUCCEEDED' OR
w.status = 'FAILED' OR w.status = 'KILLED' OR w.status = 'DONEWITHERROR') AND
w.lastModifiedTimestamp <= :lastModTime order by w.lastModifiedTimestamp"),
+
+ @NamedQuery(name = "BULK_MONITOR_BUNDLE_QUERY", query = "SELECT b.id,
b.status FROM BundleJobBean b WHERE b.appName = :appName"),
+
+ // Join query
+ @NamedQuery(name = "BULK_MONITOR_ACTIONS_QUERY", query = "SELECT a.id,
a.actionNumber, a.errorCode, a.errorMessage, a.externalId, " +
+ "a.externalStatus, a.status, a.createdTimestamp,
a.nominalTimestamp, a.missingDependencies, " +
+ "c.id, c.appName, c.status FROM CoordinatorActionBean a,
CoordinatorJobBean c " +
+ "WHERE a.jobId = c.id AND c.bundleId = :bundleId ORDER BY
a.jobId, a.createdTimestamp"),
+
+ @NamedQuery(name = "BULK_MONITOR_COUNT_QUERY", query = "SELECT
COUNT(a) FROM CoordinatorActionBean a, CoordinatorJobBean c") })
public class BundleJobBean extends JsonBundleJob implements Writable {
@Basic
Modified:
oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java
URL:
http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java?rev=1401361&r1=1401360&r2=1401361&view=diff
==============================================================================
---
oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java
(original)
+++
oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java
Tue Oct 23 17:46:51 2012
@@ -21,8 +21,12 @@ import java.sql.Timestamp;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
+
import javax.persistence.EntityManager;
import javax.persistence.Query;
@@ -39,7 +43,8 @@ import org.apache.oozie.util.DateUtils;
import org.apache.oozie.util.ParamChecker;
/**
- * Load the coordinators for specified bundle in the Coordinator job bean
+ * The query executor class for bulk monitoring queries i.e. debugging bundle
->
+ * coord actions directly
*/
public class BulkJPAExecutor implements JPAExecutor<BulkResponseInfo> {
private Map<String, List<String>> bulkFilter;
@@ -47,14 +52,6 @@ public class BulkJPAExecutor implements
private int start = 1;
private int len = 50;
- public static final String bundleQuery = "SELECT b.id, b.status FROM
BundleJobBean b WHERE b.appName = :appName";
- // Join query
- public static final String actionQuery = "SELECT a.id, a.actionNumber,
a.errorCode, a.errorMessage, a.externalId, " +
- "a.externalStatus, a.status, a.createdTimestamp,
a.nominalTimestamp, a.missingDependencies, " +
- "c.id, c.appName, c.status FROM CoordinatorActionBean a,
CoordinatorJobBean c " +
- "WHERE a.jobId = c.id AND c.bundleId = :bundleId ORDER BY a.jobId,
a.createdTimestamp";
- public static final String countQuery = "SELECT COUNT(a) FROM
CoordinatorActionBean a, CoordinatorJobBean c ";
-
public BulkJPAExecutor(Map<String, List<String>> bulkFilter, int start,
int len) {
ParamChecker.notNull(bulkFilter, "bulkFilter");
this.bulkFilter = bulkFilter;
@@ -62,7 +59,8 @@ public class BulkJPAExecutor implements
this.len = len;
}
- /* (non-Javadoc)
+ /*
+ * (non-Javadoc)
* @see org.apache.oozie.executor.jpa.JPAExecutor#getName()
*/
@Override
@@ -70,59 +68,28 @@ public class BulkJPAExecutor implements
return "BulkJPAExecutor";
}
- /* (non-Javadoc)
+ /*
+ * (non-Javadoc)
* @see
org.apache.oozie.executor.jpa.JPAExecutor#execute(javax.persistence.EntityManager)
*/
@Override
- @SuppressWarnings("unchecked")
public BulkResponseInfo execute(EntityManager em) throws
JPAExecutorException {
- BundleJobBean bundleBean = new BundleJobBean();
-
List<BulkResponseImpl> responseList = new
ArrayList<BulkResponseImpl>();
+ Map<String, Timestamp> actionTimes = new HashMap<String, Timestamp>();
try {
+ // Lightweight Query 1 on Bundle level to fetch the bundle job
+ // corresponding to name
+ BundleJobBean bundleBean = bundleQuery(em);
+
+ // Join query between coordinator job and coordinator action tables
+ // to get entries for specific bundleId only
+ String conditions = actionQuery(em, bundleBean, actionTimes,
responseList);
- // Lightweight Query 1 on Bundle level to fetch the bundle job
params corresponding to name
- String bundleName =
bulkFilter.get(BulkResponseImpl.BULK_FILTER_BUNDLE_NAME).get(0);
- Query q = em.createQuery(bundleQuery);
- q.setParameter("appName", bundleName);
- List<Object[]> bundles = (List<Object[]>) q.getResultList();
- if(bundles.isEmpty()) {
- throw new JPAExecutorException(ErrorCode.E0603, "No bundle
entries found for bundle name: " + bundleName);
- }
- if(bundles.size() > 1) { // more than one bundles running with
same name - ERROR. Fail fast
- throw new JPAExecutorException(ErrorCode.E0603, "Non-unique
bundles present for same bundle name: " + bundleName);
- }
- bundleBean = getBeanForBundleJob(bundles.get(0), bundleName);
-
- //Join query between coordinator job and coordinator action tables
to get entries for specific bundleId only
- StringBuilder conditionClause = new StringBuilder();
-
conditionClause.append(coordNamesClause(bulkFilter.get(BulkResponseImpl.BULK_FILTER_COORD_NAME)));
-
conditionClause.append(statusClause(bulkFilter.get(BulkResponseImpl.BULK_FILTER_STATUS)));
- conditionClause.append(timesClause());
- StringBuilder getActions = new StringBuilder(actionQuery);
- int offset = getActions.indexOf("ORDER");
- getActions.insert(offset-1, conditionClause);
- q = em.createQuery(getActions.toString());
- q.setParameter("bundleId", bundleBean.getId());
- // pagination
- q.setFirstResult(start - 1);
- q.setMaxResults(len);
-
- List<Object[]> response = q.getResultList();
- for (Object[] r : response) {
- BulkResponseImpl br = getResponseFromObject(bundleBean, r);
- responseList.add(br);
- }
-
- StringBuilder getTotal = new StringBuilder(countQuery);
- String query = q.toString();
- getTotal.append(query.substring(query.indexOf("WHERE"),
query.indexOf("ORDER")));
- Query qTotal = em.createQuery(getTotal.toString());
- qTotal.setParameter("bundleId", bundleBean.getId());
- long total = ((Long) qTotal.getSingleResult()).longValue();
+ // Query to get the count of records
+ long total = countQuery(conditions, em, bundleBean, actionTimes);
- BulkResponseInfo bulk = new BulkResponseInfo(responseList, start,
len , total);
+ BulkResponseInfo bulk = new BulkResponseInfo(responseList, start,
len, total);
return bulk;
}
catch (Exception e) {
@@ -130,19 +97,85 @@ public class BulkJPAExecutor implements
}
}
+ @SuppressWarnings("unchecked")
+ private BundleJobBean bundleQuery(EntityManager em) throws
JPAExecutorException {
+ BundleJobBean bundleBean = new BundleJobBean();
+ String bundleName =
bulkFilter.get(BulkResponseImpl.BULK_FILTER_BUNDLE_NAME).get(0);
+ Query q = em.createNamedQuery("BULK_MONITOR_BUNDLE_QUERY");
+ q.setParameter("appName", bundleName);
+ List<Object[]> bundles = (List<Object[]>) q.getResultList();
+ if (bundles.isEmpty()) {
+ throw new JPAExecutorException(ErrorCode.E0603, "No bundle entries
found for bundle name: "
+ + bundleName);
+ }
+ if (bundles.size() > 1) { // more than one bundles running with same
+ // name - ERROR. Fail fast
+ throw new JPAExecutorException(ErrorCode.E0603, "Non-unique
bundles present for same bundle name: "
+ + bundleName);
+ }
+ bundleBean = getBeanForBundleJob(bundles.get(0), bundleName);
+ return bundleBean;
+ }
+
+ @SuppressWarnings("unchecked")
+ private String actionQuery(EntityManager em, BundleJobBean bundleBean,
+ Map<String, Timestamp> times, List<BulkResponseImpl> responseList)
throws ParseException {
+ Query q = em.createNamedQuery("BULK_MONITOR_ACTIONS_QUERY");
+ StringBuilder getActions = new StringBuilder(q.toString());
+ StringBuilder conditionClause = new StringBuilder();
+
conditionClause.append(coordNamesClause(bulkFilter.get(BulkResponseImpl.BULK_FILTER_COORD_NAME)));
+
conditionClause.append(statusClause(bulkFilter.get(BulkResponseImpl.BULK_FILTER_STATUS)));
+ int offset = getActions.indexOf("ORDER");
+ getActions.insert(offset - 1, conditionClause);
+ timesClause(getActions, offset, times);
+ q = em.createQuery(getActions.toString());
+ Iterator<Entry<String, Timestamp>> iter = times.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<String, Timestamp> time = iter.next();
+ q.setParameter(time.getKey(), time.getValue());
+ }
+ q.setParameter("bundleId", bundleBean.getId());
+ // pagination
+ q.setFirstResult(start - 1);
+ q.setMaxResults(len);
+
+ List<Object[]> response = q.getResultList();
+ for (Object[] r : response) {
+ BulkResponseImpl br = getResponseFromObject(bundleBean, r);
+ responseList.add(br);
+ }
+ return q.toString();
+ }
+
+ private long countQuery(String clause, EntityManager em, BundleJobBean
bundleBean, Map<String, Timestamp> times) {
+ Query q = em.createNamedQuery("BULK_MONITOR_COUNT_QUERY");
+ StringBuilder getTotal = new StringBuilder(q.toString() + " ");
+ getTotal.append(clause.substring(clause.indexOf("WHERE"),
clause.indexOf("ORDER")));
+ q = em.createQuery(getTotal.toString());
+ q.setParameter("bundleId", bundleBean.getId());
+ Iterator<Entry<String, Timestamp>> iter = times.entrySet().iterator();
+ while (iter.hasNext()) {
+ Entry<String, Timestamp> time = iter.next();
+ q.setParameter(time.getKey(), time.getValue());
+ }
+ long total = ((Long) q.getSingleResult()).longValue();
+ return total;
+ }
+
// Form the where clause to filter by coordinator names
private StringBuilder coordNamesClause(List<String> coordNames) {
StringBuilder sb = new StringBuilder();
boolean firstVal = true;
for (String name : nullToEmpty(coordNames)) {
- if(firstVal) {
+ if (firstVal) {
sb.append(" AND c.appName IN (\'" + name + "\'");
firstVal = false;
- } else {
+ }
+ else {
sb.append(",\'" + name + "\'");
}
}
- if(!firstVal) {
+ if (!firstVal) {
sb.append(") ");
}
return sb;
@@ -153,40 +186,49 @@ public class BulkJPAExecutor implements
StringBuilder sb = new StringBuilder();
boolean firstVal = true;
for (String status : nullToEmpty(statuses)) {
- if(firstVal) {
+ if (firstVal) {
sb.append(" AND a.status IN (\'" + status + "\'");
firstVal = false;
- } else {
+ }
+ else {
sb.append(",\'" + status + "\'");
}
}
- if(!firstVal) {
+ if (!firstVal) {
sb.append(") ");
- } else { // statuses was null. adding default
+ }
+ else { // statuses was null. adding default
sb.append(" AND a.status IN ('KILLED', 'FAILED') ");
}
return sb;
}
- private StringBuilder timesClause() throws ParseException {
- StringBuilder sb = new StringBuilder();
+ private void timesClause(StringBuilder sb, int offset, Map<String,
Timestamp> eachTime) throws ParseException {
+ Timestamp ts = null;
List<String> times =
bulkFilter.get(BulkResponseImpl.BULK_FILTER_START_CREATED_EPOCH);
- if(times != null) {
- sb.append(" AND a.createdTimestamp >= '" + new
Timestamp(DateUtils.parseDateUTC(times.get(0)).getTime()) + "'");
+ if (times != null) {
+ ts = new Timestamp(DateUtils.parseDateUTC(times.get(0)).getTime());
+ sb.insert(offset - 1, " AND a.createdTimestamp >= :startCreated");
+ eachTime.put("startCreated", ts);
}
times = bulkFilter.get(BulkResponseImpl.BULK_FILTER_END_CREATED_EPOCH);
- if(times != null) {
- sb.append(" AND a.createdTimestamp <= '" + new
Timestamp(DateUtils.parseDateUTC(times.get(0)).getTime()) + "'");
+ if (times != null) {
+ ts = new Timestamp(DateUtils.parseDateUTC(times.get(0)).getTime());
+ sb.insert(offset - 1, " AND a.createdTimestamp <= :endCreated");
+ eachTime.put("endCreated", ts);
}
times =
bulkFilter.get(BulkResponseImpl.BULK_FILTER_START_NOMINAL_EPOCH);
- if(times != null) {
- sb.append(" AND a.nominalTimestamp >= '" + new
Timestamp(DateUtils.parseDateUTC(times.get(0)).getTime()) + "'");
+ if (times != null) {
+ ts = new Timestamp(DateUtils.parseDateUTC(times.get(0)).getTime());
+ sb.insert(offset - 1, " AND a.nominalTimestamp >= :startNominal");
+ eachTime.put("startNominal", ts);
}
times = bulkFilter.get(BulkResponseImpl.BULK_FILTER_END_NOMINAL_EPOCH);
- if(times != null) {
- sb.append(" AND a.nominalTimestamp <= '" + new
Timestamp(DateUtils.parseDateUTC(times.get(0)).getTime()) + "'");
+ if (times != null) {
+ ts = new Timestamp(DateUtils.parseDateUTC(times.get(0)).getTime());
+ sb.insert(offset - 1, " AND a.nominalTimestamp <= :endNominal");
+ eachTime.put("endNominal", ts);
}
- return sb;
}
private BulkResponseImpl getResponseFromObject(BundleJobBean bundleBean,
Object arr[]) {
@@ -239,12 +281,14 @@ public class BulkJPAExecutor implements
return bean;
}
- private BundleJobBean getBeanForBundleJob(Object[] barr, String name)
throws Exception {
+ private BundleJobBean getBeanForBundleJob(Object[] barr, String name)
throws JPAExecutorException {
BundleJobBean bean = new BundleJobBean();
if (barr[0] != null) {
bean.setId((String) barr[0]);
- } else {
- throw new JPAExecutorException(ErrorCode.E0603, "bundleId returned
by query is null - cannot retrieve bulk results");
+ }
+ else {
+ throw new JPAExecutorException(ErrorCode.E0603,
+ "bundleId returned by query is null - cannot retrieve bulk
results");
}
bean.setAppName(name);
if (barr[1] != null) {
@@ -255,7 +299,7 @@ public class BulkJPAExecutor implements
// null safeguard
public static List<String> nullToEmpty(List<String> input) {
- return input == null ? Collections.<String>emptyList() : input;
+ return input == null ? Collections.<String> emptyList() : input;
}
}
Modified:
oozie/trunk/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java
URL:
http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java?rev=1401361&r1=1401360&r2=1401361&view=diff
==============================================================================
---
oozie/trunk/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java
(original)
+++
oozie/trunk/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java
Tue Oct 23 17:46:51 2012
@@ -29,7 +29,6 @@ import org.apache.oozie.BundleJobBean;
import org.apache.oozie.client.BundleJob;
import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.rest.BulkResponseImpl;
-import org.apache.oozie.local.LocalOozie;
import org.apache.oozie.service.JPAService;
import org.apache.oozie.service.Services;
import org.apache.oozie.test.XDataTestCase;
Modified: oozie/trunk/docs/src/site/twiki/WebServicesAPI.twiki
URL:
http://svn.apache.org/viewvc/oozie/trunk/docs/src/site/twiki/WebServicesAPI.twiki?rev=1401361&r1=1401360&r2=1401361&view=diff
==============================================================================
--- oozie/trunk/docs/src/site/twiki/WebServicesAPI.twiki (original)
+++ oozie/trunk/docs/src/site/twiki/WebServicesAPI.twiki Tue Oct 23 17:46:51
2012
@@ -790,35 +790,42 @@ The valid values of job type are: =workf
A HTTP GET request retrieves a bulk response for all actions, corresponding to
a particular bundle, that satisfy user specified criteria.
This is useful for monitoring purposes, where user can find out about the
status of downstream jobs with a single bulk request.
-The criteria are used for filtering the actions returned. Valid options (case
insensitive) for these request criteria are:
+The criteria are used for filtering the actions returned. Valid options (_case
insensitive_) for these request criteria are:
- * bundle: the application name from the bundle definition
- * coordinators: the application name(s) from the coordinator definition.
- * actionStatus: the status of coordinator action (Valid values are WAITING,
READY, SUBMITTED, RUNNING, SUSPENDED, TIMEDOUT, SUCCEEDED, KILLED, FAILED)
- * startCreatedTime: the start of the window you want to look at, of the
actions' created time
- * endCreatedTime: the end of above window
- * startScheduledTime: the start of the window you want to look at, of the
actions' scheduled i.e. nominal time.
- * endScheduledTime: the end of above window
+ * *bundle*: the application name from the bundle definition
+ * *coordinators*: the application name(s) from the coordinator definition.
+ * *actionStatus*: the status of coordinator action (Valid values are
WAITING, READY, SUBMITTED, RUNNING, SUSPENDED, TIMEDOUT, SUCCEEDED, KILLED,
FAILED)
+ * *startCreatedTime*: the start of the window you want to look at, of the
actions' created time
+ * *endCreatedTime*: the end of above window
+ * *startScheduledTime*: the start of the window you want to look at, of the
actions' scheduled i.e. nominal time.
+ * *endScheduledTime*: the end of above window
Specifying 'bundle' is REQUIRED. All the rest are OPTIONAL but that might
result in thousands of results depending on the size of your job. (pagination
comes into play then)
-The query will do an AND among all the filter names.
+If no 'actionStatus' values provided, by default KILLED,FAILED will be used.
+For e.g if the query string is only "bundle=MyBundle", the response will have
all actions (across all coordinators) whose status is KILLED or FAILED
+
+The query will do an AND among all the filter names, and OR among each filter
name's values.
The syntax for the request criteria is
<verbatim>[NAME=VALUE][;NAME=VALUE]*</verbatim>
-For 'coordinators' and 'actionStatus', if user wants to check for multiple
values, they can be passed in a comma-separated manner. *Note*: The query will
do an OR among them. Hence no need to repeat the criteria name
+For 'coordinators' and 'actionStatus', if user wants to check for multiple
values, they can be passed in a comma-separated manner.
+*Note*: The query will do an OR among them. Hence no need to repeat the
criteria name
-All the time parameter values should be specified in ISO8601 format i.e.
yyyy-MM-dd'T'HH:mm'Z'
+All the time values should be specified in *ISO8601 (UTC)* format i.e.
*yyyy-MM-dd'T'HH:mm'Z'*
Additionally the =offset= and =len= parameters can be used as usual for
pagination. The start parameter is base 1.
+If you specify a coordinator in the list, that does not exist, no error is
thrown; simply the response will be empty or pertaining to the other valid
coordinators.
+However, if bundle name provided does not exist, an error is thrown.
+
*Request:*
<verbatim>
GET
/oozie/v1/jobs?bulk=bundle%3Dmy-bundle-app;coordinators%3Dmy-coord-1,my-coord-5;actionStatus%3DKILLED&offset=1&len=50
</verbatim>
-Note that the filter is URL encoded, its decoded value is
<code>user=bansalm</code>. If typing in browser URL, one can type decoded value
itself i.e. using '='
+Note that the filter is URL encoded, its decoded value is
<code>user=chitnis</code>. If typing in browser URL, one can type decoded value
itself i.e. using '='
*Response:*
Modified: oozie/trunk/release-log.txt
URL:
http://svn.apache.org/viewvc/oozie/trunk/release-log.txt?rev=1401361&r1=1401360&r2=1401361&view=diff
==============================================================================
--- oozie/trunk/release-log.txt (original)
+++ oozie/trunk/release-log.txt Tue Oct 23 17:46:51 2012
@@ -1,5 +1,6 @@
-- Oozie 3.4.0 release (trunk - unreleased)
+OOZIE-1020 BulkJPAExecutor handling date-time value incorrectly.(Mona via
Mohammad)`
OOZIE-967 Coordinator action window in web UI never finishes refreshing
(kinley via tucu)
OOZIE-1024 ooziedb.sh script should keep OOZIE_CONFIG, OOZIE_LOG, and
OOZIE_DATA if previously set (rkanter via tucu)
OOZIE-1023 Docs list OOZIE_CONF instead of OOZIE_CONFIG env var (rkanter via
tucu)
