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.util.List; 024import java.util.Set; 025 026import org.apache.hadoop.conf.Configuration; 027import org.apache.hadoop.fs.FileSystem; 028import org.apache.hadoop.fs.Path; 029import org.apache.oozie.ErrorCode; 030import org.apache.oozie.action.hadoop.FSLauncherURIHandler; 031import org.apache.oozie.action.hadoop.LauncherURIHandler; 032import org.apache.oozie.service.HadoopAccessorException; 033import org.apache.oozie.service.HadoopAccessorService; 034import org.apache.oozie.service.Services; 035 036public class FSURIHandler implements URIHandler { 037 038 private HadoopAccessorService service; 039 private Set<String> supportedSchemes; 040 private List<Class<?>> classesToShip; 041 042 @Override 043 public void init(Configuration conf) { 044 service = Services.get().get(HadoopAccessorService.class); 045 supportedSchemes = service.getSupportedSchemes(); 046 classesToShip = new FSLauncherURIHandler().getClassesForLauncher(); 047 } 048 049 @Override 050 public Set<String> getSupportedSchemes() { 051 return supportedSchemes; 052 } 053 054 @Override 055 public Class<? extends LauncherURIHandler> getLauncherURIHandlerClass() { 056 return FSLauncherURIHandler.class; 057 } 058 059 @Override 060 public List<Class<?>> getClassesForLauncher() { 061 return classesToShip; 062 } 063 064 @Override 065 public DependencyType getDependencyType(URI uri) throws URIHandlerException { 066 return DependencyType.PULL; 067 } 068 069 @Override 070 public void registerForNotification(URI uri, Configuration conf, String user, String actionID) 071 throws URIHandlerException { 072 throw new UnsupportedOperationException("Notifications are not supported for " + uri.getScheme()); 073 } 074 075 @Override 076 public boolean unregisterFromNotification(URI uri, String actionID) { 077 throw new UnsupportedOperationException("Notifications are not supported for " + uri.getScheme()); 078 } 079 080 @Override 081 public Context getContext(URI uri, Configuration conf, String user) throws URIHandlerException { 082 FileSystem fs = getFileSystem(uri, conf, user); 083 return new FSContext(conf, user, fs); 084 } 085 086 @Override 087 public boolean exists(URI uri, Context context) throws URIHandlerException { 088 try { 089 FileSystem fs = ((FSContext) context).getFileSystem(); 090 return fs.exists(getNormalizedPath(uri)); 091 } 092 catch (IOException e) { 093 throw new HadoopAccessorException(ErrorCode.E0902, e); 094 } 095 } 096 097 @Override 098 public boolean exists(URI uri, Configuration conf, String user) throws URIHandlerException { 099 try { 100 FileSystem fs = getFileSystem(uri, conf, user); 101 return fs.exists(getNormalizedPath(uri)); 102 } 103 catch (HadoopAccessorException e) { 104 if (ErrorCode.E0902.equals(e.getErrorCode()) && e.getMessage() != null 105 && e.getMessage().indexOf("Invalid path for the Har Filesystem. No index file") != -1) { 106 return false; 107 } 108 else { 109 throw e; 110 } 111 } 112 catch (IOException e) { 113 throw new HadoopAccessorException(ErrorCode.E0902, e); 114 } 115 } 116 117 @Override 118 public String getURIWithDoneFlag(String uri, String doneFlag) throws URIHandlerException { 119 if (doneFlag.length() > 0) { 120 uri += "/" + doneFlag; 121 } 122 return uri; 123 } 124 125 @Override 126 public void validate(String uri) throws URIHandlerException { 127 } 128 129 @Override 130 public void destroy() { 131 132 } 133 134 private Path getNormalizedPath(URI uri) { 135 // Normalizes uri path replacing // with / in the path which users specify by mistake 136 return new Path(uri.getScheme(), uri.getAuthority(), uri.getPath()); 137 } 138 139 private FileSystem getFileSystem(URI uri, Configuration conf, String user) throws HadoopAccessorException { 140 if (user == null) { 141 throw new HadoopAccessorException(ErrorCode.E0902, "user has to be specified to access FileSystem"); 142 } 143 Configuration fsConf = service.createJobConf(uri.getAuthority()); 144 return service.createFileSystem(user, uri, fsConf); 145 } 146 147 static class FSContext extends Context { 148 149 private FileSystem fs; 150 151 /** 152 * Create a FSContext that can be used to access a filesystem URI 153 * 154 * @param conf Configuration to access the URI 155 * @param user name of the user the URI should be accessed as 156 * @param fs FileSystem to access 157 */ 158 public FSContext(Configuration conf, String user, FileSystem fs) { 159 super(conf, user); 160 this.fs = fs; 161 } 162 163 /** 164 * Get the FileSystem to access the URI 165 * @return FileSystem to access the URI 166 */ 167 public FileSystem getFileSystem() { 168 return fs; 169 } 170 } 171 172}