You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@oozie.apache.org by ka...@apache.org on 2012/10/23 19:44:47 UTC
svn commit: r1401360 - in /oozie/branches/branch-3.3: ./
core/src/main/java/org/apache/oozie/
core/src/main/java/org/apache/oozie/executor/jpa/
core/src/test/java/org/apache/oozie/executor/jpa/ docs/src/site/twiki/
Author: kamrul
Date: Tue Oct 23 17:44:46 2012
New Revision: 1401360
URL: http://svn.apache.org/viewvc?rev=1401360&view=rev
Log:
OOZIE-1020 BulkJPAExecutor handling date-time value incorrectly.(Mona via Mohammad)
Modified:
oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/BundleJobBean.java
oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java
oozie/branches/branch-3.3/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java
oozie/branches/branch-3.3/docs/src/site/twiki/WebServicesAPI.twiki
oozie/branches/branch-3.3/release-log.txt
Modified: oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/BundleJobBean.java
URL: http://svn.apache.org/viewvc/oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/BundleJobBean.java?rev=1401360&r1=1401359&r2=1401360&view=diff
==============================================================================
--- oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/BundleJobBean.java (original)
+++ oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/BundleJobBean.java Tue Oct 23 17:44:46 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/branches/branch-3.3/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java
URL: http://svn.apache.org/viewvc/oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java?rev=1401360&r1=1401359&r2=1401360&view=diff
==============================================================================
--- oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java (original)
+++ oozie/branches/branch-3.3/core/src/main/java/org/apache/oozie/executor/jpa/BulkJPAExecutor.java Tue Oct 23 17:44:46 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/branches/branch-3.3/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java
URL: http://svn.apache.org/viewvc/oozie/branches/branch-3.3/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java?rev=1401360&r1=1401359&r2=1401360&view=diff
==============================================================================
--- oozie/branches/branch-3.3/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java (original)
+++ oozie/branches/branch-3.3/core/src/test/java/org/apache/oozie/executor/jpa/TestBulkMonitorJPAExecutor.java Tue Oct 23 17:44:46 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/branches/branch-3.3/docs/src/site/twiki/WebServicesAPI.twiki
URL: http://svn.apache.org/viewvc/oozie/branches/branch-3.3/docs/src/site/twiki/WebServicesAPI.twiki?rev=1401360&r1=1401359&r2=1401360&view=diff
==============================================================================
--- oozie/branches/branch-3.3/docs/src/site/twiki/WebServicesAPI.twiki (original)
+++ oozie/branches/branch-3.3/docs/src/site/twiki/WebServicesAPI.twiki Tue Oct 23 17:44:46 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/branches/branch-3.3/release-log.txt
URL: http://svn.apache.org/viewvc/oozie/branches/branch-3.3/release-log.txt?rev=1401360&r1=1401359&r2=1401360&view=diff
==============================================================================
--- oozie/branches/branch-3.3/release-log.txt (original)
+++ oozie/branches/branch-3.3/release-log.txt Tue Oct 23 17:44:46 2012
@@ -1,5 +1,6 @@
-- Oozie 3.3.0 release (unreleased)
+OOZIE-1020 BulkJPAExecutor handling date-time value incorrectly.(Mona via Mohammad)
OOZIE-976 add workflowgenerator into distro tarball (egashira via tucu)
OOZIE-993 Hadoop 23 doesn't accept user defined jobtracker (virag)
OOZIE-1013 Build failing as the license header comment is appearing before xml declaration in some files (virag)