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.dependency;
020
021import java.io.IOException;
022import java.net.URI;
023import java.net.URISyntaxException;
024import java.util.Arrays;
025import java.util.HashMap;
026import java.util.HashSet;
027import java.util.List;
028import java.util.Map;
029import java.util.Set;
030
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.hive.conf.HiveConf;
033import org.apache.hadoop.security.UserGroupInformation;
034import org.apache.hive.hcatalog.api.ConnectionFailureException;
035import org.apache.hive.hcatalog.api.HCatClient;
036import org.apache.hive.hcatalog.api.HCatPartition;
037import org.apache.hive.hcatalog.common.HCatException;
038import org.apache.oozie.ErrorCode;
039import org.apache.oozie.action.hadoop.HCatLauncherURIHandler;
040import org.apache.oozie.action.hadoop.LauncherURIHandler;
041import org.apache.oozie.dependency.hcat.HCatMessageHandler;
042import org.apache.oozie.service.HCatAccessorException;
043import org.apache.oozie.service.HCatAccessorService;
044import org.apache.oozie.service.PartitionDependencyManagerService;
045import org.apache.oozie.service.Services;
046import org.apache.oozie.service.URIHandlerService;
047import org.apache.oozie.util.HCatURI;
048import org.apache.oozie.util.XLog;
049
050public class HCatURIHandler implements URIHandler {
051
052    private Set<String> supportedSchemes;
053    private Map<String, DependencyType> dependencyTypes;
054    private List<Class<?>> classesToShip;
055
056    @Override
057    public void init(Configuration conf) {
058        dependencyTypes = new HashMap<String, DependencyType>();
059        supportedSchemes = new HashSet<String>();
060        String[] schemes = conf.getStrings(URIHandlerService.URI_HANDLER_SUPPORTED_SCHEMES_PREFIX
061                + this.getClass().getSimpleName() + URIHandlerService.URI_HANDLER_SUPPORTED_SCHEMES_SUFFIX, "hcat");
062        supportedSchemes.addAll(Arrays.asList(schemes));
063        classesToShip = new HCatLauncherURIHandler().getClassesForLauncher();
064    }
065
066    @Override
067    public Set<String> getSupportedSchemes() {
068        return supportedSchemes;
069    }
070
071    @Override
072    public Class<? extends LauncherURIHandler> getLauncherURIHandlerClass() {
073        return HCatLauncherURIHandler.class;
074    }
075
076    @Override
077    public List<Class<?>> getClassesForLauncher() {
078        return classesToShip;
079    }
080
081    @Override
082    public DependencyType getDependencyType(URI uri) throws URIHandlerException {
083        DependencyType depType = DependencyType.PULL;
084        // Not initializing in constructor as this will be part of oozie.services.ext
085        // and will be initialized after URIHandlerService
086        HCatAccessorService hcatService = Services.get().get(HCatAccessorService.class);
087        if (hcatService != null) {
088            depType = dependencyTypes.get(uri.getAuthority());
089            if (depType == null) {
090                depType = hcatService.isKnownPublisher(uri) ? DependencyType.PUSH : DependencyType.PULL;
091                dependencyTypes.put(uri.getAuthority(), depType);
092            }
093        }
094        return depType;
095    }
096
097    @Override
098    public void registerForNotification(URI uri, Configuration conf, String user, String actionID)
099            throws URIHandlerException {
100        HCatURI hcatURI;
101        try {
102            hcatURI = new HCatURI(uri);
103        }
104        catch (URISyntaxException e) {
105            throw new URIHandlerException(ErrorCode.E0906, uri, e);
106        }
107        HCatAccessorService hcatService = Services.get().get(HCatAccessorService.class);
108        if (!hcatService.isRegisteredForNotification(hcatURI)) {
109            HCatClient client = getHCatClient(uri, conf, user);
110            try {
111                String topic = client.getMessageBusTopicName(hcatURI.getDb(), hcatURI.getTable());
112                if (topic == null) {
113                    return;
114                }
115                hcatService.registerForNotification(hcatURI, topic, new HCatMessageHandler(uri.getAuthority()));
116            }
117            catch (HCatException e) {
118                throw new HCatAccessorException(ErrorCode.E1501, e);
119            }
120            finally {
121                closeQuietly(client, true);
122            }
123        }
124        PartitionDependencyManagerService pdmService = Services.get().get(PartitionDependencyManagerService.class);
125        pdmService.addMissingDependency(hcatURI, actionID);
126    }
127
128    @Override
129    public boolean unregisterFromNotification(URI uri, String actionID) {
130        HCatURI hcatURI;
131        try {
132            hcatURI = new HCatURI(uri);
133        }
134        catch (URISyntaxException e) {
135            throw new RuntimeException(e); // Unexpected at this point
136        }
137        PartitionDependencyManagerService pdmService = Services.get().get(PartitionDependencyManagerService.class);
138        return pdmService.removeMissingDependency(hcatURI, actionID);
139    }
140
141    @Override
142    public Context getContext(URI uri, Configuration conf, String user) throws URIHandlerException {
143        HCatClient client = getHCatClient(uri, conf, user);
144        return new HCatContext(conf, user, client);
145    }
146
147    @Override
148    public boolean exists(URI uri, Context context) throws URIHandlerException {
149        HCatClient client = ((HCatContext) context).getHCatClient();
150        return exists(uri, client, false);
151    }
152
153    @Override
154    public boolean exists(URI uri, Configuration conf, String user) throws URIHandlerException {
155        HCatClient client = getHCatClient(uri, conf, user);
156        return exists(uri, client, true);
157    }
158
159    @Override
160    public String getURIWithDoneFlag(String uri, String doneFlag) throws URIHandlerException {
161        return uri;
162    }
163
164    @Override
165    public void validate(String uri) throws URIHandlerException {
166        try {
167            new HCatURI(uri); // will fail if uri syntax is incorrect
168        }
169        catch (URISyntaxException e) {
170            throw new URIHandlerException(ErrorCode.E0906, uri, e);
171        }
172
173    }
174
175    @Override
176    public void destroy() {
177
178    }
179
180    private HCatClient getHCatClient(URI uri, Configuration conf, String user) throws HCatAccessorException {
181        HCatAccessorService hcatService = Services.get().get(HCatAccessorService.class);
182        if (hcatService.getHCatConf() != null) {
183            conf = hcatService.getHCatConf();
184        }
185        final HiveConf hiveConf = new HiveConf(conf, this.getClass());
186        String serverURI = getMetastoreConnectURI(uri);
187        if (!serverURI.equals("")) {
188            hiveConf.set("hive.metastore.local", "false");
189        }
190        hiveConf.set(HiveConf.ConfVars.METASTOREURIS.varname, serverURI);
191        try {
192            XLog.getLog(HCatURIHandler.class).info(
193                    "Creating HCatClient for user [{0}] login_user [{1}] and server [{2}] ", user,
194                    UserGroupInformation.getLoginUser(), serverURI);
195
196            // HiveMetastoreClient (hive 0.9) currently does not work if UGI has doAs
197            // We are good to connect as the oozie user since listPartitions does not require
198            // authorization
199            /*
200            UserGroupInformation ugi = ugiService.getProxyUser(user);
201            return ugi.doAs(new PrivilegedExceptionAction<HCatClient>() {
202                public HCatClient run() throws Exception {
203                    return HCatClient.create(hiveConf);
204                }
205            });
206            */
207
208            return HCatClient.create(hiveConf);
209        }
210        catch (HCatException e) {
211            throw new HCatAccessorException(ErrorCode.E1501, e);
212        }
213        catch (IOException e) {
214            throw new HCatAccessorException(ErrorCode.E1501, e);
215        }
216
217    }
218
219    private String getMetastoreConnectURI(URI uri) {
220        String metastoreURI;
221        // For unit tests
222        if (uri.getAuthority().equals("unittest-local")) {
223            metastoreURI = "";
224        }
225        else {
226            // Hardcoding hcat to thrift mapping till support for webhcat(templeton)
227            // is added
228            metastoreURI = "thrift://" + uri.getAuthority();
229        }
230        return metastoreURI;
231    }
232
233    private boolean exists(URI uri, HCatClient client, boolean closeClient) throws HCatAccessorException {
234        try {
235            HCatURI hcatURI = new HCatURI(uri.toString());
236            List<HCatPartition> partitions = client.getPartitions(hcatURI.getDb(), hcatURI.getTable(),
237                    hcatURI.getPartitionMap());
238            return (partitions != null && !partitions.isEmpty());
239        }
240        catch (ConnectionFailureException e) {
241            throw new HCatAccessorException(ErrorCode.E1501, e);
242        }
243        catch (HCatException e) {
244            throw new HCatAccessorException(ErrorCode.E0902, e);
245        }
246        catch (URISyntaxException e) {
247            throw new HCatAccessorException(ErrorCode.E0902, e);
248        }
249        finally {
250            closeQuietly(client, closeClient);
251        }
252    }
253
254    private void closeQuietly(HCatClient client, boolean close) {
255        if (close && client != null) {
256            try {
257                client.close();
258            }
259            catch (Exception ignore) {
260                XLog.getLog(HCatURIHandler.class).warn("Error closing hcat client", ignore);
261            }
262        }
263    }
264
265    static class HCatContext extends Context {
266
267        private HCatClient hcatClient;
268
269        /**
270         * Create a HCatContext that can be used to access a hcat URI
271         *
272         * @param conf Configuration to access the URI
273         * @param user name of the user the URI should be accessed as
274         * @param hcatClient HCatClient to talk to hcatalog server
275         */
276        public HCatContext(Configuration conf, String user, HCatClient hcatClient) {
277            super(conf, user);
278            this.hcatClient = hcatClient;
279        }
280
281        /**
282         * Get the HCatClient to talk to hcatalog server
283         *
284         * @return HCatClient to talk to hcatalog server
285         */
286        public HCatClient getHCatClient() {
287            return hcatClient;
288        }
289
290        @Override
291        public void destroy() {
292            try {
293                hcatClient.close();
294            }
295            catch (Exception ignore) {
296                XLog.getLog(HCatContext.class).warn("Error closing hcat client", ignore);
297            }
298        }
299
300    }
301
302}