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}