Permalink
Browse files

OOZIE-1118 improve logic of purge service (rkanter)

git-svn-id: https://svn.apache.org/repos/asf/oozie/trunk@1465054 13f79535-47bb-0310-9956-ffa450edef68
  • Loading branch information...
rkanter committed Apr 5, 2013
1 parent 27cd4e2 commit b93b6f2b34e191da8900f3cc72a8631b28b843e7
Showing with 6,379 additions and 1,457 deletions.
  1. +3 −1 core/src/main/java/org/apache/oozie/BundleActionBean.java
  2. +10 −1 core/src/main/java/org/apache/oozie/BundleJobBean.java
  3. +2 −0 core/src/main/java/org/apache/oozie/CoordinatorActionBean.java
  4. +8 −1 core/src/main/java/org/apache/oozie/CoordinatorJobBean.java
  5. +7 −1 core/src/main/java/org/apache/oozie/WorkflowJobBean.java
  6. +330 −0 core/src/main/java/org/apache/oozie/command/PurgeXCommand.java
  7. +0 −115 core/src/main/java/org/apache/oozie/command/bundle/BundlePurgeXCommand.java
  8. +0 −114 core/src/main/java/org/apache/oozie/command/coord/CoordPurgeXCommand.java
  9. +0 −98 core/src/main/java/org/apache/oozie/command/wf/PurgeXCommand.java
  10. +96 −0 core/src/main/java/org/apache/oozie/executor/jpa/BundleJobsDeleteJPAExecutor.java
  11. +10 −4 core/src/main/java/org/apache/oozie/executor/jpa/BundleJobsGetForPurgeJPAExecutor.java
  12. +69 −0 ...src/main/java/org/apache/oozie/executor/jpa/CoordJobsCountNotForPurgeFromParentIdJPAExecutor.java
  13. +96 −0 core/src/main/java/org/apache/oozie/executor/jpa/CoordJobsDeleteJPAExecutor.java
  14. +11 −5 core/src/main/java/org/apache/oozie/executor/jpa/CoordJobsGetForPurgeJPAExecutor.java
  15. +74 −0 core/src/main/java/org/apache/oozie/executor/jpa/CoordJobsGetFromParentIdJPAExecutor.java
  16. +69 −0 .../main/java/org/apache/oozie/executor/jpa/WorkflowJobsCountNotForPurgeFromParentIdJPAExecutor.java
  17. +19 −36 ...ache/oozie/executor/jpa/{BulkDeleteForPurgeJPAExecutor.java → WorkflowJobsDeleteJPAExecutor.java}
  18. +11 −5 core/src/main/java/org/apache/oozie/executor/jpa/WorkflowJobsGetForPurgeJPAExecutor.java
  19. +74 −0 core/src/main/java/org/apache/oozie/executor/jpa/WorkflowJobsGetFromParentIdJPAExecutor.java
  20. +6 −13 core/src/main/java/org/apache/oozie/service/PurgeService.java
  21. +3,850 −0 core/src/test/java/org/apache/oozie/command/TestPurgeXCommand.java
  22. +0 −278 core/src/test/java/org/apache/oozie/command/bundle/TestBundlePurgeXCommand.java
  23. +0 −227 core/src/test/java/org/apache/oozie/command/coord/TestCoordPurgeXCommand.java
  24. +13 −9 core/src/test/java/org/apache/oozie/command/wf/TestActionErrors.java
  25. +0 −297 core/src/test/java/org/apache/oozie/command/wf/TestPurgeXCommand.java
  26. +0 −216 core/src/test/java/org/apache/oozie/executor/jpa/TestBulkDeleteForPurgeJPAExecutor.java
  27. +253 −0 core/src/test/java/org/apache/oozie/executor/jpa/TestBundleJobsDeleteJPAExecutor.java
  28. +27 −13 core/src/test/java/org/apache/oozie/executor/jpa/TestBundleJobsGetForPurgeJPAExecutor.java
  29. +199 −0 ...test/java/org/apache/oozie/executor/jpa/TestCoordJobsCountNotForPurgeFromParentIdJPAExecutor.java
  30. +266 −0 core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobsDeleteJPAExecutor.java
  31. +52 −9 core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobsGetForPurgeJPAExecutor.java
  32. +149 −0 core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobsGetFromParentIdJPAExecutor.java
  33. +102 −0 ...t/java/org/apache/oozie/executor/jpa/TestWorkflowJobsCountNotForPurgeFromParentIdJPAExecutor.java
  34. +255 −0 core/src/test/java/org/apache/oozie/executor/jpa/TestWorkflowJobsDeleteJPAExecutor.java
  35. +42 −9 core/src/test/java/org/apache/oozie/executor/jpa/TestWorkflowJobsGetForPurgeJPAExecutor.java
  36. +199 −0 core/src/test/java/org/apache/oozie/executor/jpa/TestWorkflowJobsGetFromParentIdJPAExecutor.java
  37. +2 −2 core/src/test/java/org/apache/oozie/service/TestPurgeService.java
  38. +74 −3 core/src/test/java/org/apache/oozie/test/XDataTestCase.java
  39. +1 −0 release-log.txt
