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.util.ArrayList; 022import java.util.Arrays; 023import java.util.Collection; 024import java.util.List; 025import org.apache.oozie.command.CommandException; 026import org.apache.oozie.dependency.DependencyChecker; 027import org.apache.oozie.service.PartitionDependencyManagerService; 028import org.apache.oozie.service.Services; 029 030public class CoordActionUpdatePushMissingDependency extends CoordPushDependencyCheckXCommand { 031 032 public CoordActionUpdatePushMissingDependency(String actionId) { 033 super("coord_action_push_md", actionId); 034 } 035 036 @Override 037 protected Void execute() throws CommandException { 038 LOG.info("STARTED for Action id [{0}]", actionId); 039 String pushMissingDeps = coordAction.getPushMissingDependencies(); 040 if (pushMissingDeps == null || pushMissingDeps.length() == 0) { 041 LOG.info("Nothing to check. Empty push missing dependency"); 042 } 043 else { 044 PartitionDependencyManagerService pdms = Services.get().get(PartitionDependencyManagerService.class); 045 Collection<String> availDepList = pdms.getAvailableDependencyURIs(actionId); 046 if (availDepList == null || availDepList.size() == 0) { 047 LOG.info("There are no available dependencies"); 048 if (isTimeout()) { // Poll and check as one last try 049 queue(new CoordPushDependencyCheckXCommand(coordAction.getId()), 100); 050 } 051 } 052 else { 053 LOG.debug("Updating with available uris=[{0}] where missing uris=[{1}]", availDepList.toString(), 054 pushMissingDeps); 055 056 String[] missingDepsArray = DependencyChecker.dependenciesAsArray(pushMissingDeps); 057 List<String> stillMissingDepsList = new ArrayList<String>(Arrays.asList(missingDepsArray)); 058 stillMissingDepsList.removeAll(availDepList); 059 boolean isChangeInDependency = true; 060 if (stillMissingDepsList.size() == 0) { 061 // All push-based dependencies are available 062 onAllPushDependenciesAvailable(); 063 } 064 else { 065 if (stillMissingDepsList.size() == missingDepsArray.length) { 066 isChangeInDependency = false; 067 } 068 else { 069 String stillMissingDeps = DependencyChecker.dependenciesAsString(stillMissingDepsList); 070 coordAction.setPushMissingDependencies(stillMissingDeps); 071 } 072 if (isTimeout()) { // Poll and check as one last try 073 queue(new CoordPushDependencyCheckXCommand(coordAction.getId()), 100); 074 } 075 } 076 updateCoordAction(coordAction, isChangeInDependency); 077 removeAvailableDependencies(pdms, availDepList); 078 LOG.info("ENDED for Action id [{0}]", actionId); 079 } 080 } 081 return null; 082 } 083 084 private void removeAvailableDependencies(PartitionDependencyManagerService pdms, Collection<String> availDepList) { 085 if (pdms.removeAvailableDependencyURIs(actionId, availDepList)) { 086 LOG.debug("Successfully removed uris [{0}] from available list", availDepList.toString()); 087 } 088 else { 089 LOG.warn("Failed to remove uris [{0}] from available list", availDepList.toString(), actionId); 090 } 091 } 092 093 @Override 094 public String getEntityKey() { 095 return actionId; 096 } 097 098}