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.wf;
020
021import org.apache.hadoop.conf.Configuration;
022import org.apache.oozie.service.WorkflowAppService;
023import org.jdom.Element;
024import org.jdom.Namespace;
025import org.apache.oozie.client.XOozieClient;
026import org.apache.oozie.command.CommandException;
027
028import java.util.HashMap;
029import java.util.HashSet;
030import java.util.Iterator;
031import java.util.Map;
032import java.util.Set;
033
034public class SubmitMRXCommand extends SubmitHttpXCommand {
035    private static final Set<String> SKIPPED_CONFS = new HashSet<String>();
036    private static final Map<String, String> DEPRECATE_MAP = new HashMap<String, String>();
037
038    public SubmitMRXCommand(Configuration conf) {
039        super("submitMR", "submitMR", conf);
040    }
041
042    static {
043        SKIPPED_CONFS.add(WorkflowAppService.HADOOP_USER);
044        SKIPPED_CONFS.add(XOozieClient.JT);
045        SKIPPED_CONFS.add(XOozieClient.NN);
046
047        DEPRECATE_MAP.put(XOozieClient.NN, XOozieClient.NN_2);
048        DEPRECATE_MAP.put(XOozieClient.JT, XOozieClient.JT_2);
049        DEPRECATE_MAP.put(WorkflowAppService.HADOOP_USER, "mapreduce.job.user.name");
050    }
051
052    @Override
053    protected Namespace getSectionNamespace(){
054        return Namespace.getNamespace("uri:oozie:workflow:0.2");
055    }
056
057    @Override
058    protected String getWorkflowName(){
059        return "mapreduce";
060    }
061
062    private Element generateConfigurationSection(Configuration conf, Namespace ns) {
063        Element configuration = null;
064        Iterator<Map.Entry<String, String>> iter = conf.iterator();
065        while (iter.hasNext()) {
066            Map.Entry<String, String> entry = iter.next();
067            String name = entry.getKey();
068            if (MANDATORY_OOZIE_CONFS.contains(name) || OPTIONAL_OOZIE_CONFS.contains(name)
069                    || SKIPPED_CONFS.contains(name)
070                    || DEPRECATE_MAP.containsValue(name)) {
071                continue;
072            }
073
074            if (configuration == null) {
075                configuration = new Element("configuration", ns);
076            }
077
078            String value = entry.getValue();
079            Element property = new Element("property", ns);
080            Element nameElement = new Element("name", ns);
081            nameElement.addContent(name != null ? name : "");
082            property.addContent(nameElement);
083            Element valueElement = new Element("value", ns);
084            valueElement.addContent(value != null ? value : "");
085            property.addContent(valueElement);
086            configuration.addContent(property);
087        }
088
089        return configuration;
090    }
091
092    @Override
093    protected Element generateSection(Configuration conf, Namespace ns) {
094        Element mapreduce = new Element("map-reduce", ns);
095        Element jt = new Element("job-tracker", ns);
096        String newJTVal = conf.get(DEPRECATE_MAP.get(XOozieClient.JT));
097        jt.addContent(newJTVal != null ? newJTVal : (conf.get(XOozieClient.JT)));
098        mapreduce.addContent(jt);
099        Element nn = new Element("name-node", ns);
100        String newNNVal = conf.get(DEPRECATE_MAP.get(XOozieClient.NN));
101        nn.addContent(newNNVal != null ? newNNVal : (conf.get(XOozieClient.NN)));
102        mapreduce.addContent(nn);
103
104        if (conf.size() > MANDATORY_OOZIE_CONFS.size()) { // excluding JT, NN,
105                                                          // LIBPATH
106            // configuration section
107            Element configuration = generateConfigurationSection(conf, ns);
108            if (configuration != null) {
109                mapreduce.addContent(configuration);
110            }
111
112            // file section
113            addFileSection(mapreduce, conf, ns);
114
115            // archive section
116            addArchiveSection(mapreduce, conf, ns);
117        }
118
119        return mapreduce;
120    }
121
122    @Override
123    protected void checkMandatoryConf(Configuration conf) {
124        for (String key : MANDATORY_OOZIE_CONFS) {
125            String value = conf.get(key);
126            if(value == null) {
127                if(DEPRECATE_MAP.containsKey(key)) {
128                    if(conf.get(DEPRECATE_MAP.get(key)) == null) {
129                        throw new RuntimeException(key + " or " + DEPRECATE_MAP.get(key) + " is not specified");
130                    }
131                }
132                else {
133                    throw new RuntimeException(key + " is not specified");
134                }
135            }
136        }
137    }
138
139    @Override
140    public String getEntityKey() {
141        return null;
142    }
143
144    @Override
145    protected boolean isLockRequired() {
146        return false;
147    }
148
149    @Override
150    protected void loadState() {
151
152    }
153
154    @Override
155    protected void verifyPrecondition() throws CommandException {
156
157    }
158}