@@ -72,7 +72,9 @@
@NamedQuery(name = "GET_BUNDLE_ACTIONS_OLDER_THAN", query = "select OBJECT(w) from BundleActionBean w order by w.lastModifiedTimestamp"),
- @NamedQuery(name = "DELETE_COMPLETED_ACTIONS_FOR_BUNDLE", query = "delete from BundleActionBean a where a.bundleId = :bundleId and (a.status = 'SUCCEEDED' OR a.status = 'FAILED' OR a.status= 'KILLED' OR a.status = 'DONEWITHERROR')")})
+ @NamedQuery(name = "DELETE_COMPLETED_ACTIONS_FOR_BUNDLE", query = "delete from BundleActionBean a where a.bundleId = :bundleId and (a.status = 'SUCCEEDED' OR a.status = 'FAILED' OR a.status= 'KILLED' OR a.status = 'DONEWITHERROR')"),
+
+ @NamedQuery(name = "DELETE_ACTIONS_FOR_BUNDLE", query = "delete from BundleActionBean a where a.bundleId = :bundleId")})
public class BundleActionBean implements Writable, JsonBean {
@Id
@@ -66,7 +66,7 @@
@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 w.id 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"),
@@ -353,6 +353,15 @@ public void setLastModifiedTime(Date lastModifiedTime) {
this.lastModifiedTimestamp = DateUtils.convertDateToTimestamp(lastModifiedTime);
}
+ /**
+ * Get last modified time
+ *
+ * @return last modified time
+ */
+ public Date getLastModifiedTime() {
+ return DateUtils.toDate(lastModifiedTimestamp);
+ }
+
/* (non-Javadoc)
* @see org.apache.hadoop.io.Writable#write(java.io.DataOutput)
*/
@@ -65,6 +65,8 @@
@NamedQuery(name = "DELETE_COMPLETED_ACTIONS_FOR_COORDINATOR", query = "delete from CoordinatorActionBean a where a.jobId = :jobId and (a.status = 'SUCCEEDED' OR a.status = 'FAILED' OR a.status= 'KILLED')"),
+ @NamedQuery(name = "DELETE_ACTIONS_FOR_COORDINATOR", query = "delete from CoordinatorActionBean a where a.jobId = :jobId"),
+
@NamedQuery(name = "DELETE_UNSCHEDULED_ACTION", query = "delete from CoordinatorActionBean a where a.id = :id and (a.status = 'WAITING' OR a.status = 'READY')"),
// Query used by XTestcase to setup tables
@@ -63,11 +63,18 @@
@NamedQuery(name = "GET_COMPLETED_COORD_JOBS_OLDER_THAN_STATUS", query = "select OBJECT(w) from CoordinatorJobBean w where ( w.status = 'SUCCEEDED' OR w.status = 'FAILED' or w.status = 'KILLED') AND w.lastModifiedTimestamp <= :lastModTime order by w.lastModifiedTimestamp"),
+ @NamedQuery(name = "GET_COMPLETED_COORD_JOBS_WITH_NO_PARENT_OLDER_THAN_STATUS", query = "select w.id from CoordinatorJobBean w where ( w.status = 'SUCCEEDED' OR w.status = 'FAILED' or w.status = 'KILLED' or w.status = 'DONEWITHERROR') AND w.lastModifiedTimestamp <= :lastModTime and w.bundleId is null order by w.lastModifiedTimestamp"),
+
@NamedQuery(name = "GET_COORD_JOBS_UNPAUSED", query = "select OBJECT(w) from CoordinatorJobBean w where w.status = 'RUNNING' OR w.status = 'RUNNINGWITHERROR' OR w.status = 'PREP' order by w.lastModifiedTimestamp"),
@NamedQuery(name = "GET_COORD_JOBS_PAUSED", query = "select OBJECT(w) from CoordinatorJobBean w where w.status = 'PAUSED' OR w.status = 'PAUSEDWITHERROR' OR w.status = 'PREPPAUSED' order by w.lastModifiedTimestamp"),
- @NamedQuery(name = "GET_COORD_JOBS_FOR_BUNDLE", query = "select OBJECT(w) from CoordinatorJobBean w where w.bundleId = :bundleId order by w.lastModifiedTimestamp") })
+ @NamedQuery(name = "GET_COORD_JOBS_FOR_BUNDLE", query = "select OBJECT(w) from CoordinatorJobBean w where w.bundleId = :bundleId order by w.lastModifiedTimestamp"),
+
+ @NamedQuery(name = "GET_COORD_JOBS_WITH_PARENT_ID", query = "select w.id from CoordinatorJobBean w where w.bundleId = :parentId"),
+
+ @NamedQuery(name = "GET_COORD_COUNT_WITH_PARENT_ID_NOT_READY_FOR_PURGE", query = "select count(w) from CoordinatorJobBean w where w.bundleId = :parentId and (w.status NOT IN ('SUCCEEDED', 'FAILED', 'KILLED', 'DONEWITHERROR') OR w.lastModifiedTimestamp >= :lastModTime)")
+})
public class CoordinatorJobBean extends JsonCoordinatorJob implements Writable {
@Basic
@@ -57,6 +57,8 @@
@NamedQuery(name = "GET_COMPLETED_WORKFLOWS_OLDER_THAN", query = "select w from WorkflowJobBean w where w.endTimestamp < :endTime"),
+ @NamedQuery(name = "GET_COMPLETED_WORKFLOWS_WITH_NO_PARENT_OLDER_THAN", query = "select w.id from WorkflowJobBean w where w.endTimestamp < :endTime and w.parentId is null"),
+
@NamedQuery(name = "GET_WORKFLOW", query = "select OBJECT(w) from WorkflowJobBean w where w.id = :id"),
@NamedQuery(name = "GET_WORKFLOW_FOR_UPDATE", query = "select OBJECT(w) from WorkflowJobBean w where w.id = :id"),
@@ -65,7 +67,11 @@
@NamedQuery(name = "GET_WORKFLOWS_COUNT_WITH_STATUS", query = "select count(w) from WorkflowJobBean w where w.status = :status"),
- @NamedQuery(name = "GET_WORKFLOWS_COUNT_WITH_STATUS_IN_LAST_N_SECS", query = "select count(w) from WorkflowJobBean w where w.status = :status and w.lastModifiedTimestamp > :lastModTime")
+ @NamedQuery(name = "GET_WORKFLOWS_COUNT_WITH_STATUS_IN_LAST_N_SECS", query = "select count(w) from WorkflowJobBean w where w.status = :status and w.lastModifiedTimestamp > :lastModTime"),
+
+ @NamedQuery(name = "GET_WORKFLOWS_WITH_PARENT_ID", query = "select w.id from WorkflowJobBean w where w.parentId = :parentId"),
+
+ @NamedQuery(name = "GET_WORKFLOWS_COUNT_WITH_PARENT_ID_NOT_READY_FOR_PURGE", query = "select count(w) from WorkflowJobBean w where w.parentId = :parentId and (w.status = 'PREP' OR w.status = 'RUNNING' OR w.status = 'SUSPENDED' OR w.endTimestamp >= :endTime)")
})
public class WorkflowJobBean extends JsonWorkflowJob implements Writable {
Oops, something went wrong.

0 comments on commit b93b6f2

Please sign in to comment.