diff --git a/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java b/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java index 30da259fd51f8a..668ec45db5d2a7 100644 --- a/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java +++ b/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java @@ -55,6 +55,10 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.PathFilter; import org.apache.hadoop.fs.RemoteIterator; +import org.apache.hadoop.io.retry.RetryPolicies; +import org.apache.hadoop.io.retry.RetryPolicy; +import org.apache.hadoop.io.retry.RetryPolicy.RetryAction; +import org.apache.hadoop.util.Preconditions; import org.apache.hadoop.util.VersionInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -72,6 +76,7 @@ import java.util.List; import java.util.ListIterator; import java.util.NoSuchElementException; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import com.aliyun.oss.common.comm.SignVersion; @@ -85,6 +90,8 @@ public class AliyunOSSFileSystemStore { public static final Logger LOG = LoggerFactory.getLogger(AliyunOSSFileSystemStore.class); + private static final int MAX_MULTIPART_COMPLETE_RETRY_LIMIT = 10; + private static final long MAX_MULTIPART_COMPLETE_RETRY_DELAY = 10_000; private String username; private FileSystem.Statistics statistics; private OSSClient ossClient; @@ -93,6 +100,7 @@ public class AliyunOSSFileSystemStore { private int maxKeys; private String serverSideEncryptionAlgorithm; private boolean useListV1; + private RetryPolicy multipartCompleteRetryPolicy; public void initialize(URI uri, Configuration conf, String user, FileSystem.Statistics stat) throws IOException { @@ -106,6 +114,21 @@ public void initialize(URI uri, Configuration conf, String user, clientConf.setProtocol(secureConnections ? Protocol.HTTPS : Protocol.HTTP); clientConf.setMaxErrorRetry(conf.getInt(MAX_ERROR_RETRIES_KEY, MAX_ERROR_RETRIES_DEFAULT)); + int retryLimit = conf.getInt(MULTIPART_COMPLETE_RETRY_LIMIT_KEY, + MULTIPART_COMPLETE_RETRY_LIMIT_DEFAULT); + Preconditions.checkArgument(retryLimit >= 0 + && retryLimit <= MAX_MULTIPART_COMPLETE_RETRY_LIMIT, + "%s must be between 0 and %s", MULTIPART_COMPLETE_RETRY_LIMIT_KEY, + MAX_MULTIPART_COMPLETE_RETRY_LIMIT); + long retryInterval = conf.getTimeDuration( + MULTIPART_COMPLETE_RETRY_INTERVAL_KEY, + MULTIPART_COMPLETE_RETRY_INTERVAL_DEFAULT, TimeUnit.MILLISECONDS); + Preconditions.checkArgument(retryInterval > 0 + && retryInterval <= (Long.MAX_VALUE >> retryLimit), + "%s must be positive and not overflow after %s retries", + MULTIPART_COMPLETE_RETRY_INTERVAL_KEY, retryLimit); + multipartCompleteRetryPolicy = RetryPolicies.exponentialBackoffRetry( + retryLimit, retryInterval, TimeUnit.MILLISECONDS); clientConf.setConnectionTimeout(conf.getInt(ESTABLISH_TIMEOUT_KEY, ESTABLISH_TIMEOUT_DEFAULT)); clientConf.setSocketTimeout(conf.getInt(SOCKET_TIMEOUT_KEY, @@ -399,11 +422,8 @@ private boolean multipartCopy(String srcKey, long contentLength, statistics.incrementBytesWritten(size); partETags.add(partCopyResult.getPartETag()); } - CompleteMultipartUploadRequest completeMultipartUploadRequest = - new CompleteMultipartUploadRequest(bucketName, dstKey, - uploadId, partETags); CompleteMultipartUploadResult completeMultipartUploadResult = - ossClient.completeMultipartUpload(completeMultipartUploadRequest); + completeMultipartUpload(dstKey, uploadId, partETags); LOG.debug(completeMultipartUploadResult.getETag()); return true; } catch (OSSException | ClientException e) { @@ -763,7 +783,47 @@ public CompleteMultipartUploadResult completeMultipartUpload(String key, CompleteMultipartUploadRequest completeMultipartUploadRequest = new CompleteMultipartUploadRequest(bucketName, key, uploadId, partETags); - return ossClient.completeMultipartUpload(completeMultipartUploadRequest); + for (int retries = 0; ; retries++) { + try { + return submitCompleteMultipartUpload(completeMultipartUploadRequest); + } catch (OSSException e) { + // A response with this error code means OSS rejected the request. + // Do not retry timeouts or other failures: Complete is not idempotent. + if (!"QpsLimitExceeded".equals(e.getErrorCode())) { + throw e; + } + RetryAction action; + try { + action = multipartCompleteRetryPolicy.shouldRetry( + e, retries, 0, false); + } catch (Exception policyFailure) { + e.addSuppressed(policyFailure); + throw e; + } + if (action.action != RetryAction.RetryDecision.RETRY) { + throw e; + } + LOG.warn("OSS throttled multipart completion of {} (retry {})", + key, retries + 1); + try { + sleepBeforeCompleteRetry(Math.min(action.delayMillis, + MAX_MULTIPART_COMPLETE_RETRY_DELAY)); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new ClientException("Interrupted while retrying OSS multipart " + + "completion of " + key, interrupted); + } + } + } + } + + CompleteMultipartUploadResult submitCompleteMultipartUpload( + CompleteMultipartUploadRequest request) { + return ossClient.completeMultipartUpload(request); + } + + void sleepBeforeCompleteRetry(long delayMillis) throws InterruptedException { + Thread.sleep(delayMillis); } /** diff --git a/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/Constants.java b/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/Constants.java index 176669ed1527a8..ae5d8e0f5fcf3e 100644 --- a/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/Constants.java +++ b/hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/Constants.java @@ -70,6 +70,14 @@ private Constants() { public static final String MAX_ERROR_RETRIES_KEY = "fs.oss.attempts.maximum"; public static final int MAX_ERROR_RETRIES_DEFAULT = 10; + // Retry only a CompleteMultipartUpload rejected by OSS throttling. + public static final String MULTIPART_COMPLETE_RETRY_LIMIT_KEY = + "fs.oss.multipart.complete.retry.limit"; + public static final int MULTIPART_COMPLETE_RETRY_LIMIT_DEFAULT = 3; + public static final String MULTIPART_COMPLETE_RETRY_INTERVAL_KEY = + "fs.oss.multipart.complete.retry.interval"; + public static final long MULTIPART_COMPLETE_RETRY_INTERVAL_DEFAULT = 500; + // Time until we give up trying to establish a connection to oss public static final String ESTABLISH_TIMEOUT_KEY = "fs.oss.connection.establish.timeout"; diff --git a/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md b/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md index 578b9595d1d7c2..0c6bbddf8d4057 100644 --- a/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md +++ b/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md @@ -168,6 +168,27 @@ please raise your issues with them. How many times we should retry commands on transient errors. + + fs.oss.multipart.complete.retry.limit + 3 + Extra attempts after OSS explicitly rejects a multipart completion + with QpsLimitExceeded, for both uploads and copies. Must be between 0 and 10; + 0 disables these retries. The initial request is not counted, so the default + allows at most 4 completion requests. This does not change SDK retries + configured by fs.oss.attempts.maximum for other operations. + + + + fs.oss.multipart.complete.retry.interval + 500ms + Base interval for exponential backoff with jitter when + retrying a throttled multipart completion. Must be at least 1ms. + Each retry delay, including jitter, is capped at 10s. + With the default settings, the three retry delays total at most 10.5s, + excluding request execution time. + Other completion failures are not retried. + + fs.oss.connection.establish.timeout 50000 diff --git a/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSFileSystemStore.java b/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSFileSystemStore.java index cc82f8ec40cabc..df283a9e2b2fe4 100644 --- a/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSFileSystemStore.java +++ b/hadoop-tools/hadoop-aliyun/src/test/java/org/apache/hadoop/fs/aliyun/oss/TestAliyunOSSFileSystemStore.java @@ -18,15 +18,27 @@ package org.apache.hadoop.fs.aliyun.oss; +import com.aliyun.oss.ClientException; +import com.aliyun.oss.OSSClient; +import com.aliyun.oss.OSSException; +import com.aliyun.oss.model.AbortMultipartUploadRequest; +import com.aliyun.oss.model.CompleteMultipartUploadRequest; +import com.aliyun.oss.model.CompleteMultipartUploadResult; +import com.aliyun.oss.model.CopyObjectResult; +import com.aliyun.oss.model.InitiateMultipartUploadRequest; +import com.aliyun.oss.model.InitiateMultipartUploadResult; import com.aliyun.oss.model.OSSObjectSummary; import com.aliyun.oss.model.ObjectMetadata; +import com.aliyun.oss.model.PartETag; +import com.aliyun.oss.model.UploadPartCopyRequest; +import com.aliyun.oss.model.UploadPartCopyResult; +import com.aliyun.oss.model.VoidResult; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.contract.ContractTestUtils; import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import java.io.BufferedInputStream; @@ -34,17 +46,21 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.lang.reflect.Field; import java.net.URI; import java.security.DigestInputStream; import java.security.DigestOutputStream; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; -import static org.apache.hadoop.fs.aliyun.oss.Constants.MAX_PAGING_KEYS_DEFAULT; +import static org.apache.hadoop.fs.aliyun.oss.Constants.*; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assumptions.assumeTrue; @@ -53,13 +69,14 @@ * Aliyun OSS. */ public class TestAliyunOSSFileSystemStore { - private Configuration conf; private AliyunOSSFileSystemStore store; private AliyunOSSFileSystem fs; - @BeforeEach - public void setUp() throws Exception { - conf = new Configuration(); + private void initializeFileSystem() throws IOException { + Configuration conf = new Configuration(); + assumeTrue(conf.get(Constants.ACCESS_KEY_ID) != null); + assumeTrue(conf.get(Constants.ACCESS_KEY_SECRET) != null); + assumeTrue(conf.get("test.fs.oss.name") != null); fs = new AliyunOSSFileSystem(); fs.initialize(URI.create(conf.get("test.fs.oss.name")), conf); store = fs.getStore(); @@ -67,6 +84,9 @@ public void setUp() throws Exception { @AfterEach public void tearDown() throws Exception { + if (store == null) { + return; + } try { store.purge("test"); } catch (Exception e) { @@ -75,14 +95,6 @@ public void tearDown() throws Exception { } } - @BeforeAll - public static void checkSettings() throws Exception { - Configuration conf = new Configuration(); - assumeTrue(conf.get(Constants.ACCESS_KEY_ID) != null); - assumeTrue(conf.get(Constants.ACCESS_KEY_SECRET) != null); - assumeTrue(conf.get("test.fs.oss.name") != null); - } - protected void writeRenameReadCompare(Path path, long len) throws IOException, NoSuchAlgorithmException { MessageDigest digest = MessageDigest.getInstance("MD5"); @@ -123,6 +135,7 @@ protected void writeRenameReadCompare(Path path, long len) @Test public void testSmallUpload() throws IOException, NoSuchAlgorithmException { + initializeFileSystem(); // Regular upload, regular copy writeRenameReadCompare(new Path("/test/small"), 16384); } @@ -130,6 +143,7 @@ public void testSmallUpload() throws IOException, NoSuchAlgorithmException { @Test public void testLargeUpload() throws IOException, NoSuchAlgorithmException { + initializeFileSystem(); // Multipart upload, shallow copy writeRenameReadCompare(new Path("/test/xlarge"), Constants.MULTIPART_UPLOAD_PART_SIZE_DEFAULT + 1); @@ -137,6 +151,7 @@ public void testLargeUpload() @Test public void testDeleteObjects() throws IOException, NoSuchAlgorithmException { + initializeFileSystem(); // generate test files final int files = 10; final long size = 5 * 1024 * 1024; @@ -159,4 +174,236 @@ public void testDeleteObjects() throws IOException, NoSuchAlgorithmException { objects = store.listObjects(listRequest); assertEquals(0, objects.getObjectSummaries().size()); } + + private static class RecordingStore extends AliyunOSSFileSystemStore + implements AutoCloseable { + private final List responses = new ArrayList<>(); + private final List requests = + new ArrayList<>(); + private final List delays = new ArrayList<>(); + private int nextResponse; + private boolean interrupt; + + @Override + CompleteMultipartUploadResult submitCompleteMultipartUpload( + CompleteMultipartUploadRequest request) { + requests.add(request); + Object response = responses.get(nextResponse++); + if (response instanceof RuntimeException) { + throw (RuntimeException) response; + } + return (CompleteMultipartUploadResult) response; + } + + @Override + void sleepBeforeCompleteRetry(long delayMillis) + throws InterruptedException { + delays.add(delayMillis); + if (interrupt) { + throw new InterruptedException("test interruption"); + } + } + } + + private static class CopyClient extends OSSClient { + private int copiedParts; + private int aborts; + + CopyClient() { + super("http://localhost", "test-id", "test-secret"); + } + + @Override + public CopyObjectResult copyObject(String srcBucket, String srcKey, + String dstBucket, String dstKey) { + throw new ClientException("Use multipart copy"); + } + + @Override + public InitiateMultipartUploadResult initiateMultipartUpload( + InitiateMultipartUploadRequest request) { + InitiateMultipartUploadResult result = + new InitiateMultipartUploadResult(); + result.setUploadId("copy-upload-id"); + return result; + } + + @Override + public UploadPartCopyResult uploadPartCopy(UploadPartCopyRequest request) { + copiedParts++; + UploadPartCopyResult result = new UploadPartCopyResult(); + result.setPartNumber(request.getPartNumber()); + result.setETag("copy-etag"); + return result; + } + + @Override + public CompleteMultipartUploadResult completeMultipartUpload( + CompleteMultipartUploadRequest request) { + throw new AssertionError("Multipart copy bypassed completion retries"); + } + + @Override + public VoidResult abortMultipartUpload(AbortMultipartUploadRequest request) { + aborts++; + return new VoidResult(); + } + } + + private static RecordingStore configuredStore(Configuration conf) + throws IOException { + RecordingStore retryStore = new RecordingStore(); + conf.set(ENDPOINT_KEY, "http://localhost"); + conf.set(ACCESS_KEY_ID, "test-id"); + conf.set(ACCESS_KEY_SECRET, "test-secret"); + retryStore.initialize(URI.create("oss://bucket"), conf, "test-user", + new FileSystem.Statistics("oss")); + return retryStore; + } + + private static RecordingStore configuredStore(int retryLimit, + String retryInterval) throws IOException { + Configuration conf = new Configuration(false); + conf.setInt(MULTIPART_COMPLETE_RETRY_LIMIT_KEY, retryLimit); + conf.set(MULTIPART_COMPLETE_RETRY_INTERVAL_KEY, retryInterval); + return configuredStore(conf); + } + + private static OSSException ossError(String code) { + return new OSSException("OSS rejected completion", code, + "request", "host", null, null, null); + } + + private static CompleteMultipartUploadResult complete(RecordingStore retryStore) { + List parts = new ArrayList<>(Arrays.asList( + new PartETag(2, "second"), new PartETag(1, "first"))); + return retryStore.completeMultipartUpload("key", "upload-id", parts); + } + + @Test + public void testCompleteMultipartUploadRetries() throws IOException { + for (int retryLimit : new int[] {0, 1, 3, 10}) { + for (boolean succeeds : new boolean[] {true, false}) { + // Leave the default configuration unset to verify initialize() uses it. + try (RecordingStore retryStore = retryLimit == 3 + ? configuredStore(new Configuration(false)) + : configuredStore(retryLimit, retryLimit == 1 ? "1ms" : "30s")) { + OSSException throttled = ossError("QpsLimitExceeded"); + CompleteMultipartUploadResult result = + new CompleteMultipartUploadResult(); + for (int i = 0; i < retryLimit; i++) { + retryStore.responses.add(throttled); + } + retryStore.responses.add(succeeds ? result : throttled); + if (succeeds) { + assertSame(result, complete(retryStore)); + } else { + assertSame(throttled, assertThrows(OSSException.class, + () -> complete(retryStore))); + } + assertEquals(retryLimit + 1, retryStore.requests.size()); + assertEquals(retryLimit, retryStore.delays.size()); + CompleteMultipartUploadRequest request = retryStore.requests.get(0); + retryStore.requests.forEach(attempt -> assertSame(request, attempt)); + assertEquals("upload-id", request.getUploadId()); + assertEquals(1, request.getPartETags().get(0).getPartNumber()); + assertEquals("first", request.getPartETags().get(0).getETag()); + assertEquals(2, request.getPartETags().get(1).getPartNumber()); + assertEquals("second", request.getPartETags().get(1).getETag()); + if (retryLimit == 3) { + long[] minimum = {500, 1000, 2000}; + long[] maximum = {1500, 3000, 6000}; + for (int i = 0; i < retryLimit; i++) { + assertTrue(retryStore.delays.get(i) >= minimum[i] + && retryStore.delays.get(i) < maximum[i]); + } + } else if (retryLimit == 1) { + assertTrue(retryStore.delays.get(0) >= 1 + && retryStore.delays.get(0) < 3); + } else if (retryLimit == 10) { + // Cap the actual delay, including jitter, even with a larger base. + retryStore.delays.forEach(delay -> + assertEquals(10_000L, delay.longValue())); + } + } + } + } + } + + @Test + public void testCompleteMultipartUploadStopsOnFailure() throws IOException { + for (RuntimeException failure : Arrays.asList(ossError("InvalidPart"), + ossError("NoSuchUpload"), ossError("AccessDenied"), + new ClientException("response lost"))) { + for (int throttles : new int[] {0, 1}) { + try (RecordingStore retryStore = configuredStore(new Configuration(false))) { + if (throttles > 0) { + retryStore.responses.add(ossError("QpsLimitExceeded")); + } + retryStore.responses.add(failure); + assertSame(failure, assertThrows(failure.getClass(), + () -> complete(retryStore))); + assertEquals(throttles + 1, retryStore.requests.size()); + assertEquals(throttles, retryStore.delays.size()); + } + } + } + try (RecordingStore retryStore = configuredStore(new Configuration(false))) { + retryStore.responses.add(ossError("QpsLimitExceeded")); + retryStore.interrupt = true; + ClientException failure = assertThrows(ClientException.class, + () -> complete(retryStore)); + assertTrue(failure.getCause() instanceof InterruptedException); + assertTrue(Thread.currentThread().isInterrupted()); + assertEquals(1, retryStore.requests.size()); + } finally { + Thread.interrupted(); + } + } + + @Test + public void testInvalidCompletionRetryConfiguration() { + for (int retryLimit : new int[] {-1, 11}) { + IllegalArgumentException failure = assertThrows( + IllegalArgumentException.class, + () -> configuredStore(retryLimit, "500ms")); + assertTrue(failure.getMessage().contains(MULTIPART_COMPLETE_RETRY_LIMIT_KEY)); + } + for (String interval : Arrays.asList("0ms", "-1ms", + Long.MAX_VALUE + "ms")) { + IllegalArgumentException failure = assertThrows( + IllegalArgumentException.class, () -> configuredStore(3, interval)); + assertTrue(failure.getMessage().contains(MULTIPART_COMPLETE_RETRY_INTERVAL_KEY)); + } + } + + @Test + public void testMultipartCopyUsesCompletionRetries() throws Exception { + for (boolean succeeds : new boolean[] {true, false}) { + try (RecordingStore retryStore = configuredStore(1, "500ms")) { + CopyClient client = new CopyClient(); + retryStore.close(); + Field clientField = AliyunOSSFileSystemStore.class + .getDeclaredField("ossClient"); + clientField.setAccessible(true); + clientField.set(retryStore, client); + retryStore.responses.add(ossError("QpsLimitExceeded")); + retryStore.responses.add(succeeds ? new CompleteMultipartUploadResult() + : ossError("QpsLimitExceeded")); + + assertEquals(succeeds, retryStore.copyFile("src", + MULTIPART_UPLOAD_PART_SIZE_DEFAULT + 1, "dst")); + assertEquals(2, client.copiedParts); + assertEquals(succeeds ? 0 : 1, client.aborts); + assertEquals(2, retryStore.requests.size()); + assertEquals(1, retryStore.delays.size()); + assertSame(retryStore.requests.get(0), retryStore.requests.get(1)); + assertEquals("dst", retryStore.requests.get(0).getKey()); + assertEquals("copy-upload-id", retryStore.requests.get(0).getUploadId()); + assertEquals(2, retryStore.requests.get(0).getPartETags().size()); + assertEquals("copy-etag", + retryStore.requests.get(0).getPartETags().get(0).getETag()); + } + } + } }