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.command.coord; 020 021import java.io.IOException; 022import java.io.StringReader; 023import java.net.URI; 024import java.net.URISyntaxException; 025import java.text.ParseException; 026import java.util.Calendar; 027import java.util.Date; 028import java.util.List; 029 030import org.apache.hadoop.conf.Configuration; 031import org.apache.hadoop.security.AccessControlException; 032import org.apache.oozie.CoordinatorActionBean; 033import org.apache.oozie.CoordinatorJobBean; 034import org.apache.oozie.ErrorCode; 035import org.apache.oozie.client.CoordinatorAction; 036import org.apache.oozie.client.Job; 037import org.apache.oozie.client.OozieClient; 038import org.apache.oozie.command.CommandException; 039import org.apache.oozie.command.PreconditionException; 040import org.apache.oozie.coord.CoordELEvaluator; 041import org.apache.oozie.coord.CoordELFunctions; 042import org.apache.oozie.coord.TimeUnit; 043import org.apache.oozie.dependency.URIHandler; 044import org.apache.oozie.dependency.URIHandlerException; 045import org.apache.oozie.executor.jpa.CoordActionGetForInputCheckJPAExecutor; 046import org.apache.oozie.executor.jpa.CoordActionQueryExecutor; 047import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery; 048import org.apache.oozie.executor.jpa.CoordJobQueryExecutor; 049import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery; 050import org.apache.oozie.executor.jpa.JPAExecutorException; 051import org.apache.oozie.service.CallableQueueService; 052import org.apache.oozie.service.ConfigurationService; 053import org.apache.oozie.service.EventHandlerService; 054import org.apache.oozie.service.JPAService; 055import org.apache.oozie.service.Service; 056import org.apache.oozie.service.Services; 057import org.apache.oozie.service.URIHandlerService; 058import org.apache.oozie.util.DateUtils; 059import org.apache.oozie.util.ELEvaluator; 060import org.apache.oozie.util.LogUtils; 061import org.apache.oozie.util.ParamChecker; 062import org.apache.oozie.util.StatusUtils; 063import org.apache.oozie.util.XConfiguration; 064import org.apache.oozie.util.XLog; 065import org.apache.oozie.util.XmlUtils; 066import org.jdom.Element; 067 068/** 069 * The command to check if an action's data input paths exist in the file system. 070 */ 071public class CoordActionInputCheckXCommand extends CoordinatorXCommand<Void> { 072 073 public static final String COORD_EXECUTION_NONE_TOLERANCE = "oozie.coord.execution.none.tolerance"; 074 075 private final String actionId; 076 /** 077 * Property name of command re-queue interval for coordinator action input check in 078 * milliseconds. 079 */ 080 public static final String CONF_COORD_INPUT_CHECK_REQUEUE_INTERVAL = Service.CONF_PREFIX 081 + "coord.input.check.requeue.interval"; 082 private CoordinatorActionBean coordAction = null; 083 private CoordinatorJobBean coordJob = null; 084 private JPAService jpaService = null; 085 private String jobId = null; 086 087 public CoordActionInputCheckXCommand(String actionId, String jobId) { 088 super("coord_action_input", "coord_action_input", 1); 089 this.actionId = ParamChecker.notEmpty(actionId, "actionId"); 090 this.jobId = jobId; 091 } 092 093 @Override 094 protected void setLogInfo() { 095 LogUtils.setLogInfo(actionId); 096 } 097 098 @Override 099 protected Void execute() throws CommandException { 100 LOG.debug("[" + actionId + "]::ActionInputCheck:: Action is in WAITING state."); 101 102 // this action should only get processed if current time > nominal time; 103 // otherwise, requeue this action for delay execution; 104 Date nominalTime = coordAction.getNominalTime(); 105 Date currentTime = new Date(); 106 if (nominalTime.compareTo(currentTime) > 0) { 107 queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()), Math.max((nominalTime.getTime() - currentTime 108 .getTime()), getCoordInputCheckRequeueInterval())); 109 updateCoordAction(coordAction, false); 110 LOG.info("[" + actionId 111 + "]::ActionInputCheck:: nominal Time is newer than current time, so requeue and wait. Current=" 112 + DateUtils.formatDateOozieTZ(currentTime) + ", nominal=" + DateUtils.formatDateOozieTZ(nominalTime)); 113 114 return null; 115 } 116 117 StringBuilder actionXml = new StringBuilder(coordAction.getActionXml()); 118 boolean isChangeInDependency = false; 119 try { 120 Configuration actionConf = new XConfiguration(new StringReader(coordAction.getRunConf())); 121 Date now = new Date(); 122 if (coordJob.getExecutionOrder().equals(CoordinatorJobBean.Execution.LAST_ONLY)) { 123 Date nextNominalTime = CoordCommandUtils.computeNextNominalTime(coordJob, coordAction); 124 if (nextNominalTime != null) { 125 // If the current time is after the next action's nominal time, then we've passed the window where this action 126 // should be started; so set it to SKIPPED 127 if (now.after(nextNominalTime)) { 128 LOG.info("LAST_ONLY execution: Preparing to skip action [{0}] because the current time [{1}] is later than " 129 + "the nominal time [{2}] of the next action]", coordAction.getId(), 130 DateUtils.formatDateOozieTZ(now), DateUtils.formatDateOozieTZ(nextNominalTime)); 131 queue(new CoordActionSkipXCommand(coordAction, coordJob.getUser(), coordJob.getAppName())); 132 return null; 133 } else { 134 LOG.debug("LAST_ONLY execution: Not skipping action [{0}] because the current time [{1}] is earlier than " 135 + "the nominal time [{2}] of the next action]", coordAction.getId(), 136 DateUtils.formatDateOozieTZ(now), DateUtils.formatDateOozieTZ(nextNominalTime)); 137 } 138 } 139 } 140 else if (coordJob.getExecutionOrder().equals(CoordinatorJobBean.Execution.NONE)) { 141 // If the current time is after the nominal time of this action plus some tolerance, 142 // then we've passed the window where this action should be started; so set it to SKIPPED 143 Calendar cal = Calendar.getInstance(DateUtils.getTimeZone(coordJob.getTimeZone())); 144 cal.setTime(nominalTime); 145 int tolerance = ConfigurationService.getInt(COORD_EXECUTION_NONE_TOLERANCE); 146 cal.add(Calendar.MINUTE, tolerance); 147 if (now.after(cal.getTime())) { 148 LOG.info("NONE execution: Preparing to skip action [{0}] because the current time [{1}] is more than [{2}]" 149 + " minutes later than the nominal time [{3}] of the current action]", coordAction.getId(), 150 DateUtils.formatDateOozieTZ(now), tolerance, DateUtils.formatDateOozieTZ(nominalTime)); 151 queue(new CoordActionSkipXCommand(coordAction, coordJob.getUser(), coordJob.getAppName())); 152 return null; 153 } else { 154 LOG.debug("NONE execution: Not skipping action [{0}] because the current time [{1}] is earlier than [{2}]" 155 + " minutes later than the nominal time [{3}] of the current action]", coordAction.getId(), 156 DateUtils.formatDateOozieTZ(now), tolerance, DateUtils.formatDateOozieTZ(coordAction.getNominalTime())); 157 } 158 } 159 160 StringBuilder existList = new StringBuilder(); 161 StringBuilder nonExistList = new StringBuilder(); 162 StringBuilder nonResolvedList = new StringBuilder(); 163 String firstMissingDependency = ""; 164 String missingDeps = coordAction.getMissingDependencies(); 165 CoordCommandUtils.getResolvedList(missingDeps, nonExistList, nonResolvedList); 166 167 // For clarity regarding which is the missing dependency in synchronous order 168 // instead of printing entire list, some of which, may be available 169 if(nonExistList.length() > 0) { 170 firstMissingDependency = nonExistList.toString().split(CoordELFunctions.INSTANCE_SEPARATOR)[0]; 171 } 172 LOG.info("[" + actionId + "]::CoordActionInputCheck:: Missing deps:" + firstMissingDependency + " " 173 + nonResolvedList.toString()); 174 // Updating the list of data dependencies that are available and those that are yet not 175 boolean status = checkInput(actionXml, existList, nonExistList, actionConf); 176 String pushDeps = coordAction.getPushMissingDependencies(); 177 // Resolve latest/future only when all current missingDependencies and 178 // pushMissingDependencies are met 179 if (status && nonResolvedList.length() > 0) { 180 status = (pushDeps == null || pushDeps.length() == 0) ? checkUnResolvedInput(actionXml, actionConf) 181 : false; 182 } 183 coordAction.setLastModifiedTime(currentTime); 184 coordAction.setActionXml(actionXml.toString()); 185 if (nonResolvedList.length() > 0 && status == false) { 186 nonExistList.append(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR).append(nonResolvedList); 187 } 188 String nonExistListStr = nonExistList.toString(); 189 if (!nonExistListStr.equals(missingDeps) || missingDeps.isEmpty()) { 190 // missingDeps null or empty means action should become READY 191 isChangeInDependency = true; 192 coordAction.setMissingDependencies(nonExistListStr); 193 } 194 if (status && (pushDeps == null || pushDeps.length() == 0)) { 195 String newActionXml = resolveCoordConfiguration(actionXml, actionConf, actionId); 196 actionXml.replace(0, actionXml.length(), newActionXml); 197 coordAction.setActionXml(actionXml.toString()); 198 coordAction.setStatus(CoordinatorAction.Status.READY); 199 updateCoordAction(coordAction, true); 200 new CoordActionReadyXCommand(coordAction.getJobId()).call(getEntityKey()); 201 } 202 else if (!isTimeout(currentTime)) { 203 if (status == false) { 204 queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()), 205 getCoordInputCheckRequeueInterval()); 206 } 207 updateCoordAction(coordAction, isChangeInDependency); 208 } 209 else { 210 if (!nonExistListStr.isEmpty() && pushDeps == null || pushDeps.length() == 0) { 211 queue(new CoordActionTimeOutXCommand(coordAction, coordJob.getUser(), coordJob.getAppName())); 212 } 213 else { 214 // Let CoordPushDependencyCheckXCommand queue the timeout 215 queue(new CoordPushDependencyCheckXCommand(coordAction.getId())); 216 } 217 updateCoordAction(coordAction, isChangeInDependency); 218 } 219 } 220 catch (AccessControlException e) { 221 LOG.error("Permission error in ActionInputCheck", e); 222 if (isTimeout(currentTime)) { 223 LOG.debug("Queueing timeout command"); 224 Services.get().get(CallableQueueService.class) 225 .queue(new CoordActionTimeOutXCommand(coordAction, coordJob.getUser(), coordJob.getAppName())); 226 } 227 else { 228 // Requeue InputCheckCommand for permission denied error with longer interval 229 Services.get() 230 .get(CallableQueueService.class) 231 .queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()), 232 2 * getCoordInputCheckRequeueInterval()); 233 } 234 updateCoordAction(coordAction, isChangeInDependency); 235 } 236 catch (Exception e) { 237 if (isTimeout(currentTime)) { 238 LOG.debug("Queueing timeout command"); 239 // XCommand.queue() will not work when there is a Exception 240 Services.get().get(CallableQueueService.class) 241 .queue(new CoordActionTimeOutXCommand(coordAction, coordJob.getUser(), coordJob.getAppName())); 242 } 243 updateCoordAction(coordAction, isChangeInDependency); 244 throw new CommandException(ErrorCode.E1021, e.getMessage(), e); 245 } 246 return null; 247 } 248 249 250 static String resolveCoordConfiguration(StringBuilder actionXml, Configuration actionConf, String actionId) throws Exception { 251 Element eAction = XmlUtils.parseXml(actionXml.toString()); 252 ELEvaluator eval = CoordELEvaluator.createDataEvaluator(eAction, actionConf, actionId); 253 materializeDataProperties(eAction, actionConf, eval); 254 return XmlUtils.prettyPrint(eAction).toString(); 255 } 256 257 private boolean isTimeout(Date currentTime) { 258 long waitingTime = (currentTime.getTime() - Math.max(coordAction.getNominalTime().getTime(), coordAction 259 .getCreatedTime().getTime())) 260 / (60 * 1000); 261 int timeOut = coordAction.getTimeOut(); 262 return (timeOut >= 0) && (waitingTime > timeOut); 263 } 264 265 private void updateCoordAction(CoordinatorActionBean coordAction, boolean isChangeInDependency) 266 throws CommandException { 267 coordAction.setLastModifiedTime(new Date()); 268 if (jpaService != null) { 269 try { 270 if (isChangeInDependency) { 271 CoordActionQueryExecutor.getInstance().executeUpdate( 272 CoordActionQuery.UPDATE_COORD_ACTION_FOR_INPUTCHECK, coordAction); 273 if (EventHandlerService.isEnabled() && coordAction.getStatus() != CoordinatorAction.Status.READY) { 274 // since event is not to be generated unless action 275 // RUNNING via StartX 276 generateEvent(coordAction, coordJob.getUser(), coordJob.getAppName(), null); 277 } 278 } 279 else { 280 CoordActionQueryExecutor.getInstance().executeUpdate( 281 CoordActionQuery.UPDATE_COORD_ACTION_FOR_MODIFIED_DATE, coordAction); 282 } 283 } 284 catch (JPAExecutorException jex) { 285 throw new CommandException(ErrorCode.E1021, jex.getMessage(), jex); 286 } 287 } 288 } 289 290 /** 291 * This function reads the value of re-queue interval for coordinator input 292 * check command from the Oozie configuration provided by Configuration 293 * Service. If nothing defined in the configuration, it uses the code 294 * specified default value. 295 * 296 * @return re-queue interval in ms 297 */ 298 public long getCoordInputCheckRequeueInterval() { 299 long requeueInterval = ConfigurationService.getLong(CONF_COORD_INPUT_CHECK_REQUEUE_INTERVAL); 300 return requeueInterval; 301 } 302 303 /** 304 * To check the list of input paths if all of them exist 305 * 306 * @param actionXml action xml 307 * @param existList the list of existed paths 308 * @param nonExistList the list of non existed paths 309 * @param conf action configuration 310 * @return true if all input paths are existed 311 * @throws Exception thrown of unable to check input path 312 */ 313 protected boolean checkInput(StringBuilder actionXml, StringBuilder existList, StringBuilder nonExistList, 314 Configuration conf) throws Exception { 315 Element eAction = XmlUtils.parseXml(actionXml.toString()); 316 return checkResolvedUris(eAction, existList, nonExistList, conf); 317 } 318 319 protected boolean checkUnResolvedInput(StringBuilder actionXml, Configuration conf) throws Exception { 320 Element eAction = XmlUtils.parseXml(actionXml.toString()); 321 LOG.debug("[" + actionId + "]::ActionInputCheck:: Checking Latest/future"); 322 boolean allExist = checkUnresolvedInstances(eAction, conf); 323 if (allExist) { 324 actionXml.replace(0, actionXml.length(), XmlUtils.prettyPrint(eAction).toString()); 325 } 326 return allExist; 327 } 328 329 /** 330 * Materialize data properties defined in <action> tag. it includes dataIn(<DS>) and dataOut(<DS>) it creates a list 331 * of files that will be needed. 332 * 333 * @param eAction action element 334 * @param conf action configuration 335 * @throws Exception thrown if failed to resolve data properties 336 * @update modify 'Action' element with appropriate list of files. 337 */ 338 @SuppressWarnings("unchecked") 339 static void materializeDataProperties(Element eAction, Configuration conf, ELEvaluator eval) throws Exception { 340 Element configElem = eAction.getChild("action", eAction.getNamespace()).getChild("workflow", 341 eAction.getNamespace()).getChild("configuration", eAction.getNamespace()); 342 if (configElem != null) { 343 for (Element propElem : (List<Element>) configElem.getChildren("property", configElem.getNamespace())) { 344 resolveTagContents("value", propElem, eval); 345 } 346 } 347 } 348 349 /** 350 * To resolve property value which contains el functions 351 * 352 * @param tagName tag name 353 * @param elem the child element of "property" element 354 * @param eval el functions evaluator 355 * @throws Exception thrown if unable to resolve tag value 356 */ 357 private static void resolveTagContents(String tagName, Element elem, ELEvaluator eval) throws Exception { 358 if (elem == null) { 359 return; 360 } 361 Element tagElem = elem.getChild(tagName, elem.getNamespace()); 362 if (tagElem != null) { 363 String updated = CoordELFunctions.evalAndWrap(eval, tagElem.getText()); 364 tagElem.removeContent(); 365 tagElem.addContent(updated); 366 } 367 else { 368 XLog.getLog(CoordActionInputCheckXCommand.class).warn(" Value NOT FOUND " + tagName); 369 } 370 } 371 372 /** 373 * Check if any unsolved paths under data output. Resolve the unresolved data input paths. 374 * 375 * @param eAction action element 376 * @param actionConf action configuration 377 * @return true if successful to resolve input and output paths 378 * @throws Exception thrown if failed to resolve data input and output paths 379 */ 380 @SuppressWarnings("unchecked") 381 private boolean checkUnresolvedInstances(Element eAction, Configuration actionConf) throws Exception { 382 String strAction = XmlUtils.prettyPrint(eAction).toString(); 383 Date nominalTime = DateUtils.parseDateOozieTZ(eAction.getAttributeValue("action-nominal-time")); 384 String actualTimeStr = eAction.getAttributeValue("action-actual-time"); 385 Date actualTime = null; 386 if (actualTimeStr == null) { 387 LOG.debug("Unable to get action-actual-time from action xml, this job is submitted " + 388 "from previous version. Assign current date to actual time, action = " + actionId); 389 actualTime = new Date(); 390 } else { 391 actualTime = DateUtils.parseDateOozieTZ(actualTimeStr); 392 } 393 394 StringBuffer resultedXml = new StringBuffer(); 395 396 boolean ret; 397 Element inputList = eAction.getChild("input-events", eAction.getNamespace()); 398 if (inputList != null) { 399 ret = materializeUnresolvedEvent(inputList.getChildren("data-in", eAction.getNamespace()), nominalTime, 400 actualTime, actionConf); 401 if (ret == false) { 402 resultedXml.append(strAction); 403 return false; 404 } 405 } 406 407 // Using latest() or future() in output-event is not intuitive. 408 // We need to make sure, this assumption is correct. 409 Element outputList = eAction.getChild("output-events", eAction.getNamespace()); 410 if (outputList != null) { 411 for (Element dEvent : (List<Element>) outputList.getChildren("data-out", eAction.getNamespace())) { 412 if (dEvent.getChild(CoordCommandUtils.UNRESOLVED_INST_TAG, dEvent.getNamespace()) != null) { 413 throw new CommandException(ErrorCode.E1006, "coord:latest()/future()", 414 " not permitted in output-event "); 415 } 416 } 417 } 418 return true; 419 } 420 421 /** 422 * Resolve the list of data input paths 423 * 424 * @param eDataEvents the list of data input elements 425 * @param nominalTime action nominal time 426 * @param actualTime current time 427 * @param conf action configuration 428 * @return true if all unresolved URIs can be resolved 429 * @throws Exception thrown if failed to resolve data input paths 430 */ 431 @SuppressWarnings("unchecked") 432 private boolean materializeUnresolvedEvent(List<Element> eDataEvents, Date nominalTime, Date actualTime, 433 Configuration conf) throws Exception { 434 for (Element dEvent : eDataEvents) { 435 if (dEvent.getChild(CoordCommandUtils.UNRESOLVED_INST_TAG, dEvent.getNamespace()) == null) { 436 continue; 437 } 438 ELEvaluator eval = CoordELEvaluator.createLazyEvaluator(actualTime, nominalTime, dEvent, conf); 439 String uresolvedInstance = dEvent.getChild(CoordCommandUtils.UNRESOLVED_INST_TAG, dEvent.getNamespace()).getTextTrim(); 440 String unresolvedList[] = uresolvedInstance.split(CoordELFunctions.INSTANCE_SEPARATOR); 441 StringBuffer resolvedTmp = new StringBuffer(); 442 for (int i = 0; i < unresolvedList.length; i++) { 443 String ret = CoordELFunctions.evalAndWrap(eval, unresolvedList[i]); 444 Boolean isResolved = (Boolean) eval.getVariable("is_resolved"); 445 if (isResolved == false) { 446 LOG.info("[" + actionId + "]::Cannot resolve: " + ret); 447 return false; 448 } 449 if (resolvedTmp.length() > 0) { 450 resolvedTmp.append(CoordELFunctions.INSTANCE_SEPARATOR); 451 } 452 resolvedTmp.append((String) eval.getVariable("resolved_path")); 453 } 454 if (resolvedTmp.length() > 0) { 455 if (dEvent.getChild("uris", dEvent.getNamespace()) != null) { 456 resolvedTmp.append(CoordELFunctions.INSTANCE_SEPARATOR).append( 457 dEvent.getChild("uris", dEvent.getNamespace()).getTextTrim()); 458 dEvent.removeChild("uris", dEvent.getNamespace()); 459 } 460 Element uriInstance = new Element("uris", dEvent.getNamespace()); 461 uriInstance.addContent(resolvedTmp.toString()); 462 dEvent.getContent().add(1, uriInstance); 463 } 464 dEvent.removeChild(CoordCommandUtils.UNRESOLVED_INST_TAG, dEvent.getNamespace()); 465 } 466 467 return true; 468 } 469 470 /** 471 * Check all resolved URIs existence 472 * 473 * @param eAction action element 474 * @param existList the list of existed paths 475 * @param nonExistList the list of paths to check existence 476 * @param conf action configuration 477 * @return true if all nonExistList paths exist 478 * @throws IOException thrown if unable to access the path 479 */ 480 private boolean checkResolvedUris(Element eAction, StringBuilder existList, StringBuilder nonExistList, 481 Configuration conf) throws IOException { 482 Element inputList = eAction.getChild("input-events", eAction.getNamespace()); 483 if (inputList != null) { 484 if (nonExistList.length() > 0) { 485 checkListOfPaths(existList, nonExistList, conf); 486 } 487 return nonExistList.length() == 0; 488 } 489 return true; 490 } 491 492 /** 493 * Check a list of non existed paths and add to exist list if it exists 494 * 495 * @param existList the list of existed paths 496 * @param nonExistList the list of paths to check existence 497 * @param conf action configuration 498 * @return true if all nonExistList paths exist 499 * @throws IOException thrown if unable to access the path 500 */ 501 private boolean checkListOfPaths(StringBuilder existList, StringBuilder nonExistList, Configuration conf) 502 throws IOException { 503 504 String[] uriList = nonExistList.toString().split(CoordELFunctions.INSTANCE_SEPARATOR); 505 if (uriList[0] != null) { 506 LOG.info("[" + actionId + "]::ActionInputCheck:: In checkListOfPaths: " + uriList[0] + " is Missing."); 507 } 508 509 nonExistList.delete(0, nonExistList.length()); 510 boolean allExists = true; 511 String existSeparator = "", nonExistSeparator = ""; 512 String user = ParamChecker.notEmpty(conf.get(OozieClient.USER_NAME), OozieClient.USER_NAME); 513 for (int i = 0; i < uriList.length; i++) { 514 if (allExists) { 515 allExists = pathExists(uriList[i], conf, user); 516 LOG.info("[" + actionId + "]::ActionInputCheck:: File:" + uriList[i] + ", Exists? :" + allExists); 517 } 518 if (allExists) { 519 existList.append(existSeparator).append(uriList[i]); 520 existSeparator = CoordELFunctions.INSTANCE_SEPARATOR; 521 } 522 else { 523 nonExistList.append(nonExistSeparator).append(uriList[i]); 524 nonExistSeparator = CoordELFunctions.INSTANCE_SEPARATOR; 525 } 526 } 527 return allExists; 528 } 529 530 /** 531 * Check if given path exists 532 * 533 * @param sPath uri path 534 * @param actionConf action configuration 535 * @return true if path exists 536 * @throws IOException thrown if unable to access the path 537 */ 538 protected boolean pathExists(String sPath, Configuration actionConf, String user) throws IOException { 539 LOG.debug("checking for the file " + sPath); 540 try { 541 URI uri = new URI(sPath); 542 URIHandlerService service = Services.get().get(URIHandlerService.class); 543 URIHandler handler = service.getURIHandler(uri); 544 return handler.exists(uri, actionConf, user); 545 } 546 catch (URIHandlerException e) { 547 coordAction.setErrorCode(e.getErrorCode().toString()); 548 coordAction.setErrorMessage(e.getMessage()); 549 if (e.getCause() != null && e.getCause() instanceof AccessControlException) { 550 throw (AccessControlException) e.getCause(); 551 } 552 else { 553 throw new IOException(e); 554 } 555 } 556 catch (URISyntaxException e) { 557 coordAction.setErrorCode(ErrorCode.E0906.toString()); 558 coordAction.setErrorMessage(e.getMessage()); 559 throw new IOException(e); 560 } 561 } 562 563 /** 564 * The function create a list of URIs separated by "," using the instances time stamp and URI-template 565 * 566 * @param event : <data-in> event 567 * @param instances : List of time stamp seprated by "," 568 * @param unresolvedInstances : list of instance with latest/future function 569 * @return : list of URIs separated by ",". 570 * @throws Exception thrown if failed to create URIs from unresolvedInstances 571 */ 572 @SuppressWarnings("unused") 573 private String createURIs(Element event, String instances, StringBuilder unresolvedInstances) throws Exception { 574 if (instances == null || instances.length() == 0) { 575 return ""; 576 } 577 String[] instanceList = instances.split(CoordELFunctions.INSTANCE_SEPARATOR); 578 StringBuilder uris = new StringBuilder(); 579 580 for (int i = 0; i < instanceList.length; i++) { 581 int funcType = CoordCommandUtils.getFuncType(instanceList[i]); 582 if (funcType == CoordCommandUtils.LATEST || funcType == CoordCommandUtils.FUTURE) { 583 if (unresolvedInstances.length() > 0) { 584 unresolvedInstances.append(CoordELFunctions.INSTANCE_SEPARATOR); 585 } 586 unresolvedInstances.append(instanceList[i]); 587 continue; 588 } 589 ELEvaluator eval = CoordELEvaluator.createURIELEvaluator(instanceList[i]); 590 if (uris.length() > 0) { 591 uris.append(CoordELFunctions.INSTANCE_SEPARATOR); 592 } 593 uris.append(CoordELFunctions.evalAndWrap(eval, event.getChild("dataset", event.getNamespace()).getChild( 594 "uri-template", event.getNamespace()).getTextTrim())); 595 } 596 return uris.toString(); 597 } 598 599 /** 600 * getting the error code of the coord action. (used mainly for unit testing) 601 */ 602 protected String getCoordActionErrorCode() { 603 if (coordAction != null) { 604 return coordAction.getErrorCode(); 605 } 606 return null; 607 } 608 609 /** 610 * getting the error message of the coord action. (used mainly for unit testing) 611 */ 612 protected String getCoordActionErrorMsg() { 613 if (coordAction != null) { 614 return coordAction.getErrorMessage(); 615 } 616 return null; 617 } 618 619 @Override 620 public String getEntityKey() { 621 return this.jobId; 622 } 623 624 @Override 625 protected boolean isLockRequired() { 626 return true; 627 } 628 629 @Override 630 protected void loadState() throws CommandException { 631 if (jpaService == null) { 632 jpaService = Services.get().get(JPAService.class); 633 } 634 try { 635 coordAction = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(actionId)); 636 coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB_INPUT_CHECK, 637 coordAction.getJobId()); 638 } 639 catch (JPAExecutorException je) { 640 throw new CommandException(je); 641 } 642 LogUtils.setLogInfo(coordAction); 643 } 644 645 @Override 646 protected void verifyPrecondition() throws CommandException, PreconditionException { 647 if (coordAction.getStatus() != CoordinatorActionBean.Status.WAITING) { 648 throw new PreconditionException(ErrorCode.E1100, "[" + actionId 649 + "]::CoordActionInputCheck:: Ignoring action. Should be in WAITING state, but state=" 650 + coordAction.getStatus()); 651 } 652 653 // if eligible to do action input check when running with backward support is true 654 if (StatusUtils.getStatusForCoordActionInputCheck(coordJob)) { 655 return; 656 } 657 658 if (coordJob.getStatus() != Job.Status.RUNNING && coordJob.getStatus() != Job.Status.RUNNINGWITHERROR && coordJob.getStatus() != Job.Status.PAUSED 659 && coordJob.getStatus() != Job.Status.PAUSEDWITHERROR) { 660 throw new PreconditionException( 661 ErrorCode.E1100, "["+ actionId + "]::CoordActionInputCheck:: Ignoring action." + 662 " Coordinator job is not in RUNNING/RUNNINGWITHERROR/PAUSED/PAUSEDWITHERROR state, but state=" 663 + coordJob.getStatus()); 664 } 665 } 666 667 @Override 668 public String getKey(){ 669 return getName() + "_" + actionId; 670 } 671 672}