001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018 019package org.apache.oozie.executor.jpa; 020 021import java.sql.Timestamp; 022import java.util.ArrayList; 023import java.util.List; 024 025import javax.persistence.EntityManager; 026import javax.persistence.Query; 027 028import org.apache.oozie.ErrorCode; 029import org.apache.oozie.StringBlob; 030import org.apache.oozie.WorkflowActionBean; 031import org.apache.oozie.service.JPAService; 032import org.apache.oozie.service.Services; 033import org.apache.oozie.util.DateUtils; 034 035/** 036 * Query Executor that provides API to run query for Workflow Action 037 */ 038public class WorkflowActionQueryExecutor extends 039 QueryExecutor<WorkflowActionBean, WorkflowActionQueryExecutor.WorkflowActionQuery> { 040 041 public enum WorkflowActionQuery { 042 UPDATE_ACTION, 043 UPDATE_ACTION_FOR_LAST_CHECKED_TIME, 044 UPDATE_ACTION_START, 045 UPDATE_ACTION_CHECK, 046 UPDATE_ACTION_END, 047 UPDATE_ACTION_PENDING, 048 UPDATE_ACTION_STATUS_PENDING, 049 UPDATE_ACTION_PENDING_TRANS, 050 UPDATE_ACTION_PENDING_TRANS_ERROR, 051 GET_ACTION, 052 GET_ACTION_ID_TYPE_LASTCHECK, 053 GET_ACTION_FAIL, 054 GET_ACTION_SIGNAL, 055 GET_ACTION_CHECK, 056 GET_ACTION_END, 057 GET_ACTION_COMPLETED, 058 GET_RUNNING_ACTIONS, 059 GET_PENDING_ACTIONS, 060 GET_ACTIONS_FOR_WORKFLOW_RERUN 061 }; 062 063 private static WorkflowActionQueryExecutor instance = new WorkflowActionQueryExecutor(); 064 065 private WorkflowActionQueryExecutor() { 066 } 067 068 public static QueryExecutor<WorkflowActionBean, WorkflowActionQuery> getInstance() { 069 return WorkflowActionQueryExecutor.instance; 070 } 071 072 @Override 073 public Query getUpdateQuery(WorkflowActionQuery namedQuery, WorkflowActionBean actionBean, EntityManager em) 074 throws JPAExecutorException { 075 Query query = em.createNamedQuery(namedQuery.name()); 076 switch (namedQuery) { 077 case UPDATE_ACTION: 078 query.setParameter("conf", actionBean.getConfBlob()); 079 query.setParameter("consoleUrl", actionBean.getConsoleUrl()); 080 query.setParameter("data", actionBean.getDataBlob()); 081 query.setParameter("stats", actionBean.getStatsBlob()); 082 query.setParameter("externalChildIDs", actionBean.getExternalChildIDsBlob()); 083 query.setParameter("errorCode", actionBean.getErrorCode()); 084 query.setParameter("errorMessage", actionBean.getErrorMessage()); 085 query.setParameter("externalId", actionBean.getExternalId()); 086 query.setParameter("externalStatus", actionBean.getExternalStatus()); 087 query.setParameter("name", actionBean.getName()); 088 query.setParameter("cred", actionBean.getCred()); 089 query.setParameter("retries", actionBean.getRetries()); 090 query.setParameter("trackerUri", actionBean.getTrackerUri()); 091 query.setParameter("transition", actionBean.getTransition()); 092 query.setParameter("type", actionBean.getType()); 093 query.setParameter("endTime", actionBean.getEndTimestamp()); 094 query.setParameter("executionPath", actionBean.getExecutionPath()); 095 query.setParameter("lastCheckTime", actionBean.getLastCheckTimestamp()); 096 query.setParameter("logToken", actionBean.getLogToken()); 097 query.setParameter("pending", actionBean.getPending()); 098 query.setParameter("pendingAge", actionBean.getPendingAgeTimestamp()); 099 query.setParameter("signalValue", actionBean.getSignalValue()); 100 query.setParameter("slaXml", actionBean.getSlaXmlBlob()); 101 query.setParameter("startTime", actionBean.getStartTimestamp()); 102 query.setParameter("status", actionBean.getStatusStr()); 103 query.setParameter("wfId", actionBean.getWfId()); 104 query.setParameter("id", actionBean.getId()); 105 break; 106 case UPDATE_ACTION_FOR_LAST_CHECKED_TIME: 107 query.setParameter("lastCheckTime", actionBean.getLastCheckTimestamp()); 108 query.setParameter("id", actionBean.getId()); 109 break; 110 case UPDATE_ACTION_PENDING: 111 query.setParameter("pending", actionBean.getPending()); 112 query.setParameter("pendingAge", actionBean.getPendingAgeTimestamp()); 113 query.setParameter("executionPath", actionBean.getExecutionPath()); 114 query.setParameter("id", actionBean.getId()); 115 break; 116 case UPDATE_ACTION_STATUS_PENDING: 117 query.setParameter("status", actionBean.getStatus().toString()); 118 query.setParameter("pending", actionBean.getPending()); 119 query.setParameter("pendingAge", actionBean.getPendingAgeTimestamp()); 120 query.setParameter("id", actionBean.getId()); 121 break; 122 case UPDATE_ACTION_PENDING_TRANS: 123 query.setParameter("transition", actionBean.getTransition()); 124 query.setParameter("pending", actionBean.getPending()); 125 query.setParameter("pendingAge", actionBean.getPendingAgeTimestamp()); 126 query.setParameter("id", actionBean.getId()); 127 break; 128 case UPDATE_ACTION_PENDING_TRANS_ERROR: 129 query.setParameter("transition", actionBean.getTransition()); 130 query.setParameter("pending", actionBean.getPending()); 131 query.setParameter("pendingAge", actionBean.getPendingAgeTimestamp()); 132 query.setParameter("errorCode", actionBean.getErrorCode()); 133 query.setParameter("errorMessage", actionBean.getErrorMessage()); 134 query.setParameter("status", actionBean.getStatusStr()); 135 query.setParameter("id", actionBean.getId()); 136 break; 137 case UPDATE_ACTION_START: 138 query.setParameter("startTime", actionBean.getStartTimestamp()); 139 query.setParameter("externalChildIDs", actionBean.getExternalChildIDsBlob()); 140 query.setParameter("conf", actionBean.getConfBlob()); 141 query.setParameter("errorCode", actionBean.getErrorCode()); 142 query.setParameter("errorMessage", actionBean.getErrorMessage()); 143 query.setParameter("externalId", actionBean.getExternalId()); 144 query.setParameter("trackerUri", actionBean.getTrackerUri()); 145 query.setParameter("consoleUrl", actionBean.getConsoleUrl()); 146 query.setParameter("lastCheckTime", actionBean.getLastCheckTimestamp()); 147 query.setParameter("status", actionBean.getStatus().toString()); 148 query.setParameter("externalStatus", actionBean.getExternalStatus()); 149 query.setParameter("data", actionBean.getDataBlob()); 150 query.setParameter("retries", actionBean.getRetries()); 151 query.setParameter("pending", actionBean.getPending()); 152 query.setParameter("pendingAge", actionBean.getPendingAgeTimestamp()); 153 query.setParameter("userRetryCount", actionBean.getUserRetryCount()); 154 query.setParameter("id", actionBean.getId()); 155 break; 156 case UPDATE_ACTION_CHECK: 157 query.setParameter("externalChildIDs", actionBean.getExternalChildIDsBlob()); 158 query.setParameter("externalStatus", actionBean.getExternalStatus()); 159 query.setParameter("status", actionBean.getStatus().toString()); 160 query.setParameter("data", actionBean.getDataBlob()); 161 query.setParameter("pending", actionBean.getPending()); 162 query.setParameter("errorCode", actionBean.getErrorCode()); 163 query.setParameter("errorMessage", actionBean.getErrorMessage()); 164 query.setParameter("lastCheckTime", actionBean.getLastCheckTimestamp()); 165 query.setParameter("retries", actionBean.getRetries()); 166 query.setParameter("pendingAge", actionBean.getPendingAgeTimestamp()); 167 query.setParameter("startTime", actionBean.getStartTimestamp()); 168 query.setParameter("stats", actionBean.getStatsBlob()); 169 query.setParameter("userRetryCount", actionBean.getUserRetryCount()); 170 query.setParameter("id", actionBean.getId()); 171 break; 172 case UPDATE_ACTION_END: 173 query.setParameter("errorCode", actionBean.getErrorCode()); 174 query.setParameter("errorMessage", actionBean.getErrorMessage()); 175 query.setParameter("retries", actionBean.getRetries()); 176 query.setParameter("status", actionBean.getStatus().toString()); 177 query.setParameter("endTime", actionBean.getEndTimestamp()); 178 query.setParameter("pending", actionBean.getPending()); 179 query.setParameter("pendingAge", actionBean.getPendingAgeTimestamp()); 180 query.setParameter("signalValue", actionBean.getSignalValue()); 181 query.setParameter("userRetryCount", actionBean.getUserRetryCount()); 182 query.setParameter("externalStatus", actionBean.getExternalStatus()); 183 query.setParameter("stats", actionBean.getStatsBlob()); 184 query.setParameter("id", actionBean.getId()); 185 break; 186 default: 187 throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot set parameters for " 188 + namedQuery.name()); 189 } 190 return query; 191 } 192 193 @Override 194 public Query getSelectQuery(WorkflowActionQuery namedQuery, EntityManager em, Object... parameters) 195 throws JPAExecutorException { 196 Query query = em.createNamedQuery(namedQuery.name()); 197 switch (namedQuery) { 198 case GET_ACTION: 199 case GET_ACTION_ID_TYPE_LASTCHECK: 200 case GET_ACTION_FAIL: 201 case GET_ACTION_SIGNAL: 202 case GET_ACTION_CHECK: 203 case GET_ACTION_END: 204 case GET_ACTION_COMPLETED: 205 query.setParameter("id", parameters[0]); 206 break; 207 case GET_RUNNING_ACTIONS: 208 Timestamp ts = new Timestamp(System.currentTimeMillis() - (Integer) parameters[0] * 1000); 209 query.setParameter("lastCheckTime", ts); 210 break; 211 case GET_PENDING_ACTIONS: 212 Long minimumPendingAgeSecs = (Long) parameters[0]; 213 Timestamp pts = new Timestamp(System.currentTimeMillis() - minimumPendingAgeSecs * 1000); 214 Timestamp createdTimeInterval = new Timestamp((Long) parameters[1]); 215 query.setParameter("pendingAge", pts); 216 query.setParameter("createdTime", createdTimeInterval); 217 break; 218 case GET_ACTIONS_FOR_WORKFLOW_RERUN: 219 query.setParameter("wfId", parameters[0]); 220 break; 221 default: 222 throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot set parameters for " 223 + namedQuery.name()); 224 } 225 return query; 226 } 227 228 @Override 229 public int executeUpdate(WorkflowActionQuery namedQuery, WorkflowActionBean actionBean) throws JPAExecutorException { 230 JPAService jpaService = Services.get().get(JPAService.class); 231 EntityManager em = jpaService.getEntityManager(); 232 Query query = getUpdateQuery(namedQuery, actionBean, em); 233 int ret = jpaService.executeUpdate(namedQuery.name(), query, em); 234 return ret; 235 } 236 237 private WorkflowActionBean constructBean(WorkflowActionQuery namedQuery, Object ret) throws JPAExecutorException { 238 WorkflowActionBean bean; 239 Object[] arr; 240 switch (namedQuery) { 241 case GET_ACTION: 242 bean = (WorkflowActionBean) ret; 243 break; 244 case GET_ACTION_ID_TYPE_LASTCHECK: 245 bean = new WorkflowActionBean(); 246 arr = (Object[]) ret; 247 bean.setId((String) arr[0]); 248 bean.setType((String) arr[1]); 249 bean.setLastCheckTime(DateUtils.toDate((Timestamp) arr[2])); 250 break; 251 case GET_ACTION_FAIL: 252 bean = new WorkflowActionBean(); 253 arr = (Object[]) ret; 254 bean.setId((String) arr[0]); 255 bean.setJobId((String) arr[1]); 256 bean.setName((String) arr[2]); 257 bean.setStatusStr((String) arr[3]); 258 bean.setPending((Integer) arr[4]); 259 bean.setType((String) arr[5]); 260 bean.setLogToken((String) arr[6]); 261 bean.setTransition((String) arr[7]); 262 bean.setErrorInfo((String) arr[8], (String) arr[9]); 263 break; 264 case GET_ACTION_SIGNAL: 265 bean = new WorkflowActionBean(); 266 arr = (Object[]) ret; 267 bean.setId((String) arr[0]); 268 bean.setJobId((String) arr[1]); 269 bean.setName((String) arr[2]); 270 bean.setStatusStr((String) arr[3]); 271 bean.setPending((Integer) arr[4]); 272 bean.setPendingAge(DateUtils.toDate((Timestamp) arr[5])); 273 bean.setType((String) arr[6]); 274 bean.setLogToken((String) arr[7]); 275 bean.setTransition((String) arr[8]); 276 bean.setErrorInfo((String) arr[9], (String) arr[10]); 277 bean.setExecutionPath((String) arr[11]); 278 bean.setSignalValue((String) arr[12]); 279 bean.setSlaXmlBlob((StringBlob) arr[13]); 280 bean.setExternalId((String) arr[14]); 281 break; 282 case GET_ACTION_CHECK: 283 bean = new WorkflowActionBean(); 284 arr = (Object[]) ret; 285 bean.setId((String) arr[0]); 286 bean.setJobId((String) arr[1]); 287 bean.setName((String) arr[2]); 288 bean.setStatusStr((String) arr[3]); 289 bean.setPending((Integer) arr[4]); 290 bean.setPendingAge(DateUtils.toDate((Timestamp) arr[5])); 291 bean.setType((String) arr[6]); 292 bean.setLogToken((String) arr[7]); 293 bean.setTransition((String) arr[8]); 294 bean.setRetries((Integer) arr[9]); 295 bean.setUserRetryCount((Integer) arr[10]); 296 bean.setUserRetryMax((Integer) arr[11]); 297 bean.setUserRetryInterval((Integer) arr[12]); 298 bean.setTrackerUri((String) arr[13]); 299 bean.setStartTime(DateUtils.toDate((Timestamp) arr[14])); 300 bean.setEndTime(DateUtils.toDate((Timestamp) arr[15])); 301 bean.setLastCheckTime(DateUtils.toDate((Timestamp) arr[16])); 302 bean.setErrorInfo((String) arr[17], (String) arr[18]); 303 bean.setExternalId((String) arr[19]); 304 bean.setExternalStatus((String) arr[20]); 305 bean.setExternalChildIDsBlob((StringBlob) arr[21]); 306 bean.setConfBlob((StringBlob) arr[22]); 307 break; 308 case GET_ACTION_END: 309 bean = new WorkflowActionBean(); 310 arr = (Object[]) ret; 311 bean.setId((String) arr[0]); 312 bean.setJobId((String) arr[1]); 313 bean.setName((String) arr[2]); 314 bean.setStatusStr((String) arr[3]); 315 bean.setPending((Integer) arr[4]); 316 bean.setPendingAge(DateUtils.toDate((Timestamp) arr[5])); 317 bean.setType((String) arr[6]); 318 bean.setLogToken((String) arr[7]); 319 bean.setTransition((String) arr[8]); 320 bean.setRetries((Integer) arr[9]); 321 bean.setTrackerUri((String) arr[10]); 322 bean.setUserRetryCount((Integer) arr[11]); 323 bean.setUserRetryMax((Integer) arr[12]); 324 bean.setUserRetryInterval((Integer) arr[13]); 325 bean.setStartTime(DateUtils.toDate((Timestamp) arr[14])); 326 bean.setEndTime(DateUtils.toDate((Timestamp) arr[15])); 327 bean.setErrorInfo((String) arr[16], (String) arr[17]); 328 bean.setExternalId((String) arr[18]); 329 bean.setExternalStatus((String) arr[19]); 330 bean.setExternalChildIDsBlob((StringBlob) arr[20]); 331 bean.setConfBlob((StringBlob) arr[21]); 332 bean.setDataBlob((StringBlob) arr[22]); 333 bean.setStatsBlob((StringBlob) arr[23]); 334 break; 335 case GET_ACTION_COMPLETED: 336 bean = new WorkflowActionBean(); 337 arr = (Object[]) ret; 338 bean.setId((String) arr[0]); 339 bean.setJobId((String) arr[1]); 340 bean.setStatusStr((String) arr[2]); 341 bean.setType((String) arr[3]); 342 bean.setLogToken((String) arr[4]); 343 break; 344 case GET_RUNNING_ACTIONS: 345 bean = new WorkflowActionBean(); 346 bean.setId((String)ret); 347 break; 348 case GET_PENDING_ACTIONS: 349 bean = new WorkflowActionBean(); 350 arr = (Object[]) ret; 351 bean.setId((String) arr[0]); 352 bean.setJobId((String) arr[1]); 353 bean.setStatusStr((String) arr[2]); 354 bean.setType((String) arr[3]); 355 bean.setPendingAge(DateUtils.toDate((Timestamp) arr[4])); 356 break; 357 case GET_ACTIONS_FOR_WORKFLOW_RERUN: 358 bean = new WorkflowActionBean(); 359 arr = (Object[]) ret; 360 bean.setId((String) arr[0]); 361 bean.setName((String) arr[1]); 362 bean.setStatusStr((String) arr[2]); 363 bean.setEndTime(DateUtils.toDate((Timestamp) arr[3])); 364 bean.setType((String) arr[4]); 365 break; 366 default: 367 throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot construct action bean for " 368 + namedQuery.name()); 369 } 370 return bean; 371 } 372 373 @Override 374 public WorkflowActionBean get(WorkflowActionQuery namedQuery, Object... parameters) throws JPAExecutorException { 375 JPAService jpaService = Services.get().get(JPAService.class); 376 EntityManager em = jpaService.getEntityManager(); 377 Query query = getSelectQuery(namedQuery, em, parameters); 378 Object ret = jpaService.executeGet(namedQuery.name(), query, em); 379 if (ret == null) { 380 throw new JPAExecutorException(ErrorCode.E0605, query.toString()); 381 } 382 WorkflowActionBean bean = constructBean(namedQuery, ret); 383 return bean; 384 } 385 386 @Override 387 public List<WorkflowActionBean> getList(WorkflowActionQuery namedQuery, Object... parameters) 388 throws JPAExecutorException { 389 JPAService jpaService = Services.get().get(JPAService.class); 390 EntityManager em = jpaService.getEntityManager(); 391 Query query = getSelectQuery(namedQuery, em, parameters); 392 List<?> retList = (List<?>) jpaService.executeGetList(namedQuery.name(), query, em); 393 List<WorkflowActionBean> beanList = new ArrayList<WorkflowActionBean>(); 394 if (retList != null) { 395 for (Object ret : retList) { 396 beanList.add(constructBean(namedQuery, ret)); 397 } 398 } 399 return beanList; 400 } 401 402 @Override 403 public Object getSingleValue(WorkflowActionQuery namedQuery, Object... parameters) throws JPAExecutorException { 404 throw new UnsupportedOperationException(); 405 } 406}