diff --git a/docs/src/main/sphinx/object-storage/file-system-azure.md b/docs/src/main/sphinx/object-storage/file-system-azure.md index 9a43474e6b6e..9f4de22b4197 100644 --- a/docs/src/main/sphinx/object-storage/file-system-azure.md +++ b/docs/src/main/sphinx/object-storage/file-system-azure.md @@ -59,6 +59,9 @@ system support: for all requests sent to Azure Storage. Defaults to `Trino`. * - `azure.multipart-write-enabled` - Enable multipart writes for large files. Defaults to `false`. +* - `azure.max-error-retries` + - Maximum [integer](prop-type-integer) number of retries for transient Azure + HTTP request failures using exponential backoff. Defaults to `4`. ::: (azure-user-assigned-managed-identity-authentication)= diff --git a/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystem.java b/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystem.java index a17a3452bfa1..0ad15e2df6a8 100644 --- a/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystem.java +++ b/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystem.java @@ -27,6 +27,8 @@ import com.azure.storage.blob.sas.BlobSasPermission; import com.azure.storage.blob.sas.BlobServiceSasSignatureValues; import com.azure.storage.blob.specialized.BlockBlobClient; +import com.azure.storage.common.policy.RequestRetryOptions; +import com.azure.storage.common.policy.RetryPolicyType; import com.azure.storage.common.sas.SasProtocol; import com.azure.storage.file.datalake.DataLakeDirectoryClient; import com.azure.storage.file.datalake.DataLakeFileClient; @@ -94,6 +96,7 @@ public class AzureFileSystem private final int maxWriteConcurrency; private final long maxSingleUploadSizeBytes; private final boolean multipartWriteEnabled; + private final int maxErrorRetries; public AzureFileSystem( HttpClient httpClient, @@ -106,7 +109,8 @@ public AzureFileSystem( DataSize writeBlockSize, int maxWriteConcurrency, DataSize maxSingleUploadSize, - boolean multipartWriteEnabled) + boolean multipartWriteEnabled, + int maxErrorRetries) { this.httpClient = requireNonNull(httpClient, "httpClient is null"); this.concurrencyPolicy = requireNonNull(concurrencyPolicy, "concurrencyPolicy is null"); @@ -120,6 +124,7 @@ public AzureFileSystem( this.maxWriteConcurrency = maxWriteConcurrency; this.maxSingleUploadSizeBytes = maxSingleUploadSize.toBytes(); this.multipartWriteEnabled = multipartWriteEnabled; + this.maxErrorRetries = maxErrorRetries; } @Override @@ -684,6 +689,7 @@ private BlobContainerClient createBlobContainerClient(AzureLocation location, Op BlobContainerClientBuilder builder = new BlobContainerClientBuilder() .httpClient(httpClient) .addPolicy(concurrencyPolicy) + .retryOptions(new RequestRetryOptions(RetryPolicyType.EXPONENTIAL, maxErrorRetries, (Integer) null, null, null, null)) .clientOptions(new ClientOptions().setTracingOptions(tracingOptions)) .endpoint("https://%s.blob.%s".formatted(location.account(), validatedEndpoint(location))); @@ -701,6 +707,7 @@ private DataLakeFileSystemClient createFileSystemClient(AzureLocation location, DataLakeServiceClientBuilder builder = new DataLakeServiceClientBuilder() .httpClient(httpClient) .addPolicy(concurrencyPolicy) + .retryOptions(new RequestRetryOptions(RetryPolicyType.EXPONENTIAL, maxErrorRetries, (Integer) null, null, null, null)) .clientOptions(new ClientOptions().setTracingOptions(tracingOptions)) .endpoint("https://%s.dfs.%s".formatted(location.account(), validatedEndpoint(location))); key.ifPresent(encryption -> builder.customerProvidedKey(lakeCustomerProvidedKey(encryption))); diff --git a/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystemConfig.java b/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystemConfig.java index 8d3b83737950..155739a24802 100644 --- a/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystemConfig.java +++ b/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystemConfig.java @@ -52,6 +52,7 @@ public enum AuthType private Duration httpRequestTimeout = new Duration(10, TimeUnit.MINUTES); private String applicationId = "Trino"; private boolean multipartWriteEnabled; + private int maxErrorRetries = 4; @NotNull public AuthType getAuthType() @@ -214,4 +215,18 @@ public AzureFileSystemConfig setMultipartWriteEnabled(boolean multipartWriteEnab this.multipartWriteEnabled = multipartWriteEnabled; return this; } + + @Min(0) + public int getMaxErrorRetries() + { + return maxErrorRetries; + } + + @Config("azure.max-error-retries") + @ConfigDescription("Maximum number of retries for transient Azure HTTP request failures") + public AzureFileSystemConfig setMaxErrorRetries(int maxErrorRetries) + { + this.maxErrorRetries = maxErrorRetries; + return this; + } } diff --git a/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystemFactory.java b/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystemFactory.java index ece7cfd1c08e..c9fd677d40f3 100644 --- a/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystemFactory.java +++ b/lib/trino-filesystem-azure/src/main/java/io/trino/filesystem/azure/AzureFileSystemFactory.java @@ -61,6 +61,7 @@ public class AzureFileSystemFactory private final EventLoopGroup eventLoopGroup; private final boolean multipart; private final HttpPipelinePolicy concurrencyPolicy; + private final int maxErrorRetries; @Inject public AzureFileSystemFactory(OpenTelemetry openTelemetry, AzureAuth azureAuth, AzureFileSystemConfig config) @@ -77,7 +78,8 @@ public AzureFileSystemFactory(OpenTelemetry openTelemetry, AzureAuth azureAuth, config.getConnectionPoolMaxIdleTime(), config.getHttpRequestTimeout(), config.getApplicationId(), - config.isMultipartWriteEnabled()); + config.isMultipartWriteEnabled(), + config.getMaxErrorRetries()); } public AzureFileSystemFactory( @@ -93,7 +95,8 @@ public AzureFileSystemFactory( Duration connectionPoolMaxIdleTime, Duration httpRequestTimeout, String applicationId, - boolean multipart) + boolean multipart, + int maxErrorRetries) { this.auth = requireNonNull(azureAuth, "azureAuth is null"); this.endpoint = requireNonNull(endpoint, "endpoint is null"); @@ -116,6 +119,7 @@ public AzureFileSystemFactory( clientOptions.setMaximumConnectionPoolSize(maxHttpConnections); httpClient = createAzureHttpClient(connectionProvider, eventLoopGroup, clientOptions); this.multipart = multipart; + this.maxErrorRetries = maxErrorRetries; this.concurrencyPolicy = new ConcurrencyLimitHttpPipelinePolicy( maxHttpRequests, httpRequestTimeout.toJavaTime()); @@ -147,7 +151,7 @@ public void destroy() public TrinoFileSystem create(ConnectorIdentity identity) { AzureAuth effectiveAuth = getEffectiveAuth(identity); - return new AzureFileSystem(httpClient, concurrencyPolicy, uploadExecutor, tracingOptions, effectiveAuth, endpoint, readBlockSize, writeBlockSize, maxWriteConcurrency, maxSingleUploadSize, multipart); + return new AzureFileSystem(httpClient, concurrencyPolicy, uploadExecutor, tracingOptions, effectiveAuth, endpoint, readBlockSize, writeBlockSize, maxWriteConcurrency, maxSingleUploadSize, multipart, maxErrorRetries); } private AzureAuth getEffectiveAuth(ConnectorIdentity identity) diff --git a/lib/trino-filesystem-azure/src/test/java/io/trino/filesystem/azure/TestAzureFileSystemConfig.java b/lib/trino-filesystem-azure/src/test/java/io/trino/filesystem/azure/TestAzureFileSystemConfig.java index a1751a32248c..6c22d8a07ed7 100644 --- a/lib/trino-filesystem-azure/src/test/java/io/trino/filesystem/azure/TestAzureFileSystemConfig.java +++ b/lib/trino-filesystem-azure/src/test/java/io/trino/filesystem/azure/TestAzureFileSystemConfig.java @@ -45,7 +45,8 @@ void testDefaults() .setConnectionPoolMaxIdleTime(new Duration(5, MINUTES)) .setHttpRequestTimeout(new Duration(10, MINUTES)) .setApplicationId("Trino") - .setMultipartWriteEnabled(false)); + .setMultipartWriteEnabled(false) + .setMaxErrorRetries(4)); } @Test @@ -64,6 +65,7 @@ public void testExplicitPropertyMappings() .put("azure.http-request-timeout", "1m") .put("azure.application-id", "application id") .put("azure.multipart-write-enabled", "true") + .put("azure.max-error-retries", "10") .buildOrThrow(); AzureFileSystemConfig expected = new AzureFileSystemConfig() @@ -78,7 +80,8 @@ public void testExplicitPropertyMappings() .setConnectionPoolMaxIdleTime(new Duration(1, MINUTES)) .setHttpRequestTimeout(new Duration(1, MINUTES)) .setApplicationId("application id") - .setMultipartWriteEnabled(true); + .setMultipartWriteEnabled(true) + .setMaxErrorRetries(10); assertFullMapping(properties, expected); }