big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HADOOP-13591. Unit test failure in TestOSSContractGetFileStatus and TestOSSContractRootDir. Contributed by Genmao Yu

Browse Source
This commit is contained in:
Kai Zheng 2016-09-20 15:12:02 +08:00
parent 9cd4760257
commit 08b37603d9
5 changed files with 89 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystem.java
View File

@ -24,6 +24,8 @@
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@ -124,9 +126,6 @@ private boolean innerDelete(FileStatus status, boolean recursive)
Path f = status.getPath();
String key = pathToKey(f);
if (status.isDirectory()) {
if (!key.endsWith("/")) {
key += "/";
}
if (!recursive) {
FileStatus[] statuses = listStatus(status.getPath());
// Check whether it is an empty directory or not
@ -135,6 +134,7 @@ private boolean innerDelete(FileStatus status, boolean recursive)
": It is not empty!");
} else {
// Delete empty directory without '-r'
key = AliyunOSSUtils.maybeAddTrailingSlash(key);
store.deleteObject(key);
}
} else {
@ -149,15 +149,9 @@ private boolean innerDelete(FileStatus status, boolean recursive)
}
private void createFakeDirectoryIfNecessary(Path f) throws IOException {
try {
Path pPath = f.getParent();
FileStatus pStatus = getFileStatus(pPath);
if (pStatus.isFile()) {
throw new IOException("Path " + pPath +
" is assumed to be a directory!");
}
} catch (FileNotFoundException fnfe) {
// Make sure the parent directory exists
String key = pathToKey(f);
if (StringUtils.isNotEmpty(key) && !exists(f)) {
LOG.debug("Creating new fake directory at {}", f);
mkdir(pathToKey(f.getParent()));
}
}
@ -175,14 +169,14 @@ public FileStatus getFileStatus(Path path) throws IOException {
ObjectMetadata meta = store.getObjectMetadata(key);
// If key not found and key does not end with "/"
if (meta == null && !key.endsWith("/")) {
// Case: dir + "/"
// In case of 'dir + "/"'
key += "/";
meta = store.getObjectMetadata(key);
}
if (meta == null) {
ObjectListing listing = store.listObjects(key, 1, "/", null);
if (!listing.getObjectSummaries().isEmpty() ||
!listing.getCommonPrefixes().isEmpty()) {
ObjectListing listing = store.listObjects(key, 1, null, false);
if (CollectionUtils.isNotEmpty(listing.getObjectSummaries()) ||
CollectionUtils.isNotEmpty(listing.getCommonPrefixes())) {
return new FileStatus(0, true, 1, 0, 0, qualifiedPath);
} else {
throw new FileNotFoundException(path + ": No such file or directory!");
@ -251,7 +245,7 @@ public void initialize(URI name, Configuration conf) throws IOException {
*/
private boolean objectRepresentsDirectory(final String name,
final long size) {
return !name.isEmpty() && name.endsWith("/") && size == 0L;
return StringUtils.isNotEmpty(name) && name.endsWith("/") && size == 0L;
}
/**
@ -265,10 +259,6 @@ private String pathToKey(Path path) {
path = new Path(workingDir, path);
}
if (path.toUri().getScheme() != null && path.toUri().getPath().isEmpty()) {
return "";
}
return path.toUri().getPath().substring(1);
}
@ -287,26 +277,23 @@ public FileStatus[] listStatus(Path path) throws IOException {
final FileStatus fileStatus = getFileStatus(path);
if (fileStatus.isDirectory()) {
if (!key.endsWith("/")) {
key = key + "/";
}
if (LOG.isDebugEnabled()) {
LOG.debug("listStatus: doing listObjects for directory " + key);
}
ObjectListing objects = store.listObjects(key, maxKeys, "/", null);
ObjectListing objects = store.listObjects(key, maxKeys, null, false);
while (true) {
statistics.incrementReadOps(1);
for (OSSObjectSummary objectSummary : objects.getObjectSummaries()) {
Path keyPath = keyToPath(objectSummary.getKey())
.makeQualified(uri, workingDir);
if (keyPath.equals(path)) {
String objKey = objectSummary.getKey();
if (objKey.equals(key + "/")) {
if (LOG.isDebugEnabled()) {
LOG.debug("Ignoring: " + keyPath);
LOG.debug("Ignoring: " + objKey);
}
continue;
} else {
Path keyPath = keyToPath(objectSummary.getKey())
.makeQualified(uri, workingDir);
if (LOG.isDebugEnabled()) {
LOG.debug("Adding: fi: " + keyPath);
}
@ -317,10 +304,13 @@ public FileStatus[] listStatus(Path path) throws IOException {
}
for (String prefix : objects.getCommonPrefixes()) {
Path keyPath = keyToPath(prefix).makeQualified(uri, workingDir);
if (keyPath.equals(path)) {
if (prefix.equals(key + "/")) {
if (LOG.isDebugEnabled()) {
LOG.debug("Ignoring: " + prefix);
}
continue;
} else {
Path keyPath = keyToPath(prefix).makeQualified(uri, workingDir);
if (LOG.isDebugEnabled()) {
LOG.debug("Adding: rd: " + keyPath);
}
@ -332,8 +322,8 @@ public FileStatus[] listStatus(Path path) throws IOException {
if (LOG.isDebugEnabled()) {
LOG.debug("listStatus: list truncated - getting next batch");
}
objects = store.listObjects(key, maxKeys, "/",
objects.getNextMarker());
String nextMarker = objects.getNextMarker();
objects = store.listObjects(key, maxKeys, nextMarker, false);
statistics.incrementReadOps(1);
} else {
break;
@ -358,10 +348,12 @@ public FileStatus[] listStatus(Path path) throws IOException {
*/
private boolean mkdir(final String key) throws IOException {
String dirName = key;
if (!key.endsWith("/")) {
dirName += "/";
if (StringUtils.isNotEmpty(key)) {
if (!key.endsWith("/")) {
dirName += "/";
}
store.storeEmptyFile(dirName);
}
store.storeEmptyFile(dirName);
return true;
}
@ -506,16 +498,11 @@ private boolean copyFile(Path srcPath, Path dstPath) {
* @param dstPath destination path.
* @return true if directory is successfully copied.
*/
private boolean copyDirectory(Path srcPath, Path dstPath) {
String srcKey = pathToKey(srcPath);
String dstKey = pathToKey(dstPath);
if (!srcKey.endsWith("/")) {
srcKey = srcKey + "/";
}
if (!dstKey.endsWith("/")) {
dstKey = dstKey + "/";
}
private boolean copyDirectory(Path srcPath, Path dstPath) throws IOException {
String srcKey = AliyunOSSUtils
.maybeAddTrailingSlash(pathToKey(srcPath));
String dstKey = AliyunOSSUtils
.maybeAddTrailingSlash(pathToKey(dstPath));
if (dstKey.startsWith(srcKey)) {
if (LOG.isDebugEnabled()) {
@ -524,7 +511,8 @@ private boolean copyDirectory(Path srcPath, Path dstPath) {
return false;
}
ObjectListing objects = store.listObjects(srcKey, maxKeys, null, null);
store.storeEmptyFile(dstKey);
ObjectListing objects = store.listObjects(srcKey, maxKeys, null, true);
statistics.incrementReadOps(1);
// Copy files from src folder to dst
while (true) {
@ -534,8 +522,8 @@ private boolean copyDirectory(Path srcPath, Path dstPath) {
store.copyFile(objectSummary.getKey(), newKey);
}
if (objects.isTruncated()) {
objects = store.listObjects(srcKey, maxKeys, null,
objects.getNextMarker());
String nextMarker = objects.getNextMarker();
objects = store.listObjects(srcKey, maxKeys, nextMarker, true);
statistics.incrementReadOps(1);
} else {
break;

36
hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java
View File

@ -42,6 +42,8 @@
import com.aliyun.oss.model.UploadPartCopyResult;
import com.aliyun.oss.model.UploadPartRequest;
import com.aliyun.oss.model.UploadPartResult;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.slf4j.Logger;
@ -89,7 +91,7 @@ public void initialize(URI uri, Configuration conf,
String proxyHost = conf.getTrimmed(PROXY_HOST_KEY, "");
int proxyPort = conf.getInt(PROXY_PORT_KEY, -1);
if (!proxyHost.isEmpty()) {
if (StringUtils.isNotEmpty(proxyHost)) {
clientConf.setProxyHost(proxyHost);
if (proxyPort >= 0) {
clientConf.setProxyPort(proxyPort);
@ -123,7 +125,7 @@ public void initialize(URI uri, Configuration conf,
String endPoint = conf.getTrimmed(ENDPOINT_KEY, "");
CredentialsProvider provider =
AliyunOSSUtils.getCredentialsProvider(uri, conf);
AliyunOSSUtils.getCredentialsProvider(conf);
ossClient = new OSSClient(endPoint, provider, clientConf);
uploadPartSize = conf.getLong(MULTIPART_UPLOAD_SIZE_KEY,
MULTIPART_UPLOAD_SIZE_DEFAULT);
@ -153,7 +155,7 @@ public void initialize(URI uri, Configuration conf,
}
String cannedACLName = conf.get(CANNED_ACL_KEY, CANNED_ACL_DEFAULT);
if (!cannedACLName.isEmpty()) {
if (StringUtils.isNotEmpty(cannedACLName)) {
CannedAccessControlList cannedACL =
CannedAccessControlList.valueOf(cannedACLName);
ossClient.setBucketAcl(bucketName, cannedACL);
@ -179,11 +181,13 @@ public void deleteObject(String key) {
* @param keysToDelete collection of keys to delete.
*/
public void deleteObjects(List<String> keysToDelete) {
DeleteObjectsRequest deleteRequest =
new DeleteObjectsRequest(bucketName);
deleteRequest.setKeys(keysToDelete);
ossClient.deleteObjects(deleteRequest);
statistics.incrementWriteOps(keysToDelete.size());
if (CollectionUtils.isNotEmpty(keysToDelete)) {
DeleteObjectsRequest deleteRequest =
new DeleteObjectsRequest(bucketName);
deleteRequest.setKeys(keysToDelete);
ossClient.deleteObjects(deleteRequest);
statistics.incrementWriteOps(keysToDelete.size());
}
}
/**
@ -192,8 +196,10 @@ public void deleteObjects(List<String> keysToDelete) {
* @param key directory key to delete.
*/
public void deleteDirs(String key) {
key = AliyunOSSUtils.maybeAddTrailingSlash(key);
ListObjectsRequest listRequest = new ListObjectsRequest(bucketName);
listRequest.setPrefix(key);
listRequest.setDelimiter(null);
listRequest.setMaxKeys(maxKeys);
while (true) {
@ -299,7 +305,7 @@ private boolean multipartCopy(String srcKey, long contentLength,
InitiateMultipartUploadRequest initiateMultipartUploadRequest =
new InitiateMultipartUploadRequest(bucketName, dstKey);
ObjectMetadata meta = new ObjectMetadata();
if (!serverSideEncryptionAlgorithm.isEmpty()) {
if (StringUtils.isNotEmpty(serverSideEncryptionAlgorithm)) {
meta.setServerSideEncryption(serverSideEncryptionAlgorithm);
}
initiateMultipartUploadRequest.setObjectMetadata(meta);
@ -353,7 +359,7 @@ public void uploadObject(String key, File file) throws IOException {
FileInputStream fis = new FileInputStream(object);
ObjectMetadata meta = new ObjectMetadata();
meta.setContentLength(object.length());
if (!serverSideEncryptionAlgorithm.isEmpty()) {
if (StringUtils.isNotEmpty(serverSideEncryptionAlgorithm)) {
meta.setServerSideEncryption(serverSideEncryptionAlgorithm);
}
try {
@ -384,7 +390,7 @@ public void multipartUploadObject(String key, File file) throws IOException {
InitiateMultipartUploadRequest initiateMultipartUploadRequest =
new InitiateMultipartUploadRequest(bucketName, key);
ObjectMetadata meta = new ObjectMetadata();
if (!serverSideEncryptionAlgorithm.isEmpty()) {
if (StringUtils.isNotEmpty(serverSideEncryptionAlgorithm)) {
meta.setServerSideEncryption(serverSideEncryptionAlgorithm);
}
initiateMultipartUploadRequest.setObjectMetadata(meta);
@ -435,12 +441,14 @@ public void multipartUploadObject(String key, File file) throws IOException {
*
* @param prefix prefix.
* @param maxListingLength max no. of entries
* @param delimiter delimiter.
* @param marker last key in any previous search.
* @param recursive whether to list directory recursively.
* @return a list of matches.
*/
public ObjectListing listObjects(String prefix, int maxListingLength,
String delimiter, String marker) {
String marker, boolean recursive) {
String delimiter = recursive ? null : "/";
prefix = AliyunOSSUtils.maybeAddTrailingSlash(prefix);
ListObjectsRequest listRequest = new ListObjectsRequest(bucketName);
listRequest.setPrefix(prefix);
listRequest.setDelimiter(delimiter);
@ -488,7 +496,7 @@ public void close() {
public void purge(String prefix) {
String key;
try {
ObjectListing objects = listObjects(prefix, maxKeys, null, null);
ObjectListing objects = listObjects(prefix, maxKeys, null, true);
for (OSSObjectSummary object : objects.getObjectSummaries()) {
key = object.getKey();
ossClient.deleteObject(bucketName, key);

24
hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSUtils.java
View File

@ -20,7 +20,6 @@
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import com.aliyun.oss.common.auth.CredentialsProvider;
import org.apache.commons.lang.StringUtils;
@ -106,16 +105,13 @@ public static long calculatePartSize(long contentLength, long minPartSize) {
* Create credential provider specified by configuration, or create default
* credential provider if not specified.
*
* @param name the uri of the file system
* @param conf configuration
* @return a credential provider
* @throws IOException on any problem. Class construction issues may be
* nested inside the IOE.
*/
public static CredentialsProvider getCredentialsProvider(URI name,
Configuration conf) throws IOException {
URI uri = java.net.URI.create(
name.getScheme() + "://" + name.getAuthority());
public static CredentialsProvider getCredentialsProvider(Configuration conf)
throws IOException {
CredentialsProvider credentials;
String className = conf.getTrimmed(ALIYUN_OSS_CREDENTIALS_PROVIDER_KEY);
@ -152,4 +148,20 @@ public static CredentialsProvider getCredentialsProvider(URI name,
return credentials;
}
/**
* Turns a path (relative or otherwise) into an OSS key, adding a trailing
* "/" if the path is not the root <i>and</i> does not already have a "/"
* at the end.
*
* @param key OSS key or ""
* @return the with a trailing "/", or, if it is the root key, "".
*/
public static String maybeAddTrailingSlash(String key) {
if (StringUtils.isNotEmpty(key) && !key.endsWith("/")) {
return key + '/';
} else {
return key;
}
}
}

2
hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSTestUtils.java
View File

@ -48,7 +48,7 @@ public static AliyunOSSFileSystem createTestFileSystem(Configuration conf)
String fsname = conf.getTrimmed(
TestAliyunOSSFileSystemContract.TEST_FS_OSS_NAME, "");
boolean liveTest = !StringUtils.isEmpty(fsname);
boolean liveTest = StringUtils.isNotEmpty(fsname);
URI testURI = null;
if (liveTest) {
testURI = URI.create(fsname);

10
hadoop-tools/hadoop-aliyun/src/test/resources/contract/aliyun-oss.xml
View File

@ -98,6 +98,16 @@
<value>true</value>
</property>
<property>
<name>fs.contract.test.root-tests-enabled</name>
<value>true</value>
</property>
<property>
<name>fs.contract.supports-getfilestatus</name>
<value>true</value>
</property>
<property>
<name>fs.oss.multipart.download.size</name>
<value>102400</value>