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; 020 021import org.apache.hadoop.io.Writable; 022import org.apache.oozie.client.CoordinatorAction; 023import org.apache.oozie.client.rest.JsonBean; 024import org.apache.oozie.client.rest.JsonTags; 025import org.apache.oozie.client.rest.JsonUtils; 026import org.apache.oozie.util.DateUtils; 027import org.apache.oozie.util.WritableUtils; 028import org.apache.openjpa.persistence.jdbc.Index; 029import org.apache.openjpa.persistence.jdbc.Strategy; 030import org.json.simple.JSONArray; 031import org.json.simple.JSONObject; 032 033import javax.persistence.Basic; 034import javax.persistence.Column; 035import javax.persistence.ColumnResult; 036import javax.persistence.Entity; 037import javax.persistence.Id; 038import javax.persistence.Lob; 039import javax.persistence.NamedNativeQueries; 040import javax.persistence.NamedNativeQuery; 041import javax.persistence.NamedQueries; 042import javax.persistence.NamedQuery; 043import javax.persistence.SqlResultSetMapping; 044import javax.persistence.Table; 045import java.io.DataInput; 046import java.io.DataOutput; 047import java.io.IOException; 048import java.sql.Timestamp; 049import java.text.MessageFormat; 050import java.util.Date; 051import java.util.List; 052 053@Entity 054@NamedQueries({ 055 056 @NamedQuery(name = "UPDATE_COORD_ACTION", query = "update CoordinatorActionBean w set w.actionNumber = :actionNumber, w.actionXml = :actionXml, w.consoleUrl = :consoleUrl, w.createdConf = :createdConf, w.errorCode = :errorCode, w.errorMessage = :errorMessage, w.externalStatus = :externalStatus, w.missingDependencies = :missingDependencies, w.runConf = :runConf, w.timeOut = :timeOut, w.trackerUri = :trackerUri, w.type = :type, w.createdTimestamp = :createdTime, w.externalId = :externalId, w.jobId = :jobId, w.lastModifiedTimestamp = :lastModifiedTime, w.nominalTimestamp = :nominalTime, w.slaXml = :slaXml, w.statusStr = :status where w.id = :id"), 057 058 @NamedQuery(name = "UPDATE_COORD_ACTION_MIN", query = "update CoordinatorActionBean w set w.actionXml = :actionXml, w.missingDependencies = :missingDependencies, w.lastModifiedTimestamp = :lastModifiedTime, w.statusStr = :status where w.id = :id"), 059 // Query to update the action status, pending status and last modified time stamp of a Coordinator action 060 @NamedQuery(name = "UPDATE_COORD_ACTION_STATUS_PENDING_TIME", query = "update CoordinatorActionBean w set w.statusStr =:status, w.pending =:pending, w.lastModifiedTimestamp = :lastModifiedTime where w.id = :id"), 061 // Update query for InputCheck 062 @NamedQuery(name = "UPDATE_COORD_ACTION_FOR_INPUTCHECK", query = "update CoordinatorActionBean w set w.statusStr = :status, w.lastModifiedTimestamp = :lastModifiedTime, w.actionXml = :actionXml, w.missingDependencies = :missingDependencies where w.id = :id"), 063 // Update query for Push-based missing dependency check 064 @NamedQuery(name = "UPDATE_COORD_ACTION_FOR_PUSH_INPUTCHECK", query = "update CoordinatorActionBean w set w.statusStr = :status, w.lastModifiedTimestamp = :lastModifiedTime, w.actionXml = :actionXml, w.pushMissingDependencies = :pushMissingDependencies where w.id = :id"), 065 // Update query for Push-based missing dependency check 066 @NamedQuery(name = "UPDATE_COORD_ACTION_DEPENDENCIES", query = "update CoordinatorActionBean w set w.missingDependencies = :missingDependencies, w.pushMissingDependencies = :pushMissingDependencies where w.id = :id"), 067 // Update query for Start 068 @NamedQuery(name = "UPDATE_COORD_ACTION_FOR_START", query = "update CoordinatorActionBean w set w.statusStr =:status, w.lastModifiedTimestamp = :lastModifiedTime, w.runConf = :runConf, w.externalId = :externalId, w.pending = :pending, w.errorCode = :errorCode, w.errorMessage = :errorMessage where w.id = :id"), 069 070 @NamedQuery(name = "UPDATE_COORD_ACTION_FOR_MODIFIED_DATE", query = "update CoordinatorActionBean w set w.lastModifiedTimestamp = :lastModifiedTime where w.id = :id"), 071 072 @NamedQuery(name = "UPDATE_COORD_ACTION_RERUN", query = "update CoordinatorActionBean w set w.actionXml =:actionXml, w.statusStr = :status, w.externalId = :externalId, w.externalStatus = :externalStatus, w.rerunTimestamp = :rerunTime, w.lastModifiedTimestamp = :lastModifiedTime, w.createdTimestamp = :createdTime, w.createdConf = :createdConf, w.runConf = :runConf, w.missingDependencies = :missingDependencies, w.pushMissingDependencies = :pushMissingDependencies, w.errorCode = :errorCode, w.errorMessage = :errorMessage where w.id = :id"), 073 074 @NamedQuery(name = "DELETE_COMPLETED_ACTIONS_FOR_COORDINATOR", query = "delete from CoordinatorActionBean a where a.jobId = :jobId and (a.statusStr = 'SUCCEEDED' OR a.statusStr = 'FAILED' OR a.statusStr= 'KILLED')"), 075 076 @NamedQuery(name = "DELETE_ACTIONS_FOR_LONG_RUNNING_COORDINATOR", query = "delete from CoordinatorActionBean a where a.id IN (:actionId)"), 077 078 @NamedQuery(name = "DELETE_UNSCHEDULED_ACTION", query = "delete from CoordinatorActionBean a where a.id = :id and (a.statusStr = 'WAITING' OR a.statusStr = 'READY')"), 079 080 @NamedQuery(name = "GET_COORD_ACTIONS_FOR_COORDINATOR", query = "select a.id from CoordinatorActionBean a where a.jobId = :jobId"), 081 082 // Query used by XTestcase to setup tables 083 @NamedQuery(name = "GET_COORD_ACTIONS", query = "select OBJECT(w) from CoordinatorActionBean w"), 084 // Select query used only by test cases 085 @NamedQuery(name = "GET_COORD_ACTION", query = "select OBJECT(a) from CoordinatorActionBean a where a.id = :id"), 086 087 // Select query used by SLAService on restart 088 @NamedQuery(name = "GET_COORD_ACTION_FOR_SLA", query = "select a.id, a.jobId, a.statusStr, a.externalId, a.lastModifiedTimestamp from CoordinatorActionBean a where a.id = :id"), 089 // Select query used by ActionInfo command 090 @NamedQuery(name = "GET_COORD_ACTION_FOR_INFO", query = "select a.id, a.jobId, a.actionNumber, a.consoleUrl, a.errorCode, a.errorMessage, a.externalId, a.externalStatus, a.trackerUri, a.createdTimestamp, a.nominalTimestamp, a.statusStr, a.lastModifiedTimestamp, a.missingDependencies, a.pushMissingDependencies from CoordinatorActionBean a where a.id = :id"), 091 // Select Query used by Timeout and skip commands 092 @NamedQuery(name = "GET_COORD_ACTION_FOR_TIMEOUT", query = "select a.id, a.jobId, a.statusStr, a.runConf, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.id = :id"), 093 // Select query used by InputCheck command 094 @NamedQuery(name = "GET_COORD_ACTION_FOR_INPUTCHECK", query = "select a.id, a.actionNumber, a.jobId, a.statusStr, a.runConf, a.nominalTimestamp, a.createdTimestamp, a.actionXml, a.missingDependencies, a.pushMissingDependencies, a.timeOut, a.externalId from CoordinatorActionBean a where a.id = :id"), 095 // Select query used by CoordActionUpdate command 096 @NamedQuery(name = "GET_COORD_ACTION_FOR_EXTERNALID", query = "select a.id, a.jobId, a.statusStr, a.pending, a.externalId, a.lastModifiedTimestamp, a.slaXml, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.externalId = :externalId"), 097 // Select query used by Check command 098 @NamedQuery(name = "GET_COORD_ACTION_FOR_CHECK", query = "select a.id, a.jobId, a.statusStr, a.pending, a.externalId, a.lastModifiedTimestamp, a.slaXml, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.id = :id"), 099 // Select query used by Start command 100 @NamedQuery(name = "GET_COORD_ACTION_FOR_START", query = "select a.id, a.jobId, a.statusStr, a.pending, a.createdConf, a.slaXml, a.actionXml, a.externalId, a.errorMessage, a.errorCode, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.id = :id"), 101 102 @NamedQuery(name = "GET_COORD_ACTIONS_FOR_JOB_FIFO", query = "select a.id, a.actionNumber, a.jobId, a.statusStr, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr = 'READY' order by a.nominalTimestamp"), 103 104 @NamedQuery(name = "GET_COORD_ACTIONS_FOR_JOB_LIFO", query = "select a.id, a.actionNumber, a.jobId, a.statusStr, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr = 'READY' order by a.nominalTimestamp desc"), 105 106 @NamedQuery(name = "GET_COORD_RUNNING_ACTIONS_COUNT", query = "select count(a) from CoordinatorActionBean a where a.jobId = :jobId AND (a.statusStr = 'RUNNING' OR a.statusStr='SUBMITTED')"), 107 108 @NamedQuery(name = "GET_COORD_ACTIONS_COUNT_BY_JOBID", query = "select count(a) from CoordinatorActionBean a where a.jobId = :jobId"), 109 110 @NamedQuery(name = "GET_COORD_ACTIVE_ACTIONS_COUNT_BY_JOBID", query = "select count(a) from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr = 'WAITING'"), 111 112 @NamedQuery(name = "GET_COORD_ACTIONS_PENDING_FALSE_COUNT", query = "select count(a) from CoordinatorActionBean a where a.jobId = :jobId AND a.pending = 0 AND (a.statusStr = 'SUSPENDED' OR a.statusStr = 'TIMEDOUT' OR a.statusStr = 'SUCCEEDED' OR a.statusStr = 'KILLED' OR a.statusStr = 'FAILED')"), 113 114 @NamedQuery(name = "GET_COORD_ACTIONS_PENDING_FALSE_STATUS_COUNT", query = "select count(a) from CoordinatorActionBean a where a.jobId = :jobId AND a.pending = 0 AND a.statusStr = :status"), 115 116 @NamedQuery(name = "GET_ACTIONS_FOR_COORD_JOB", query = "select count(a) from CoordinatorActionBean a where a.jobId = :jobId"), 117 // Query to retrieve Coordinator actions sorted by nominal time 118 @NamedQuery(name = "GET_ACTIONS_FOR_COORD_JOB_ORDER_BY_NOMINAL_TIME", query = "select a.id, a.actionNumber, a.consoleUrl, a.errorCode, a.errorMessage, a.externalId, a.externalStatus, a.jobId, a.trackerUri, a.createdTimestamp, a.nominalTimestamp, a.statusStr, a.lastModifiedTimestamp, a.missingDependencies, a.pushMissingDependencies, a.timeOut from CoordinatorActionBean a where a.jobId = :jobId order by a.nominalTimestamp"), 119 // Query to maintain backward compatibility for coord job info command 120 @NamedQuery(name = "GET_ALL_COLS_FOR_ACTIONS_FOR_COORD_JOB_ORDER_BY_NOMINAL_TIME", query = "select OBJECT(a) from CoordinatorActionBean a where a.jobId = :jobId order by a.nominalTimestamp"), 121 // Query to retrieve action id, action status, pending status and external Id of not completed Coordinator actions 122 @NamedQuery(name = "GET_COORD_ACTIONS_NOT_COMPLETED", query = "select a.id, a.statusStr, a.pending, a.externalId, a.pushMissingDependencies, a.nominalTimestamp, a.createdTimestamp, a.jobId from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr <> 'FAILED' AND a.statusStr <> 'TIMEDOUT' AND a.statusStr <> 'SUCCEEDED' AND a.statusStr <> 'KILLED' AND a.statusStr <> 'IGNORED'"), 123 124 // Query to retrieve action id, action status, pending status and external Id of running Coordinator actions 125 @NamedQuery(name = "GET_COORD_ACTIONS_RUNNING", query = "select a.id, a.statusStr, a.pending, a.externalId, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId and a.statusStr = 'RUNNING'"), 126 127 // Query to retrieve action id, action status, pending status and external Id of suspended Coordinator actions 128 @NamedQuery(name = "GET_COORD_ACTIONS_SUSPENDED", query = "select a.id, a.statusStr, a.pending, a.externalId, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId and a.statusStr = 'SUSPENDED'"), 129 130 // Query to retrieve count of Coordinator actions which are pending 131 @NamedQuery(name = "GET_COORD_ACTIONS_PENDING_COUNT", query = "select count(a) from CoordinatorActionBean a where a.jobId = :jobId AND a.pending > 0"), 132 133 // Query to retrieve status of Coordinator actions 134 @NamedQuery(name = "GET_COORD_ACTIONS_STATUS_UNIGNORED", query = "select a.statusStr, a.pending from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr <> 'IGNORED'"), 135 136 // Query to retrieve status of Coordinator actions 137 @NamedQuery(name = "GET_COORD_ACTION_STATUS", query = "select a.statusStr from CoordinatorActionBean a where a.id = :id"), 138 139 @NamedQuery(name = "GET_COORD_ACTION_FOR_COORD_JOB_BY_ACTION_NUMBER", query = "select a.id from CoordinatorActionBean a where a.jobId = :jobId AND a.actionNumber = :actionNumber"), 140 141 @NamedQuery(name = "GET_COORD_ACTIONS_BY_LAST_MODIFIED_TIME", query = "select a.jobId from CoordinatorActionBean a where a.lastModifiedTimestamp >= :lastModifiedTime"), 142 143 //Used by coordinator store only 144 @NamedQuery(name = "GET_RUNNING_ACTIONS_FOR_COORD_JOB", query = "select OBJECT(a) from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr = 'RUNNING'"), 145 146 @NamedQuery(name = "GET_RUNNING_ACTIONS_OLDER_THAN", query = "select a.id from CoordinatorActionBean a where a.statusStr = 'RUNNING' AND a.lastModifiedTimestamp <= :lastModifiedTime"), 147 148 @NamedQuery(name = "GET_COORD_ACTIONS_WAITING_SUBMITTED_OLDER_THAN", query = "select a.id, a.jobId, a.statusStr, a.externalId, a.pushMissingDependencies from CoordinatorActionBean a where (a.statusStr = 'WAITING' OR a.statusStr = 'SUBMITTED') AND a.lastModifiedTimestamp <= :lastModifiedTime"), 149 150 @NamedQuery(name = "GET_COORD_ACTIONS_FOR_RECOVERY_OLDER_THAN", query = "select a.id, a.jobId, a.statusStr, a.externalId, a.pending from CoordinatorActionBean a where a.pending > 0 AND (a.statusStr = 'SUSPENDED' OR a.statusStr = 'KILLED' OR a.statusStr = 'RUNNING') AND a.lastModifiedTimestamp <= :lastModifiedTime"), 151 // Select query used by rerun, requires almost all columns so select * is used 152 @NamedQuery(name = "GET_ACTIONS_FOR_DATES", query = "select OBJECT(a) from CoordinatorActionBean a where a.jobId = :jobId AND (a.statusStr = 'TIMEDOUT' OR a.statusStr = 'SUCCEEDED' OR a.statusStr = 'KILLED' OR a.statusStr = 'FAILED' OR a.statusStr = 'IGNORED') AND a.nominalTimestamp >= :startTime AND a.nominalTimestamp <= :endTime"), 153 // Select query used by log 154 @NamedQuery(name = "GET_ACTION_IDS_FOR_DATES", query = "select a.id from CoordinatorActionBean a where a.jobId = :jobId AND (a.statusStr = 'TIMEDOUT' OR a.statusStr = 'SUCCEEDED' OR a.statusStr = 'KILLED' OR a.statusStr = 'FAILED') AND a.nominalTimestamp >= :startTime AND a.nominalTimestamp <= :endTime"), 155 // Select query used by rerun, requires almost all columns so select * is used 156 @NamedQuery(name = "GET_ACTION_FOR_NOMINALTIME", query = "select OBJECT(a) from CoordinatorActionBean a where a.jobId = :jobId AND a.nominalTimestamp = :nominalTime"), 157 158 @NamedQuery(name = "GET_ACTIONS_BY_DATES_FOR_KILL", query = "select a.id, a.jobId, a.statusStr, a.externalId, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId AND (a.statusStr <> 'FAILED' AND a.statusStr <> 'KILLED' AND a.statusStr <> 'SUCCEEDED' AND a.statusStr <> 'TIMEDOUT') AND a.nominalTimestamp >= :startTime AND a.nominalTimestamp <= :endTime"), 159 160 @NamedQuery(name = "GET_COORD_ACTIONS_COUNT", query = "select count(w) from CoordinatorActionBean w"), 161 162 @NamedQuery(name = "GET_COORD_ACTIONS_COUNT_RUNNING_FOR_RANGE", query = "select count(w) from CoordinatorActionBean w where w.statusStr = 'RUNNING' and w.jobId= :jobId and w.id >= :startAction AND w.id <= :endAction"), 163 164 @NamedQuery(name = "GET_COORD_ACTIONS_MAX_MODIFIED_DATE_FOR_RANGE", query = "select max(w.lastModifiedTimestamp) from CoordinatorActionBean w where w.jobId= :jobId and w.id >= :startAction AND w.id <= :endAction"), 165 166 @NamedQuery(name = "GET_READY_ACTIONS_GROUP_BY_JOBID", query = "select a.jobId, min(a.lastModifiedTimestamp) from CoordinatorActionBean a where a.statusStr = 'READY' group by a.jobId having min(a.lastModifiedTimestamp) < :lastModifiedTime")}) 167 168@Table(name = "COORD_ACTIONS") 169public class CoordinatorActionBean implements 170 Writable,CoordinatorAction,JsonBean { 171 172 @Id 173 private String id; 174 175 @Basic 176 @Index 177 @Column(name = "job_id") 178 private String jobId; 179 180 @Basic 181 @Index 182 @Column(name = "status") 183 private String statusStr = CoordinatorAction.Status.WAITING.toString(); 184 185 @Basic 186 @Index 187 @Column(name = "nominal_time") 188 private java.sql.Timestamp nominalTimestamp = null; 189 190 @Basic 191 @Index 192 @Column(name = "last_modified_time") 193 private java.sql.Timestamp lastModifiedTimestamp = null; 194 195 @Basic 196 @Index 197 @Column(name = "created_time") 198 private java.sql.Timestamp createdTimestamp = null; 199 200 @Basic 201 @Index 202 @Column(name = "rerun_time") 203 private java.sql.Timestamp rerunTimestamp = null; 204 205 @Basic 206 @Index 207 @Column(name = "external_id") 208 private String externalId; 209 210 @Basic 211 @Column(name = "sla_xml") 212 @Lob 213 @Strategy("org.apache.oozie.executor.jpa.StringBlobValueHandler") 214 private StringBlob slaXml = null; 215 216 @Basic 217 @Column(name = "pending") 218 private int pending = 0; 219 220 @Basic 221 @Column(name = "job_type") 222 private String type; 223 224 @Basic 225 @Column(name = "action_number") 226 private int actionNumber; 227 228 @Basic 229 @Column(name = "created_conf") 230 @Lob 231 @Strategy("org.apache.oozie.executor.jpa.StringBlobValueHandler") 232 private StringBlob createdConf; 233 234 @Basic 235 @Column(name = "time_out") 236 private int timeOut = 0; 237 238 @Basic 239 @Column(name = "run_conf") 240 @Lob 241 @Strategy("org.apache.oozie.executor.jpa.StringBlobValueHandler") 242 private StringBlob runConf; 243 244 @Basic 245 @Column(name = "action_xml") 246 @Lob 247 @Strategy("org.apache.oozie.executor.jpa.StringBlobValueHandler") 248 private StringBlob actionXml; 249 250 @Basic 251 @Column(name = "missing_dependencies") 252 @Lob 253 @Strategy("org.apache.oozie.executor.jpa.StringBlobValueHandler") 254 private StringBlob missingDependencies; 255 256 @Basic 257 @Column(name = "push_missing_dependencies") 258 @Lob 259 @Strategy("org.apache.oozie.executor.jpa.StringBlobValueHandler") 260 private StringBlob pushMissingDependencies; 261 262 @Basic 263 @Column(name = "external_status") 264 private String externalStatus; 265 266 @Basic 267 @Column(name = "tracker_uri") 268 private String trackerUri; 269 270 @Basic 271 @Column(name = "console_url") 272 private String consoleUrl; 273 274 @Basic 275 @Column(name = "error_code") 276 private String errorCode; 277 278 @Basic 279 @Column(name = "error_message") 280 private String errorMessage; 281 282 @SuppressWarnings("unchecked") 283 public JSONObject toJSONObject() { 284 return toJSONObject("GMT"); 285 } 286 287 public CoordinatorActionBean() { 288 } 289 290 /** 291 * Serialize the coordinator bean to a data output. 292 * 293 * @param dataOutput data output. 294 * @throws IOException thrown if the coordinator bean could not be 295 * serialized. 296 */ 297 @Override 298 public void write(DataOutput dataOutput) throws IOException { 299 WritableUtils.writeStr(dataOutput, getJobId()); 300 WritableUtils.writeStr(dataOutput, getType()); 301 WritableUtils.writeStr(dataOutput, getId()); 302 WritableUtils.writeStr(dataOutput, getCreatedConf()); 303 WritableUtils.writeStr(dataOutput, getStatus().toString()); 304 dataOutput.writeInt(getActionNumber()); 305 WritableUtils.writeStr(dataOutput, getRunConf()); 306 WritableUtils.writeStr(dataOutput, getExternalStatus()); 307 WritableUtils.writeStr(dataOutput, getTrackerUri()); 308 WritableUtils.writeStr(dataOutput, getConsoleUrl()); 309 WritableUtils.writeStr(dataOutput, getErrorCode()); 310 WritableUtils.writeStr(dataOutput, getErrorMessage()); 311 dataOutput.writeLong((getCreatedTime() != null) ? getCreatedTime().getTime() : -1); 312 dataOutput.writeLong((getLastModifiedTime() != null) ? getLastModifiedTime().getTime() : -1); 313 } 314 315 /** 316 * Deserialize a coordinator bean from a data input. 317 * 318 * @param dataInput data input. 319 * @throws IOException thrown if the workflow bean could not be 320 * deserialized. 321 */ 322 @Override 323 public void readFields(DataInput dataInput) throws IOException { 324 setJobId(WritableUtils.readStr(dataInput)); 325 setType(WritableUtils.readStr(dataInput)); 326 setId(WritableUtils.readStr(dataInput)); 327 setCreatedConf(WritableUtils.readStr(dataInput)); 328 setStatus(CoordinatorAction.Status.valueOf(WritableUtils.readStr(dataInput))); 329 setActionNumber(dataInput.readInt()); 330 setRunConf(WritableUtils.readStr(dataInput)); 331 setExternalStatus(WritableUtils.readStr(dataInput)); 332 setTrackerUri(WritableUtils.readStr(dataInput)); 333 setConsoleUrl(WritableUtils.readStr(dataInput)); 334 setErrorCode(WritableUtils.readStr(dataInput)); 335 setErrorMessage(WritableUtils.readStr(dataInput)); 336 long d = dataInput.readLong(); 337 if (d != -1) { 338 setCreatedTime(new Date(d)); 339 } 340 d = dataInput.readLong(); 341 if (d != -1) { 342 setLastModifiedTime(new Date(d)); 343 } 344 } 345 346 @Override 347 public String getJobId() { 348 return this.jobId; 349 } 350 351 public void setJobId(String id) { 352 this.jobId = id; 353 } 354 355 @Override 356 public Status getStatus() { 357 return Status.valueOf(statusStr); 358 } 359 360 /** 361 * Return the status in string 362 * @return 363 */ 364 public String getStatusStr() { 365 return statusStr; 366 } 367 368 public void setStatus(Status status) { 369 this.statusStr = status.toString(); 370 } 371 372 public void setStatusStr(String statusStr) { 373 this.statusStr = statusStr; 374 } 375 376 public void setCreatedTime(Date createdTime) { 377 this.createdTimestamp = DateUtils.convertDateToTimestamp(createdTime); 378 } 379 380 public void setRerunTime(Date rerunTime) { 381 this.rerunTimestamp = DateUtils.convertDateToTimestamp(rerunTime); 382 } 383 384 public void setNominalTime(Date nominalTime) { 385 this.nominalTimestamp = DateUtils.convertDateToTimestamp(nominalTime); 386 } 387 388 public void setLastModifiedTime(Date lastModifiedTime) { 389 this.lastModifiedTimestamp = DateUtils.convertDateToTimestamp(lastModifiedTime); 390 } 391 392 public Date getCreatedTime() { 393 return DateUtils.toDate(createdTimestamp); 394 } 395 396 public Timestamp getCreatedTimestamp() { 397 return createdTimestamp; 398 } 399 400 public Date getRerunTime() { 401 return DateUtils.toDate(rerunTimestamp); 402 } 403 404 public Timestamp getRerunTimestamp() { 405 return rerunTimestamp; 406 } 407 408 @Override 409 public Date getLastModifiedTime() { 410 return DateUtils.toDate(lastModifiedTimestamp); 411 } 412 413 public Timestamp getLastModifiedTimestamp() { 414 return lastModifiedTimestamp; 415 } 416 417 @Override 418 public Date getNominalTime() { 419 return DateUtils.toDate(nominalTimestamp); 420 } 421 422 public Timestamp getNominalTimestamp() { 423 return nominalTimestamp; 424 } 425 426 @Override 427 public String getExternalId() { 428 return externalId; 429 } 430 431 public void setExternalId(String externalId) { 432 this.externalId = externalId; 433 } 434 435 public StringBlob getSlaXmlBlob() { 436 return slaXml; 437 } 438 439 public void setSlaXmlBlob(StringBlob slaXml) { 440 this.slaXml = slaXml; 441 } 442 443 public String getSlaXml() { 444 return slaXml == null ? null : slaXml.getString(); 445 } 446 447 public void setSlaXml(String slaXml) { 448 if (this.slaXml == null) { 449 this.slaXml = new StringBlob(slaXml); 450 } 451 else { 452 this.slaXml.setString(slaXml); 453 } 454 } 455 456 /** 457 * @return true if in terminal status 458 */ 459 public boolean isTerminalStatus() { 460 boolean isTerminal = true; 461 switch (getStatus()) { 462 case WAITING: 463 case READY: 464 case SUBMITTED: 465 case RUNNING: 466 case SUSPENDED: 467 isTerminal = false; 468 break; 469 default: 470 isTerminal = true; 471 break; 472 } 473 return isTerminal; 474 } 475 476 /** 477 * Return if the action is complete with failure. 478 * 479 * @return if the action is complete with failure. 480 */ 481 public boolean isTerminalWithFailure() { 482 boolean result = false; 483 switch (getStatus()) { 484 case FAILED: 485 case KILLED: 486 case TIMEDOUT: 487 result = true; 488 } 489 return result; 490 } 491 492 /** 493 * Set some actions are in progress for particular coordinator action. 494 * 495 * @param pending set pending to true 496 */ 497 public void setPending(int pending) { 498 this.pending = pending; 499 } 500 501 /** 502 * increment pending and return it 503 * 504 * @return pending 505 */ 506 public int incrementAndGetPending() { 507 this.pending++; 508 return pending; 509 } 510 511 /** 512 * decrement pending and return it 513 * 514 * @return pending 515 */ 516 public int decrementAndGetPending() { 517 this.pending = Math.max(this.pending - 1, 0); 518 return pending; 519 } 520 521 /** 522 * Get some actions are in progress for particular bundle action. 523 * 524 * @return pending 525 */ 526 public int getPending() { 527 return this.pending; 528 } 529 530 /** 531 * Return if the action is pending. 532 * 533 * @return if the action is pending. 534 */ 535 public boolean isPending() { 536 return pending > 0 ? true : false; 537 } 538 539 @Override 540 public String getId() { 541 return id; 542 } 543 544 public void setId(String id) { 545 this.id = id; 546 } 547 548 public String getType() { 549 return type; 550 } 551 552 public void setType(String type) { 553 this.type = type; 554 } 555 556 public void setActionNumber(int actionNumber) { 557 this.actionNumber = actionNumber; 558 } 559 560 @Override 561 public int getActionNumber() { 562 return actionNumber; 563 } 564 565 @Override 566 public String getCreatedConf() { 567 return createdConf == null ? null : createdConf.getString(); 568 } 569 570 public void setCreatedConf(String createdConf) { 571 if (this.createdConf == null) { 572 this.createdConf = new StringBlob(createdConf); 573 } 574 else { 575 this.createdConf.setString(createdConf); 576 } 577 } 578 579 public void setCreatedConfBlob(StringBlob createdConf) { 580 this.createdConf = createdConf; 581 } 582 583 public StringBlob getCreatedConfBlob() { 584 return createdConf; 585 } 586 587 public void setRunConf(String runConf) { 588 if (this.runConf == null) { 589 this.runConf = new StringBlob(runConf); 590 } 591 else { 592 this.runConf.setString(runConf); 593 } 594 } 595 596 @Override 597 public String getRunConf() { 598 return runConf == null ? null : runConf.getString(); 599 } 600 601 public void setRunConfBlob(StringBlob runConf) { 602 this.runConf = runConf; 603 } 604 605 public StringBlob getRunConfBlob() { 606 return runConf; 607 } 608 609 610 public void setMissingDependencies(String missingDependencies) { 611 if (this.missingDependencies == null) { 612 this.missingDependencies = new StringBlob(missingDependencies); 613 } 614 else { 615 this.missingDependencies.setString(missingDependencies); 616 } 617 } 618 619 @Override 620 public String getMissingDependencies() { 621 return missingDependencies == null ? null : missingDependencies.getString(); 622 } 623 624 public void setMissingDependenciesBlob(StringBlob missingDependencies) { 625 this.missingDependencies = missingDependencies; 626 } 627 628 public StringBlob getMissingDependenciesBlob() { 629 return missingDependencies; 630 } 631 632 @Override 633 public String getPushMissingDependencies() { 634 return pushMissingDependencies == null ? null : pushMissingDependencies.getString(); 635 } 636 637 public void setPushMissingDependencies(String pushMissingDependencies) { 638 if (this.pushMissingDependencies == null) { 639 this.pushMissingDependencies = new StringBlob(pushMissingDependencies); 640 } 641 else { 642 this.pushMissingDependencies.setString(pushMissingDependencies); 643 } 644 } 645 646 public void setPushMissingDependenciesBlob(StringBlob pushMissingDependencies) { 647 this.pushMissingDependencies = pushMissingDependencies; 648 } 649 650 public StringBlob getPushMissingDependenciesBlob() { 651 return pushMissingDependencies; 652 } 653 654 public String getExternalStatus() { 655 return externalStatus; 656 } 657 658 public void setExternalStatus(String externalStatus) { 659 this.externalStatus = externalStatus; 660 } 661 662 @Override 663 public String getTrackerUri() { 664 return trackerUri; 665 } 666 667 public void setTrackerUri(String trackerUri) { 668 this.trackerUri = trackerUri; 669 } 670 671 @Override 672 public String getConsoleUrl() { 673 return consoleUrl; 674 } 675 676 public void setConsoleUrl(String consoleUrl) { 677 this.consoleUrl = consoleUrl; 678 } 679 680 @Override 681 public String getErrorCode() { 682 return errorCode; 683 } 684 685 @Override 686 public String getErrorMessage() { 687 return errorMessage; 688 } 689 690 public void setErrorInfo(String errorCode, String errorMessage) { 691 this.errorCode = errorCode; 692 this.errorMessage = errorMessage; 693 } 694 695 public String getActionXml() { 696 return actionXml == null ? null : actionXml.getString(); 697 } 698 699 public void setActionXml(String actionXml) { 700 if (this.actionXml == null) { 701 this.actionXml = new StringBlob(actionXml); 702 } 703 else { 704 this.actionXml.setString(actionXml); 705 } 706 } 707 708 public void setActionXmlBlob(StringBlob actionXml) { 709 this.actionXml = actionXml; 710 } 711 712 public StringBlob getActionXmlBlob() { 713 return actionXml; 714 } 715 716 @Override 717 public String toString() { 718 return MessageFormat.format("CoordinatorAction name[{0}] status[{1}]", 719 getId(), getStatus()); 720 } 721 722 public int getTimeOut() { 723 return timeOut; 724 } 725 726 public void setTimeOut(int timeOut) { 727 this.timeOut = timeOut; 728 } 729 730 731 public void setErrorCode(String errorCode) { 732 this.errorCode = errorCode; 733 } 734 735 public void setErrorMessage(String errorMessage) { 736 this.errorMessage = errorMessage; 737 } 738 739 @SuppressWarnings("unchecked") 740 public JSONObject toJSONObject(String timeZoneId) { 741 JSONObject json = new JSONObject(); 742 json.put(JsonTags.COORDINATOR_ACTION_ID, id); 743 json.put(JsonTags.COORDINATOR_JOB_ID, jobId); 744 json.put(JsonTags.COORDINATOR_ACTION_TYPE, type); 745 json.put(JsonTags.COORDINATOR_ACTION_NUMBER, actionNumber); 746 json.put(JsonTags.COORDINATOR_ACTION_CREATED_CONF, getCreatedConf()); 747 json.put(JsonTags.COORDINATOR_ACTION_CREATED_TIME, JsonUtils 748 .formatDateRfc822(getCreatedTime(), timeZoneId)); 749 json.put(JsonTags.COORDINATOR_ACTION_NOMINAL_TIME, JsonUtils 750 .formatDateRfc822(getNominalTime(), timeZoneId)); 751 json.put(JsonTags.COORDINATOR_ACTION_EXTERNALID, externalId); 752 // json.put(JsonTags.COORDINATOR_ACTION_START_TIME, JsonUtils 753 // .formatDateRfc822(startTime), timeZoneId); 754 json.put(JsonTags.COORDINATOR_ACTION_STATUS, statusStr); 755 json.put(JsonTags.COORDINATOR_ACTION_RUNTIME_CONF, getRunConf()); 756 json.put(JsonTags.COORDINATOR_ACTION_LAST_MODIFIED_TIME, JsonUtils 757 .formatDateRfc822(getLastModifiedTime(), timeZoneId)); 758 // json.put(JsonTags.COORDINATOR_ACTION_START_TIME, JsonUtils 759 // .formatDateRfc822(startTime), timeZoneId); 760 // json.put(JsonTags.COORDINATOR_ACTION_END_TIME, JsonUtils 761 // .formatDateRfc822(endTime), timeZoneId); 762 json.put(JsonTags.COORDINATOR_ACTION_MISSING_DEPS, getMissingDependencies()); 763 json.put(JsonTags.COORDINATOR_ACTION_PUSH_MISSING_DEPS, getPushMissingDependencies()); 764 json.put(JsonTags.COORDINATOR_ACTION_EXTERNAL_STATUS, externalStatus); 765 json.put(JsonTags.COORDINATOR_ACTION_TRACKER_URI, trackerUri); 766 json.put(JsonTags.COORDINATOR_ACTION_CONSOLE_URL, consoleUrl); 767 json.put(JsonTags.COORDINATOR_ACTION_ERROR_CODE, errorCode); 768 json.put(JsonTags.COORDINATOR_ACTION_ERROR_MESSAGE, errorMessage); 769 json.put(JsonTags.TO_STRING, toString()); 770 return json; 771 } 772 773 /** 774 * Convert a nodes list into a JSONArray. 775 * 776 * @param actions nodes list. 777 * @param timeZoneId time zone to use for dates in the JSON array. 778 * @return the corresponding JSON array. 779 */ 780 @SuppressWarnings("unchecked") 781 public static JSONArray toJSONArray(List<CoordinatorActionBean> actions, String timeZoneId) { 782 JSONArray array = new JSONArray(); 783 for (CoordinatorActionBean action : actions) { 784 array.add(action.toJSONObject(timeZoneId)); 785 } 786 return array; 787 } 788 789 @Override 790 public int hashCode() { 791 final int prime = 31; 792 int result = 1; 793 result = prime * result + ((id == null) ? 0 : id.hashCode()); 794 return result; 795 } 796 797 @Override 798 public boolean equals(Object obj) { 799 if (this == obj) { 800 return true; 801 } 802 if (obj == null) { 803 return false; 804 } 805 if (getClass() != obj.getClass()) { 806 return false; 807 } 808 CoordinatorActionBean other = (CoordinatorActionBean) obj; 809 if (id == null) { 810 if (other.id != null) { 811 return false; 812 } 813 } 814 else if (!id.equals(other.id)) { 815 return false; 816 } 817 return true; 818 } 819 820 821}