CosNativeFileSystemStore.java
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.fs.cosn;
import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import com.qcloud.cos.COSClient;
import com.qcloud.cos.ClientConfig;
import com.qcloud.cos.auth.BasicCOSCredentials;
import com.qcloud.cos.auth.COSCredentials;
import com.qcloud.cos.endpoint.SuffixEndpointBuilder;
import com.qcloud.cos.exception.CosClientException;
import com.qcloud.cos.exception.CosServiceException;
import com.qcloud.cos.http.HttpProtocol;
import com.qcloud.cos.model.AbortMultipartUploadRequest;
import com.qcloud.cos.model.COSObject;
import com.qcloud.cos.model.COSObjectSummary;
import com.qcloud.cos.model.CompleteMultipartUploadRequest;
import com.qcloud.cos.model.CompleteMultipartUploadResult;
import com.qcloud.cos.model.CopyObjectRequest;
import com.qcloud.cos.model.DeleteObjectRequest;
import com.qcloud.cos.model.GetObjectMetadataRequest;
import com.qcloud.cos.model.GetObjectRequest;
import com.qcloud.cos.model.InitiateMultipartUploadRequest;
import com.qcloud.cos.model.InitiateMultipartUploadResult;
import com.qcloud.cos.model.ListObjectsRequest;
import com.qcloud.cos.model.ObjectListing;
import com.qcloud.cos.model.ObjectMetadata;
import com.qcloud.cos.model.PartETag;
import com.qcloud.cos.model.PutObjectRequest;
import com.qcloud.cos.model.PutObjectResult;
import com.qcloud.cos.model.UploadPartRequest;
import com.qcloud.cos.model.UploadPartResult;
import com.qcloud.cos.region.Region;
import com.qcloud.cos.utils.Base64;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.cosn.auth.COSCredentialsProviderList;
import org.apache.hadoop.util.VersionInfo;
import org.apache.http.HttpStatus;
/**
* The class actually performs access operation to the COS blob store.
* It provides the bridging logic for the Hadoop's abstract filesystem and COS.
*/
@InterfaceAudience.Private
@InterfaceStability.Unstable
class CosNativeFileSystemStore implements NativeFileSystemStore {
private COSClient cosClient;
private String bucketName;
private int maxRetryTimes;
public static final Logger LOG =
LoggerFactory.getLogger(CosNativeFileSystemStore.class);
/**
* Initialize the client to access COS blob storage.
*
* @param conf Hadoop configuration with COS configuration options.
* @throws IOException Initialize the COS client failed,
* caused by incorrect options.
*/
private void initCOSClient(URI uri, Configuration conf) throws IOException {
COSCredentialsProviderList credentialProviderList =
CosNUtils.createCosCredentialsProviderSet(uri, conf);
String region = conf.get(CosNConfigKeys.COSN_REGION_KEY);
String endpointSuffix = conf.get(
CosNConfigKeys.COSN_ENDPOINT_SUFFIX_KEY);
if (null == region && null == endpointSuffix) {
String exceptionMsg = String.format("config %s and %s at least one",
CosNConfigKeys.COSN_REGION_KEY,
CosNConfigKeys.COSN_ENDPOINT_SUFFIX_KEY);
throw new IOException(exceptionMsg);
}
COSCredentials cosCred;
cosCred = new BasicCOSCredentials(
credentialProviderList.getCredentials().getCOSAccessKeyId(),
credentialProviderList.getCredentials().getCOSSecretKey());
boolean useHttps = conf.getBoolean(CosNConfigKeys.COSN_USE_HTTPS_KEY,
CosNConfigKeys.DEFAULT_USE_HTTPS);
ClientConfig config;
if (null == region) {
config = new ClientConfig(new Region(""));
config.setEndpointBuilder(new SuffixEndpointBuilder(endpointSuffix));
} else {
config = new ClientConfig(new Region(region));
}
if (useHttps) {
config.setHttpProtocol(HttpProtocol.https);
}
config.setUserAgent(conf.get(CosNConfigKeys.USER_AGENT,
CosNConfigKeys.DEFAULT_USER_AGENT) + " For " + " Hadoop "
+ VersionInfo.getVersion());
this.maxRetryTimes = conf.getInt(CosNConfigKeys.COSN_MAX_RETRIES_KEY,
CosNConfigKeys.DEFAULT_MAX_RETRIES);
config.setMaxConnectionsCount(
conf.getInt(CosNConfigKeys.MAX_CONNECTION_NUM,
CosNConfigKeys.DEFAULT_MAX_CONNECTION_NUM));
this.cosClient = new COSClient(cosCred, config);
}
/**
* Initialize the CosNativeFileSystemStore object, including
* its COS client and default COS bucket.
*
* @param uri The URI of the COS bucket accessed by default.
* @param conf Hadoop configuration with COS configuration options.
* @throws IOException Initialize the COS client failed.
*/
@Override
public void initialize(URI uri, Configuration conf) throws IOException {
try {
initCOSClient(uri, conf);
this.bucketName = uri.getHost();
} catch (Exception e) {
handleException(e, "");
}
}
/**
* Store a file into COS from the specified input stream, which would be
* retried until the success or maximum number.
*
* @param key COS object key.
* @param inputStream Input stream to be uploaded into COS.
* @param md5Hash MD5 value of the content to be uploaded.
* @param length Length of uploaded content.
* @throws IOException Upload the file failed.
*/
private void storeFileWithRetry(String key, InputStream inputStream,
byte[] md5Hash, long length) throws IOException {
try {
ObjectMetadata objectMetadata = new ObjectMetadata();
objectMetadata.setContentMD5(Base64.encodeAsString(md5Hash));
objectMetadata.setContentLength(length);
PutObjectRequest putObjectRequest =
new PutObjectRequest(bucketName, key, inputStream, objectMetadata);
PutObjectResult putObjectResult =
(PutObjectResult) callCOSClientWithRetry(putObjectRequest);
LOG.debug("Store file successfully. COS key: [{}], ETag: [{}].",
key, putObjectResult.getETag());
} catch (Exception e) {
String errMsg = String.format("Store file failed. COS key: [%s], "
+ "exception: [%s]", key, e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), key);
}
}
/**
* Store a local file into COS.
*
* @param key COS object key.
* @param file The local file to be uploaded.
* @param md5Hash The MD5 value of the file to be uploaded.
* @throws IOException Upload the file failed.
*/
@Override
public void storeFile(String key, File file, byte[] md5Hash)
throws IOException {
LOG.info("Store file from local path: [{}]. file length: [{}] COS key: " +
"[{}]", file.getCanonicalPath(), file.length(), key);
storeFileWithRetry(key, new BufferedInputStream(new FileInputStream(file)),
md5Hash, file.length());
}
/**
* Store a file into COS from the specified input stream.
*
* @param key COS object key.
* @param inputStream The Input stream to be uploaded.
* @param md5Hash The MD5 value of the content to be uploaded.
* @param contentLength Length of uploaded content.
* @throws IOException Upload the file failed.
*/
@Override
public void storeFile(
String key,
InputStream inputStream,
byte[] md5Hash,
long contentLength) throws IOException {
LOG.info("Store file from input stream. COS key: [{}], "
+ "length: [{}].", key, contentLength);
storeFileWithRetry(key, inputStream, md5Hash, contentLength);
}
// For cos, storeEmptyFile means creating a directory
@Override
public void storeEmptyFile(String key) throws IOException {
if (!key.endsWith(CosNFileSystem.PATH_DELIMITER)) {
key = key + CosNFileSystem.PATH_DELIMITER;
}
ObjectMetadata objectMetadata = new ObjectMetadata();
objectMetadata.setContentLength(0);
InputStream input = new ByteArrayInputStream(new byte[0]);
PutObjectRequest putObjectRequest =
new PutObjectRequest(bucketName, key, input, objectMetadata);
try {
PutObjectResult putObjectResult =
(PutObjectResult) callCOSClientWithRetry(putObjectRequest);
LOG.debug("Store empty file successfully. COS key: [{}], ETag: [{}].",
key, putObjectResult.getETag());
} catch (Exception e) {
String errMsg = String.format("Store empty file failed. "
+ "COS key: [%s], exception: [%s]", key, e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), key);
}
}
public PartETag uploadPart(File file, String key, String uploadId,
int partNum) throws IOException {
InputStream inputStream = new FileInputStream(file);
try {
return uploadPart(inputStream, key, uploadId, partNum, file.length());
} finally {
inputStream.close();
}
}
@Override
public PartETag uploadPart(InputStream inputStream, String key,
String uploadId, int partNum, long partSize) throws IOException {
UploadPartRequest uploadPartRequest = new UploadPartRequest();
uploadPartRequest.setBucketName(this.bucketName);
uploadPartRequest.setUploadId(uploadId);
uploadPartRequest.setInputStream(inputStream);
uploadPartRequest.setPartNumber(partNum);
uploadPartRequest.setPartSize(partSize);
uploadPartRequest.setKey(key);
try {
UploadPartResult uploadPartResult =
(UploadPartResult) callCOSClientWithRetry(uploadPartRequest);
return uploadPartResult.getPartETag();
} catch (Exception e) {
String errMsg = String.format("Current thread: [%d], COS key: [%s], "
+ "upload id: [%s], part num: [%d], exception: [%s]",
Thread.currentThread().getId(), key, uploadId, partNum, e.toString());
handleException(new Exception(errMsg), key);
}
return null;
}
public void abortMultipartUpload(String key, String uploadId) {
LOG.info("Abort the multipart upload. COS key: [{}], upload id: [{}].",
key, uploadId);
AbortMultipartUploadRequest abortMultipartUploadRequest =
new AbortMultipartUploadRequest(bucketName, key, uploadId);
cosClient.abortMultipartUpload(abortMultipartUploadRequest);
}
/**
* Initialize a multipart upload and return the upload id.
*
* @param key The COS object key initialized to multipart upload.
* @return The multipart upload id.
*/
public String getUploadId(String key) {
if (null == key || key.length() == 0) {
return "";
}
LOG.info("Initiate a multipart upload. bucket: [{}], COS key: [{}].",
bucketName, key);
InitiateMultipartUploadRequest initiateMultipartUploadRequest =
new InitiateMultipartUploadRequest(bucketName, key);
InitiateMultipartUploadResult initiateMultipartUploadResult =
cosClient.initiateMultipartUpload(initiateMultipartUploadRequest);
return initiateMultipartUploadResult.getUploadId();
}
/**
* Finish a multipart upload process, which will merge all parts uploaded.
*
* @param key The COS object key to be finished.
* @param uploadId The upload id of the multipart upload to be finished.
* @param partETagList The etag list of the part that has been uploaded.
* @return The result object of completing the multipart upload process.
*/
public CompleteMultipartUploadResult completeMultipartUpload(
String key, String uploadId, List<PartETag> partETagList) {
Collections.sort(partETagList, new Comparator<PartETag>() {
@Override
public int compare(PartETag o1, PartETag o2) {
return o1.getPartNumber() - o2.getPartNumber();
}
});
LOG.info("Complete the multipart upload. bucket: [{}], COS key: [{}], "
+ "upload id: [{}].", bucketName, key, uploadId);
CompleteMultipartUploadRequest completeMultipartUploadRequest =
new CompleteMultipartUploadRequest(
bucketName, key, uploadId, partETagList);
return cosClient.completeMultipartUpload(completeMultipartUploadRequest);
}
private FileMetadata queryObjectMetadata(String key) throws IOException {
GetObjectMetadataRequest getObjectMetadataRequest =
new GetObjectMetadataRequest(bucketName, key);
try {
ObjectMetadata objectMetadata =
(ObjectMetadata) callCOSClientWithRetry(getObjectMetadataRequest);
long mtime = 0;
if (objectMetadata.getLastModified() != null) {
mtime = objectMetadata.getLastModified().getTime();
}
long fileSize = objectMetadata.getContentLength();
FileMetadata fileMetadata = new FileMetadata(key, fileSize, mtime,
!key.endsWith(CosNFileSystem.PATH_DELIMITER));
LOG.debug("Retrieve file metadata. COS key: [{}], ETag: [{}], "
+ "length: [{}].", key, objectMetadata.getETag(),
objectMetadata.getContentLength());
return fileMetadata;
} catch (CosServiceException e) {
if (e.getStatusCode() != HttpStatus.SC_NOT_FOUND) {
String errorMsg = String.format("Retrieve file metadata file failed. "
+ "COS key: [%s], CosServiceException: [%s].", key, e.toString());
LOG.error(errorMsg);
handleException(new Exception(errorMsg), key);
}
}
return null;
}
@Override
public FileMetadata retrieveMetadata(String key) throws IOException {
if (key.endsWith(CosNFileSystem.PATH_DELIMITER)) {
key = key.substring(0, key.length() - 1);
}
if (!key.isEmpty()) {
FileMetadata fileMetadata = queryObjectMetadata(key);
if (fileMetadata != null) {
return fileMetadata;
}
}
// If the key is a directory.
key = key + CosNFileSystem.PATH_DELIMITER;
return queryObjectMetadata(key);
}
/**
* Download a COS object and return the input stream associated with it.
*
* @param key The object key that is being retrieved from the COS bucket
* @return This method returns null if the key is not found
* @throws IOException if failed to download.
*/
@Override
public InputStream retrieve(String key) throws IOException {
LOG.debug("Retrieve object key: [{}].", key);
GetObjectRequest getObjectRequest =
new GetObjectRequest(this.bucketName, key);
try {
COSObject cosObject =
(COSObject) callCOSClientWithRetry(getObjectRequest);
return cosObject.getObjectContent();
} catch (Exception e) {
String errMsg = String.format("Retrieving key: [%s] occurs "
+ "an exception: [%s].", key, e.toString());
LOG.error("Retrieving COS key: [{}] occurs an exception: [{}].", key, e);
handleException(new Exception(errMsg), key);
}
// never will get here
return null;
}
/**
* Retrieved a part of a COS object, which is specified the start position.
*
* @param key The object key that is being retrieved from
* the COS bucket.
* @param byteRangeStart The start position of the part to be retrieved in
* the object.
* @return The input stream associated with the retrieved object.
* @throws IOException if failed to retrieve.
*/
@Override
public InputStream retrieve(String key, long byteRangeStart)
throws IOException {
try {
LOG.debug("Retrieve COS key:[{}]. range start:[{}].",
key, byteRangeStart);
long fileSize = getFileLength(key);
long byteRangeEnd = fileSize - 1;
GetObjectRequest getObjectRequest =
new GetObjectRequest(this.bucketName, key);
if (byteRangeEnd >= byteRangeStart) {
getObjectRequest.setRange(byteRangeStart, fileSize - 1);
}
COSObject cosObject =
(COSObject) callCOSClientWithRetry(getObjectRequest);
return cosObject.getObjectContent();
} catch (Exception e) {
String errMsg =
String.format("Retrieving COS key: [%s] occurs an exception. " +
"byte range start: [%s], exception: [%s].",
key, byteRangeStart, e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), key);
}
// never will get here
return null;
}
/**
* Download a part of a COS object, which is specified the start and
* end position.
*
* @param key The object key that is being downloaded
* @param byteRangeStart The start position of the part to be retrieved in
* the object.
* @param byteRangeEnd The end position of the part to be retrieved in
* the object.
* @return The input stream associated with the retrieved objects.
* @throws IOException If failed to retrieve.
*/
@Override
public InputStream retrieveBlock(String key, long byteRangeStart,
long byteRangeEnd) throws IOException {
try {
GetObjectRequest request = new GetObjectRequest(this.bucketName, key);
request.setRange(byteRangeStart, byteRangeEnd);
COSObject cosObject = (COSObject) this.callCOSClientWithRetry(request);
return cosObject.getObjectContent();
} catch (CosServiceException e) {
String errMsg =
String.format("Retrieving key [%s] with byteRangeStart [%d] occurs " +
"an CosServiceException: [%s].",
key, byteRangeStart, e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), key);
return null;
} catch (CosClientException e) {
String errMsg =
String.format("Retrieving key [%s] with byteRangeStart [%d] "
+ "occurs an exception: [%s].",
key, byteRangeStart, e.toString());
LOG.error("Retrieving COS key: [{}] with byteRangeStart: [{}] " +
"occurs an exception: [{}].", key, byteRangeStart, e);
handleException(new Exception(errMsg), key);
}
return null;
}
@Override
public PartialListing list(String prefix, int maxListingLength)
throws IOException {
return list(prefix, maxListingLength, null, false);
}
@Override
public PartialListing list(String prefix, int maxListingLength,
String priorLastKey, boolean recurse) throws IOException {
return list(prefix, recurse ? null : CosNFileSystem.PATH_DELIMITER,
maxListingLength, priorLastKey);
}
/**
* List the metadata for all objects that
* the object key has the specified prefix.
*
* @param prefix The prefix to be listed.
* @param delimiter The delimiter is a sign, the same paths between
* are listed.
* @param maxListingLength The maximum number of listed entries.
* @param priorLastKey The last key in any previous search.
* @return A metadata list on the match.
* @throws IOException If list objects failed.
*/
private PartialListing list(String prefix, String delimiter,
int maxListingLength, String priorLastKey) throws IOException {
LOG.debug("List objects. prefix: [{}], delimiter: [{}], " +
"maxListLength: [{}], priorLastKey: [{}].",
prefix, delimiter, maxListingLength, priorLastKey);
if (!prefix.startsWith(CosNFileSystem.PATH_DELIMITER)) {
prefix += CosNFileSystem.PATH_DELIMITER;
}
ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
listObjectsRequest.setBucketName(bucketName);
listObjectsRequest.setPrefix(prefix);
listObjectsRequest.setDelimiter(delimiter);
listObjectsRequest.setMarker(priorLastKey);
listObjectsRequest.setMaxKeys(maxListingLength);
ObjectListing objectListing = null;
try {
objectListing =
(ObjectListing) callCOSClientWithRetry(listObjectsRequest);
} catch (Exception e) {
String errMsg = String.format("prefix: [%s], delimiter: [%s], "
+ "maxListingLength: [%d], priorLastKey: [%s]. "
+ "List objects occur an exception: [%s].", prefix,
(delimiter == null) ? "" : delimiter, maxListingLength, priorLastKey,
e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), prefix);
}
ArrayList<FileMetadata> fileMetadataArray = new ArrayList<>();
ArrayList<FileMetadata> commonPrefixArray = new ArrayList<>();
if (null == objectListing) {
String errMsg = String.format("List the prefix: [%s] failed. " +
"delimiter: [%s], max listing length:" +
" [%s], prior last key: [%s]",
prefix, delimiter, maxListingLength, priorLastKey);
handleException(new Exception(errMsg), prefix);
}
List<COSObjectSummary> summaries = objectListing.getObjectSummaries();
for (COSObjectSummary cosObjectSummary : summaries) {
String filePath = cosObjectSummary.getKey();
if (!filePath.startsWith(CosNFileSystem.PATH_DELIMITER)) {
filePath = CosNFileSystem.PATH_DELIMITER + filePath;
}
if (filePath.equals(prefix)) {
continue;
}
long mtime = 0;
if (cosObjectSummary.getLastModified() != null) {
mtime = cosObjectSummary.getLastModified().getTime();
}
long fileLen = cosObjectSummary.getSize();
fileMetadataArray.add(
new FileMetadata(filePath, fileLen, mtime, true));
}
List<String> commonPrefixes = objectListing.getCommonPrefixes();
for (String commonPrefix : commonPrefixes) {
if (!commonPrefix.startsWith(CosNFileSystem.PATH_DELIMITER)) {
commonPrefix = CosNFileSystem.PATH_DELIMITER + commonPrefix;
}
commonPrefixArray.add(
new FileMetadata(commonPrefix, 0, 0, false));
}
FileMetadata[] fileMetadata = new FileMetadata[fileMetadataArray.size()];
for (int i = 0; i < fileMetadataArray.size(); ++i) {
fileMetadata[i] = fileMetadataArray.get(i);
}
FileMetadata[] commonPrefixMetaData =
new FileMetadata[commonPrefixArray.size()];
for (int i = 0; i < commonPrefixArray.size(); ++i) {
commonPrefixMetaData[i] = commonPrefixArray.get(i);
}
// when truncated is false, it means that listing is finished.
if (!objectListing.isTruncated()) {
return new PartialListing(
null, fileMetadata, commonPrefixMetaData);
} else {
return new PartialListing(
objectListing.getNextMarker(), fileMetadata, commonPrefixMetaData);
}
}
@Override
public void delete(String key) throws IOException {
LOG.debug("Delete object key: [{}] from bucket: {}.", key, this.bucketName);
try {
DeleteObjectRequest deleteObjectRequest =
new DeleteObjectRequest(bucketName, key);
callCOSClientWithRetry(deleteObjectRequest);
} catch (Exception e) {
String errMsg =
String.format("Delete key: [%s] occurs an exception: [%s].",
key, e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), key);
}
}
public void rename(String srcKey, String dstKey) throws IOException {
LOG.debug("Rename source key: [{}] to dest key: [{}].", srcKey, dstKey);
try {
CopyObjectRequest copyObjectRequest =
new CopyObjectRequest(bucketName, srcKey, bucketName, dstKey);
callCOSClientWithRetry(copyObjectRequest);
DeleteObjectRequest deleteObjectRequest =
new DeleteObjectRequest(bucketName, srcKey);
callCOSClientWithRetry(deleteObjectRequest);
} catch (Exception e) {
String errMsg = String.format("Rename object unsuccessfully. "
+ "source cos key: [%s], dest COS " +
"key: [%s], exception: [%s]",
srcKey,
dstKey, e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), srcKey);
}
}
@Override
public void copy(String srcKey, String dstKey) throws IOException {
LOG.debug("Copy source key: [{}] to dest key: [{}].", srcKey, dstKey);
try {
CopyObjectRequest copyObjectRequest =
new CopyObjectRequest(bucketName, srcKey, bucketName, dstKey);
callCOSClientWithRetry(copyObjectRequest);
} catch (Exception e) {
String errMsg = String.format("Copy object unsuccessfully. "
+ "source COS key: %s, dest COS key: " +
"%s, exception: %s",
srcKey,
dstKey, e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), srcKey);
}
}
@Override
public void purge(String prefix) throws IOException {
throw new IOException("purge not supported");
}
@Override
public void dump() throws IOException {
throw new IOException("dump not supported");
}
// process Exception and print detail
private void handleException(Exception e, String key) throws IOException {
String cosPath = CosNFileSystem.SCHEME + "://" + bucketName + key;
String exceptInfo = String.format("%s : %s", cosPath, e.toString());
throw new IOException(exceptInfo);
}
@Override
public long getFileLength(String key) throws IOException {
LOG.debug("Get file length. COS key: {}", key);
GetObjectMetadataRequest getObjectMetadataRequest =
new GetObjectMetadataRequest(bucketName, key);
try {
ObjectMetadata objectMetadata =
(ObjectMetadata) callCOSClientWithRetry(getObjectMetadataRequest);
return objectMetadata.getContentLength();
} catch (Exception e) {
String errMsg = String.format("Getting file length occurs an exception." +
"COS key: %s, exception: %s", key,
e.toString());
LOG.error(errMsg);
handleException(new Exception(errMsg), key);
return 0; // never will get here
}
}
private <X> Object callCOSClientWithRetry(X request)
throws CosServiceException, IOException {
String sdkMethod = "";
int retryIndex = 1;
while (true) {
try {
if (request instanceof PutObjectRequest) {
sdkMethod = "putObject";
return this.cosClient.putObject((PutObjectRequest) request);
} else if (request instanceof UploadPartRequest) {
sdkMethod = "uploadPart";
if (((UploadPartRequest) request).getInputStream()
instanceof ByteBufferInputStream) {
((UploadPartRequest) request).getInputStream()
.mark((int) ((UploadPartRequest) request).getPartSize());
}
return this.cosClient.uploadPart((UploadPartRequest) request);
} else if (request instanceof GetObjectMetadataRequest) {
sdkMethod = "queryObjectMeta";
return this.cosClient.getObjectMetadata(
(GetObjectMetadataRequest) request);
} else if (request instanceof DeleteObjectRequest) {
sdkMethod = "deleteObject";
this.cosClient.deleteObject((DeleteObjectRequest) request);
return new Object();
} else if (request instanceof CopyObjectRequest) {
sdkMethod = "copyFile";
return this.cosClient.copyObject((CopyObjectRequest) request);
} else if (request instanceof GetObjectRequest) {
sdkMethod = "getObject";
return this.cosClient.getObject((GetObjectRequest) request);
} else if (request instanceof ListObjectsRequest) {
sdkMethod = "listObjects";
return this.cosClient.listObjects((ListObjectsRequest) request);
} else {
throw new IOException("no such method");
}
} catch (CosServiceException cse) {
String errMsg = String.format("Call cos sdk failed, "
+ "retryIndex: [%d / %d], "
+ "call method: %s, exception: %s",
retryIndex, this.maxRetryTimes, sdkMethod, cse.toString());
int statusCode = cse.getStatusCode();
// Retry all server errors
if (statusCode / 100 == 5) {
if (retryIndex <= this.maxRetryTimes) {
LOG.info(errMsg);
long sleepLeast = retryIndex * 300L;
long sleepBound = retryIndex * 500L;
try {
if (request instanceof UploadPartRequest) {
if (((UploadPartRequest) request).getInputStream()
instanceof ByteBufferInputStream) {
((UploadPartRequest) request).getInputStream().reset();
}
}
Thread.sleep(
ThreadLocalRandom.current().nextLong(sleepLeast, sleepBound));
++retryIndex;
} catch (InterruptedException e) {
throw new IOException(e.toString());
}
} else {
LOG.error(errMsg);
throw new IOException(errMsg);
}
} else {
throw cse;
}
} catch (Exception e) {
String errMsg = String.format("Call cos sdk failed, "
+ "call method: %s, exception: %s", sdkMethod, e.toString());
LOG.error(errMsg);
throw new IOException(errMsg);
}
}
}
@Override
public void close() {
if (null != this.cosClient) {
this.cosClient.shutdown();
}
}
}